from __future__ import annotations import json import subprocess import tempfile import unittest from pathlib import Path from unittest.mock import Mock from work_item.core import Config, GitRepo, Runner, Story, WorkItemError, Workflow, find_config def run(*args: str, cwd: Path) -> str: result = subprocess.run(args, cwd=cwd, text=True, capture_output=True, check=True) return result.stdout.strip() def config_for(slug: str, **changes) -> Config: values = dict( repo_slug=slug, login="test", assignee="agent", remote="origin", main_branch="main", branch_prefix="us", require_ci=False, max_file_bytes=64, required_tests=(), denied_patterns=("_IGNORE/**", "**/_IGNORE/**", "pictures/**", "*.env", "*.jpg"), allowed_patterns=("tests/fixtures/**",), ) values.update(changes) return Config(**values) class RepositoryFixture(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.base = Path(self.temp.name) remote_parent = self.base / "domverse" remote_parent.mkdir() self.seed = self.base / "seed" self.seed.mkdir() run("git", "init", "-b", "main", cwd=self.seed) run("git", "config", "user.name", "Test Agent", cwd=self.seed) run("git", "config", "user.email", "agent@example.test", cwd=self.seed) (self.seed / "README.md").write_text("seed\n", encoding="utf-8") run("git", "add", "README.md", cwd=self.seed) run("git", "commit", "-m", "seed", cwd=self.seed) self.remote = remote_parent / "photoanalyzer.git" run("git", "clone", "--bare", str(self.seed), str(self.remote), cwd=self.base) self.repo = self.base / "work" run("git", "clone", str(self.remote), str(self.repo), cwd=self.base) run("git", "config", "user.name", "Test Agent", cwd=self.repo) run("git", "config", "user.email", "agent@example.test", cwd=self.repo) self.config = config_for("domverse/photoanalyzer") def tearDown(self) -> None: self.temp.cleanup() class StoryTests(unittest.TestCase): def test_parse_and_branch_slug(self) -> None: story = Story.from_issue( { "number": 12, "title": "US02-05 — Build the Static Application Shell", "labels": [{"name": "type/feature"}], "assignees": [{"login": "agent"}], } ) self.assertIsNotNone(story) assert story is not None self.assertEqual((story.epic, story.sequence), (2, 5)) self.assertEqual(story.branch_slug, "build-the-static-application-shell") self.assertEqual(story.assignees, ("agent",)) def test_non_story_issue_is_ignored(self) -> None: self.assertIsNone(Story.from_issue({"number": 1, "title": "Maintenance"})) def test_branch_slug_is_portable_and_bounded(self) -> None: story = Story.from_issue( { "number": 12, "title": "US02-05 — Über-long title! " + "word " * 30, } ) assert story is not None self.assertLessEqual(len(story.branch_slug), 48) self.assertRegex(story.branch_slug, r"^[a-z0-9-]+$") self.assertFalse(story.branch_slug.endswith("-")) class ConfigTests(unittest.TestCase): def test_loads_defaults_and_overrides(self) -> None: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "workflow.toml" path.write_text( """ [repository] slug = "owner/repo" login = "local" assignee = "agent" [workflow] require_ci = true required_tests = ["test one", "test two"] [safety] max_file_bytes = 123 deny = ["*.env"] allow = ["tests/**"] """.strip(), encoding="utf-8", ) config = Config.load(path) self.assertEqual(config.repo_slug, "owner/repo") self.assertEqual(config.remote, "origin") self.assertEqual(config.main_branch, "main") self.assertEqual(config.branch_prefix, "us") self.assertTrue(config.require_ci) self.assertEqual(config.required_tests, ("test one", "test two")) self.assertEqual(config.max_file_bytes, 123) def test_loads_yaml_configuration(self) -> None: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / ".work-item.yml" path.write_text( """ repository: slug: owner/yaml-repo login: yaml-login assignee: yaml-agent workflow: branch_prefix: story require_ci: true required_tests: - pytest -q safety: max_file_bytes: 2048 deny: - "*.env" allow: - tests/fixtures/** """.strip(), encoding="utf-8", ) config = Config.load(path, environment={}) self.assertEqual(config.repo_slug, "owner/yaml-repo") self.assertEqual(config.login, "yaml-login") self.assertEqual(config.branch_prefix, "story") self.assertTrue(config.require_ci) self.assertEqual(config.required_tests, ("pytest -q",)) self.assertEqual(config.denied_patterns, ("*.env",)) def test_loads_dotenv_configuration_and_json_lists(self) -> None: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / ".work-item.env" path.write_text( """ # Repository identity export WORK_ITEM_REPO_SLUG=owner/env-repo WORK_ITEM_LOGIN='env-login' WORK_ITEM_ASSIGNEE=env-agent # local account WORK_ITEM_REQUIRE_CI=yes WORK_ITEM_REQUIRED_TESTS='["pytest -q", "ruff check ."]' WORK_ITEM_MAX_FILE_BYTES=4096 WORK_ITEM_DENY='["*.env", "data/**"]' WORK_ITEM_ALLOW='["tests/fixtures/**"]' """.strip(), encoding="utf-8", ) config = Config.load(path, environment={}) self.assertEqual(config.repo_slug, "owner/env-repo") self.assertEqual(config.login, "env-login") self.assertTrue(config.require_ci) self.assertEqual(config.required_tests, ("pytest -q", "ruff check .")) self.assertEqual(config.denied_patterns, ("*.env", "data/**")) self.assertEqual(config.max_file_bytes, 4096) def test_process_environment_overrides_file_values(self) -> None: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / ".work-item.toml" path.write_text( """ [repository] slug = "owner/file" login = "file-login" assignee = "file-agent" [workflow] require_ci = false """.strip(), encoding="utf-8", ) config = Config.load( path, environment={ "WORK_ITEM_REPO_SLUG": "owner/override", "WORK_ITEM_LOGIN": "ci-login", "WORK_ITEM_REQUIRE_CI": "true", "WORK_ITEM_REQUIRED_TESTS": '["pytest -q"]', }, ) self.assertEqual(config.repo_slug, "owner/override") self.assertEqual(config.login, "ci-login") self.assertEqual(config.assignee, "file-agent") self.assertTrue(config.require_ci) self.assertEqual(config.required_tests, ("pytest -q",)) def test_missing_file_and_required_repository_values_fail(self) -> None: with tempfile.TemporaryDirectory() as directory: base = Path(directory) with self.assertRaisesRegex(WorkItemError, "Missing workflow configuration"): Config.load(base / "missing.toml") path = base / "workflow.toml" path.write_text('[repository]\nslug = "owner/repo"\n', encoding="utf-8") with self.assertRaisesRegex(WorkItemError, "login, assignee"): Config.load(path) def test_malformed_toml_is_reported_as_workflow_error(self) -> None: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "workflow.toml" path.write_text("[repository\n", encoding="utf-8") with self.assertRaisesRegex(WorkItemError, "Invalid workflow configuration"): Config.load(path) def test_invalid_format_dotenv_and_typed_values_fail_cleanly(self) -> None: with tempfile.TemporaryDirectory() as directory: base = Path(directory) unsupported = base / "config.json" unsupported.write_text("{}", encoding="utf-8") with self.assertRaisesRegex(WorkItemError, "Unsupported workflow configuration"): Config.load(unsupported, environment={}) dotenv = base / ".work-item.env" dotenv.write_text("NOT_AN_ASSIGNMENT", encoding="utf-8") with self.assertRaisesRegex(WorkItemError, "Invalid dotenv assignment"): Config.load(dotenv, environment={}) toml = base / ".work-item.toml" toml.write_text( '[repository]\nslug="owner/repo"\nlogin="tea"\nassignee="agent"\n', encoding="utf-8", ) with self.assertRaisesRegex(WorkItemError, "must be true or false"): Config.load(toml, environment={"WORK_ITEM_REQUIRE_CI": "sometimes"}) with self.assertRaisesRegex(WorkItemError, "positive integer"): Config.load(toml, environment={"WORK_ITEM_MAX_FILE_BYTES": "0"}) def test_config_discovery_order_and_explicit_path(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) yaml_path = root / ".work-item.yml" env_path = root / ".work-item.env" yaml_path.write_text("repository: {}\n", encoding="utf-8") env_path.write_text("WORK_ITEM_REPO_SLUG=owner/repo\n", encoding="utf-8") self.assertEqual(find_config(root, None), yaml_path) self.assertEqual(find_config(root, ".work-item.env"), env_path) external = root.parent / "external.yml" self.assertEqual(find_config(root, str(external)), external) def test_config_discovery_reports_all_supported_names(self) -> None: with tempfile.TemporaryDirectory() as directory: with self.assertRaisesRegex(WorkItemError, r"\.work-item\.toml.*\.work-item\.env"): find_config(Path(directory), None) def test_config_discovery_supports_self_contained_subproject(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) subproject = root / "work_item" subproject.mkdir() config = subproject / ".work-item.yaml" config.write_text("repository: {}\n", encoding="utf-8") self.assertEqual(find_config(root, None), config) class RunnerTests(unittest.TestCase): def test_command_and_shell_failures_are_reported(self) -> None: runner = Runner() with tempfile.TemporaryDirectory() as directory: cwd = Path(directory) result = runner.run(("sh", "-c", "exit 7"), cwd=cwd, check=False) self.assertEqual(result.returncode, 7) with self.assertRaisesRegex(WorkItemError, r"Command failed \(7\)"): runner.run(("sh", "-c", "echo failure >&2; exit 7"), cwd=cwd) with self.assertRaisesRegex(WorkItemError, "Test command failed"): runner.run_shell("exit 8", cwd=cwd) class GitSafetyTests(RepositoryFixture): def test_verify_remote_and_clean_tree(self) -> None: git = GitRepo(self.repo, self.config, Mock(wraps=None)) # Use the real runner; assigning after construction keeps the test explicit. from work_item.core import Runner git.runner = Runner() git.verify() git.ensure_clean() (self.repo / "change.txt").write_text("change\n", encoding="utf-8") with self.assertRaisesRegex(WorkItemError, "clean"): git.ensure_clean() def test_rejects_private_paths_large_files_and_secrets(self) -> None: from work_item.core import Runner git = GitRepo(self.repo, self.config, Runner()) (self.repo / "pictures").mkdir() (self.repo / "pictures" / "private.jpg").write_bytes(b"x") with self.assertRaisesRegex(WorkItemError, "denied path"): git.assert_safe_changes() (self.repo / "pictures" / "private.jpg").unlink() (self.repo / "pictures").rmdir() (self.repo / "large.txt").write_text("x" * 65, encoding="utf-8") with self.assertRaisesRegex(WorkItemError, "exceeds"): git.assert_safe_changes() (self.repo / "large.txt").unlink() api_key_sentinel = "api_" + "key = " + "abcdefghijklmnopqrstuvwxyz\n" (self.repo / "secret.txt").write_text(api_key_sentinel, encoding="utf-8") with self.assertRaisesRegex(WorkItemError, "possible secret"): git.assert_safe_changes() def test_fixture_allowlist_overrides_image_deny(self) -> None: from work_item.core import Runner git = GitRepo(self.repo, self.config, Runner()) fixture = self.repo / "tests" / "fixtures" / "synthetic.jpg" fixture.parent.mkdir(parents=True) fixture.write_bytes(b"synthetic") self.assertEqual(git.assert_safe_changes(), [Path("tests/fixtures/synthetic.jpg")]) def test_requires_changes_and_rejects_each_secret_form(self) -> None: git = GitRepo(self.repo, self.config, Runner()) with self.assertRaisesRegex(WorkItemError, "No changes"): git.assert_safe_changes() secrets = ( "access_" + "token = " + "abcdefghijklmnopqrstuvwxyz\n", "-----BEGIN " + "OPENSSH PRIVATE KEY-----\n", "Author" + "ization: Bearer " + "abcdefghijklmnopqrstuvwxyz\n", ) for index, secret in enumerate(secrets): path = self.repo / f"secret-{index}.txt" path.write_text(secret, encoding="utf-8") with ( self.subTest(secret=index), self.assertRaisesRegex(WorkItemError, "possible secret"), ): git.assert_safe_changes() path.unlink() def test_changed_paths_handles_git_rename_records(self) -> None: git = GitRepo(self.repo, self.config, Runner()) run("git", "mv", "README.md", "RENAMED.md", cwd=self.repo) self.assertEqual(git.changed_paths(), [Path("RENAMED.md")]) def test_changed_paths_preserves_modified_and_untracked_filenames(self) -> None: git = GitRepo(self.repo, self.config, Runner()) (self.repo / "README.md").write_text("changed\n", encoding="utf-8") (self.repo / "Another file.md").write_text("new\n", encoding="utf-8") self.assertEqual(set(git.changed_paths()), {Path("README.md"), Path("Another file.md")}) def test_diff_check_rejects_whitespace_errors(self) -> None: git = GitRepo(self.repo, self.config, Runner()) (self.repo / "README.md").write_text("seed\ntrailing whitespace \n", encoding="utf-8") with self.assertRaisesRegex(WorkItemError, "git diff --check"): git.assert_safe_changes() def test_staged_diff_check_catches_whitespace_in_new_files(self) -> None: git = GitRepo(self.repo, self.config, Runner()) story = Story(1, "US01-01", "Safe Commit", 1, 1, frozenset(), (), {}) branch = git.create_story_branch(story) (self.repo / "bad.txt").write_text("trailing whitespace \n", encoding="utf-8") with self.assertRaisesRegex(WorkItemError, "git diff --cached --check"): git.stage_commit_push(story, branch) def test_remote_mismatch_is_rejected(self) -> None: from work_item.core import Runner git = GitRepo(self.repo, config_for("someone/else"), Runner()) with self.assertRaisesRegex(WorkItemError, "does not match"): git.verify() class SelectionTests(unittest.TestCase): def make_workflow(self) -> Workflow: workflow = object.__new__(Workflow) workflow.gitea = Mock() return workflow def test_next_story_uses_numeric_story_order(self) -> None: from work_item.core import Gitea gitea = object.__new__(Gitea) gitea.open_stories = Mock( return_value=[ Story(2, "US01-02", "Second", 1, 2, frozenset(), (), {}), Story(1, "US01-01", "First", 1, 1, frozenset(), (), {}), ] ) gitea.dependencies_closed = Mock(return_value=True) # open_stories normally sorts; make the contract explicit here too. gitea.open_stories.return_value.sort(key=lambda x: (x.epic, x.sequence)) self.assertEqual(gitea.next_story().story_id, "US01-01") def test_next_story_skips_claimed_blocked_and_open_dependencies(self) -> None: from work_item.core import Gitea gitea = object.__new__(Gitea) gitea.open_stories = Mock( return_value=[ Story(1, "US01-01", "Busy", 1, 1, frozenset({"status/in-progress"}), (), {}), Story(2, "US01-02", "Blocked", 1, 2, frozenset({"status/blocked"}), (), {}), Story(3, "US01-03", "Dependency", 1, 3, frozenset(), (), {}), Story(4, "US01-04", "Ready", 1, 4, frozenset(), (), {}), ] ) gitea.dependencies_closed = Mock(side_effect=lambda issue: issue == 4) self.assertEqual(gitea.next_story().number, 4) def test_no_eligible_story_fails(self) -> None: from work_item.core import Gitea gitea = object.__new__(Gitea) gitea.open_stories = Mock(return_value=[]) with self.assertRaisesRegex(WorkItemError, "No eligible"): gitea.next_story() class StateTests(RepositoryFixture): def test_state_is_stored_inside_git_directory(self) -> None: workflow = Workflow(self.repo, self.config) workflow.save_state({"issue": 1}) self.assertTrue(workflow.state_path.is_file()) self.assertIn(".git", workflow.state_path.parts) self.assertEqual(json.loads(workflow.state_path.read_text()), {"issue": 1}) self.assertEqual(run("git", "status", "--porcelain", cwd=self.repo), "") workflow.clear_state() self.assertFalse(workflow.state_path.exists()) def test_invalid_state_file_is_reported(self) -> None: workflow = Workflow(self.repo, self.config) workflow.state_path.write_text("not json", encoding="utf-8") with self.assertRaisesRegex(WorkItemError, "Invalid workflow state"): workflow.load_state() class RecoveryTests(RepositoryFixture): def story(self) -> Story: return Story(1, "US01-01", "Recover Workflow", 1, 1, frozenset(), (), {}) def test_failed_remote_claim_rolls_back_created_branch(self) -> None: workflow = Workflow(self.repo, self.config) workflow.gitea = Mock() workflow.gitea.next_story.return_value = self.story() workflow.gitea.claim.side_effect = WorkItemError("remote claim failed") with self.assertRaisesRegex(WorkItemError, "remote claim failed"): workflow.claim() self.assertEqual(workflow.git.current_branch(), "main") self.assertNotIn("us/US01-01", workflow.git.git("branch")) self.assertFalse(workflow.state_path.exists()) def test_pr_creation_failure_resumes_after_push_without_recommit(self) -> None: config = config_for("domverse/photoanalyzer", max_file_bytes=1000) workflow = Workflow(self.repo, config) story = self.story() branch = workflow.git.create_story_branch(story) workflow.save_state( { "issue": 1, "story_id": story.story_id, "title": story.title, "branch": branch, "status": "in-progress", } ) (self.repo / "feature.txt").write_text("implemented\n", encoding="utf-8") workflow.gitea = Mock() workflow.gitea.find_open_pr.return_value = None workflow.gitea.create_pr.side_effect = [ WorkItemError("response lost"), {"number": 9, "index": 9, "url": "https://example.test/pr/9"}, ] with self.assertRaisesRegex(WorkItemError, "response lost"): workflow.submit(("true",), confirmed=True) pushed = workflow.load_state() self.assertEqual(pushed["status"], "pushed") commit = pushed["commit"] self.assertEqual(workflow.git.git("status", "--porcelain"), "") result = workflow.submit((), confirmed=True) self.assertEqual(result["status"], "review") self.assertEqual(result["commit"], commit) self.assertEqual(workflow.gitea.create_pr.call_count, 2) workflow.gitea.mark_review.assert_called_once() def test_review_label_failure_resumes_from_pushed_commit(self) -> None: config = config_for("domverse/photoanalyzer", max_file_bytes=1000) workflow = Workflow(self.repo, config) story = self.story() branch = workflow.git.create_story_branch(story) workflow.save_state( { "issue": 1, "story_id": story.story_id, "title": story.title, "branch": branch, "status": "in-progress", } ) (self.repo / "feature.txt").write_text("implemented\n", encoding="utf-8") pr = {"number": 9, "index": 9, "url": "https://example.test/pr/9"} workflow.gitea = Mock() workflow.gitea.find_open_pr.side_effect = (None, pr) workflow.gitea.create_pr.return_value = pr workflow.gitea.mark_review.side_effect = (WorkItemError("label update failed"), None) with self.assertRaisesRegex(WorkItemError, "label update failed"): workflow.submit(("true",), confirmed=True) pushed = workflow.load_state() self.assertEqual(pushed["status"], "pushed") commit = pushed["commit"] result = workflow.submit((), confirmed=True) self.assertEqual(result["status"], "review") self.assertEqual(result["commit"], commit) self.assertEqual(workflow.gitea.create_pr.call_count, 1) if __name__ == "__main__": unittest.main()