diff --git a/.work-item.toml b/.work-item.toml deleted file mode 100644 index 3aa3c0e..0000000 --- a/.work-item.toml +++ /dev/null @@ -1,46 +0,0 @@ -[repository] -slug = "domverse/photoanalyzer" -login = "domverse" -assignee = "domverse" -remote = "origin" -main_branch = "main" - -[workflow] -branch_prefix = "us" -require_ci = false -required_tests = ["scripts/python -m unittest discover -s tests -v"] - -[safety] -max_file_bytes = 5000000 -allow = ["tests/fixtures/**"] -deny = [ - "_IGNORE/**", - "**/_IGNORE/**", - "pictures/**", - "photos/**", - "_archive/**", - "_todo/**", - "data/**", - "downloads/**", - "*.db", - "*.db-*", - "*.sqlite", - "*.sqlite3", - "*.log", - "*.env", - "*.pem", - "*.key", - "*.jpg", - "*.jpeg", - "*.png", - "*.gif", - "*.webp", - "*.heic", - "*.heif", - "*.tif", - "*.tiff", - "*.mp4", - "*.mov", - "*.avi", - "*.mkv", -] diff --git a/AGENTS.md b/AGENTS.md index 01b7353..cca53aa 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -6,31 +6,75 @@ the established project invariants: never inspect or process content beneath any `immich-go` uploads rather than External Libraries; keep operations resume-safe; and treat the existing CLI implementations as primary donors rather than rewriting them. +## Work-item helper + +The reusable workflow helper is a self-contained subproject under `work_item/`: + +- launcher: `work_item/scripts/work-item`; +- Python wrapper: `work_item/scripts/python`; +- active project configuration: `work_item/.work-item.yml`; +- implementation: `work_item/src/work_item/`; +- helper tests: `work_item/tests/`; +- complete usage and configuration reference: `work_item/README.md`. + +Do not recreate root-level `scripts/`, `tests/`, or `.work-item.*` files for the +helper. Its launcher adds `work_item/src/` to `PYTHONPATH` and selects Conda base when +it provides Python 3.12 or newer, with supported system Python as fallback. + +The helper supports TOML, YAML, dotenv, and `WORK_ITEM_*` environment overrides. This +repository intentionally keeps its configuration inside `work_item/.work-item.yml`. +Do not change its repository identity, required tests, allow list, deny list, size +limit, or CI policy merely to bypass a workflow failure. + +Workflow state is local at `.git/work-item-state.json` and must never be committed. +`claim` is resume-aware: + +- with no active state, it selects the lowest-numbered eligible story whose Gitea + dependencies are closed, updates `main`, creates the feature branch, assigns the + issue, and applies `status/in-progress`; +- on the active story branch, rerunning it preserves and resumes dirty implementation + work; +- from another clean branch, rerunning it switches back to the claimed branch; +- for a blocked story, rerunning it explicitly returns that same story to + `status/in-progress` instead of selecting a later story; +- a story already in review cannot be replaced by another claim. + +`submit` always combines configured required tests with repeated story-specific +`--test` arguments and runs duplicate commands only once. It checks denied/allowed +paths, maximum file size, likely credentials/private keys, and Git whitespace errors +before allowing a commit. The first invocation without `--yes` is a mandatory review +step; only the confirmed invocation may stage, commit, push, open/reuse a pull request, +record test evidence, and move the issue to review. If push or PR-related remote work +partially succeeds, rerun the same command so its recovery logic can reuse the existing +commit and pull request. + ## Mandatory implementation workflow When asked to start, resume, or continue implementation, work from this isolated Git -repository and use `scripts/work-item`. Gitea issues and their dependencies are the -authoritative backlog. +repository and use `work_item/scripts/work-item`. Gitea issues and their dependencies +are the authoritative backlog. -1. Run `scripts/work-item claim`. -2. Work only on the claimed story and its generated feature branch. +1. Run `work_item/scripts/work-item claim`. +2. Run `work_item/scripts/work-item status` when resuming or whenever branch/state is + uncertain. Work only on the claimed story and its generated feature branch. 3. Read the entire issue, linked specification, dependencies, and acceptance criteria. 4. Inspect the legacy CLI donors before replacing applicable behavior. Update the donor ledger and characterization tests required by the story. 5. Implement every acceptance criterion and its automated tests. 6. Run story-specific tests and the accumulated regression suite required by the epic. -7. Run `scripts/work-item submit --test ""` without `--yes` and - review the reported diff and file list. +7. Run `work_item/scripts/work-item submit --test ""` without + `--yes` and review the reported diff and file list. 8. If the changes are correct and safe, rerun with `--yes`. This commits, pushes, opens a pull request, records test evidence, and moves the issue to review. -9. After review and successful CI, run `scripts/work-item complete --merge`. This - verifies the merge, closes the issue, returns to updated `main`, and removes the - local feature branch. +9. After review and successful CI, run + `work_item/scripts/work-item complete --merge`. This verifies the merge, closes the + issue, returns to updated `main`, and removes the local feature branch. 10. Continue with the next eligible story by returning to step 1. -If genuinely blocked, run `scripts/work-item block --reason ""` and -stop. Never skip to a later story whose dependencies are open. Never mark work done -when tests are missing, skipped, or failing. +If genuinely blocked, run +`work_item/scripts/work-item block --reason ""` and stop. Never skip to +a later story whose dependencies are open. Never mark work done when tests are +missing, skipped, or failing. ## Git and privacy safety @@ -39,10 +83,20 @@ when tests are missing, skipped, or failing. credentials, environment files, caches, generated thumbnails, or runtime data. - Never use force-push, destructive reset, untracked-file deletion, or automatic merge conflict resolution. -- Do not bypass `scripts/work-item` safety checks or manually close implementation - issues to conceal a failed transition. +- Do not bypass `work_item/scripts/work-item` safety checks or manually close + implementation issues to conceal a failed transition. +- Do not edit `.git/work-item-state.json` manually. Use `claim`, `block`, `submit`, and + `complete` so local Git state and Gitea state remain consistent. - Preserve unrelated user changes and stop if the working tree is unexpectedly dirty. +When modifying the helper itself, run all of its own checks from the repository root: + +```bash +ruff check --no-cache work_item/src work_item/tests +ruff format --check --no-cache work_item/src work_item/tests +work_item/scripts/python -m unittest discover -s work_item/tests -v +``` + ## Completion standard A story is complete only when its pull request is merged, required automated tests and diff --git a/README.md b/README.md index 33b5727..c78cb81 100644 --- a/README.md +++ b/README.md @@ -3,32 +3,3 @@ Integrated, restart-safe photo analysis, duplicate review, metadata, upload, and archive workflow. Planning lives in `INTEGRATED_PIPELINE_CONCEPT.md` and `delivery_backlog/`. - -## Agent workflow helper - -Requirements: Python 3.12+, Git, and an authenticated `tea` login named `domverse`. -`scripts/python` prefers Conda base when it satisfies this version, otherwise it uses -an installed Python 3.12+ interpreter. - -```bash -scripts/work-item next -scripts/work-item claim -scripts/work-item status -scripts/work-item submit --test "scripts/python -m unittest discover -s tests -v" -scripts/work-item submit --test "scripts/python -m unittest discover -s tests -v" --yes -scripts/work-item complete --merge -scripts/work-item block --reason "precise blocking condition" -``` - -The helper selects the lowest numbered open story whose dependencies are closed, -creates a story branch, runs required tests, rejects private or unsafe files, creates a -pull request, and closes the issue only after the merge is verified. - -Workflow state is stored inside `.git/` and is never committed. Status labels are the -machine-readable source of truth; the Gitea Project board is the visual view. - -## Tests - -```bash -scripts/python -m unittest discover -s tests -v -``` diff --git a/tests/test_cli_e2e.py b/tests/test_cli_e2e.py deleted file mode 100644 index 466e8b1..0000000 --- a/tests/test_cli_e2e.py +++ /dev/null @@ -1,234 +0,0 @@ -from __future__ import annotations - -import json -import os -import shutil -import subprocess -import sys -import tempfile -import textwrap -import unittest -from pathlib import Path - - -SOURCE_ROOT = Path(__file__).resolve().parents[1] - - -def run(args, *, cwd: Path, env=None, check=True): - result = subprocess.run(args, cwd=cwd, env=env, text=True, capture_output=True) - if check and result.returncode: - raise AssertionError(f"{args}\nstdout={result.stdout}\nstderr={result.stderr}") - return result - - -FAKE_TEA = r'''#!/usr/bin/env python3 -import json, os, subprocess, sys -from pathlib import Path - -state_path = Path(os.environ["FAKE_TEA_STATE"]) -state = json.loads(state_path.read_text()) -args = sys.argv[1:] - -def save(): - state_path.write_text(json.dumps(state, indent=2, sort_keys=True) + "\n") - -def out(value): - print(json.dumps(value)) - -if args[0] == "api": - endpoint = args[1] - if endpoint.endswith("/issues?state=open&type=issues&limit=100"): - out([issue for issue in state["issues"] if issue["state"] == "open"]) - elif "/dependencies" in endpoint: - issue = int(endpoint.split("/issues/")[1].split("/")[0]) - blockers = state["dependencies"].get(str(issue), []) - out([next(x for x in state["issues"] if x["number"] == number) for number in blockers]) - elif endpoint.endswith("/pulls?state=open&limit=50"): - out([pr for pr in state["prs"].values() if not pr["merged"]]) - elif "/pulls/" in endpoint: - number = int(endpoint.rsplit("/", 1)[1]) - out(state["prs"][str(number)]) - elif "/commits/" in endpoint and endpoint.endswith("/status"): - out({"state": "success"}) - else: - raise SystemExit(f"unsupported api: {endpoint}") -elif args[:2] == ["issues", "edit"]: - issue = next(x for x in state["issues"] if x["number"] == int(args[2])) - if "--add-assignees" in args: - issue["assignees"] = [{"login": args[args.index("--add-assignees") + 1]}] - labels = {x["name"] for x in issue.get("labels", [])} - if "--remove-labels" in args: - labels -= set(args[args.index("--remove-labels") + 1].split(",")) - if "--add-labels" in args: - labels |= set(args[args.index("--add-labels") + 1].split(",")) - issue["labels"] = [{"name": x} for x in sorted(labels)] - save(); out(issue) -elif args[:2] == ["comments", "add"]: - state["comments"].append({"issue": int(args[2]), "body": args[3]}) - save(); out(state["comments"][-1]) -elif args[:2] == ["pulls", "create"]: - number = len(state["prs"]) + 1 - branch = args[args.index("--head") + 1] - sha = subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip() - pr = {"number": number, "index": number, "url": f"https://example.test/pr/{number}", - "merged": False, "head": {"sha": sha, "ref": branch}} - state["prs"][str(number)] = pr - save(); out(pr) -elif args[:2] == ["pulls", "merge"]: - number = int(args[2]); pr = state["prs"][str(number)] - subprocess.check_call(["git", "push", "origin", f"{pr['head']['ref']}:main"]) - pr["merged"] = True - save(); out(pr) -elif args[:2] == ["issues", "close"]: - issue = next(x for x in state["issues"] if x["number"] == int(args[2])) - issue["state"] = "closed"; save(); out(issue) -else: - raise SystemExit("unsupported tea command: " + repr(args)) -''' - - -class CliEndToEndTests(unittest.TestCase): - def setUp(self) -> None: - self.temp = tempfile.TemporaryDirectory() - self.base = Path(self.temp.name) - remote_parent = self.base / "domverse" - remote_parent.mkdir() - seed = self.base / "seed" - seed.mkdir() - run(["git", "init", "-b", "main"], cwd=seed) - run(["git", "config", "user.name", "Test Agent"], cwd=seed) - run(["git", "config", "user.email", "agent@example.test"], cwd=seed) - (seed / "README.md").write_text("seed\n") - run(["git", "add", "README.md"], cwd=seed) - run(["git", "commit", "-m", "seed"], cwd=seed) - self.remote = remote_parent / "photoanalyzer.git" - run(["git", "clone", "--bare", str(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) - - config = textwrap.dedent(f'''\ - [repository] - slug = "domverse/photoanalyzer" - login = "fake" - assignee = "agent" - remote = "origin" - main_branch = "main" - [workflow] - branch_prefix = "us" - require_ci = true - required_tests = [] - [safety] - max_file_bytes = 100000 - allow = ["tests/fixtures/**"] - deny = ["_IGNORE/**", "**/_IGNORE/**", "pictures/**", "*.env", "*.jpg"] - ''') - (self.repo / ".work-item.toml").write_text(config) - run(["git", "add", ".work-item.toml"], cwd=self.repo) - run(["git", "commit", "-m", "config"], cwd=self.repo) - run(["git", "push", "origin", "main"], cwd=self.repo) - - bin_dir = self.base / "bin" - bin_dir.mkdir() - tea = bin_dir / "tea" - tea.write_text(FAKE_TEA) - tea.chmod(0o755) - self.state_path = self.base / "tea-state.json" - self.state_path.write_text(json.dumps({ - "issues": [{ - "number": 1, - "title": "US01-01 — Implement Safe Workflow", - "state": "open", - "labels": [{"name": "status/backlog"}], - "assignees": [], - }], - "dependencies": {"1": []}, - "comments": [], - "prs": {}, - })) - self.env = os.environ.copy() - self.env["PATH"] = f"{bin_dir}{os.pathsep}{self.env['PATH']}" - self.env["FAKE_TEA_STATE"] = str(self.state_path) - self.env["PYTHONPATH"] = str(SOURCE_ROOT) - - def tearDown(self) -> None: - self.temp.cleanup() - - def cli(self, *args, check=True): - return run([sys.executable, "-m", "work_item", *args], cwd=self.repo, env=self.env, check=check) - - def state(self): - return json.loads(self.state_path.read_text()) - - def test_claim_submit_merge_complete_lifecycle(self) -> None: - next_result = self.cli("next") - self.assertEqual(json.loads(next_result.stdout)["story_id"], "US01-01") - - claim = json.loads(self.cli("claim").stdout) - self.assertEqual(claim["status"], "in-progress") - self.assertEqual(run(["git", "branch", "--show-current"], cwd=self.repo).stdout.strip(), claim["branch"]) - issue = self.state()["issues"][0] - self.assertEqual(issue["assignees"], [{"login": "agent"}]) - self.assertIn({"name": "status/in-progress"}, issue["labels"]) - - (self.repo / "private.env").write_text("not-a-real-secret\n") - unsafe = self.cli("submit", "--test", "true", "--yes", check=False) - self.assertEqual(unsafe.returncode, 2) - self.assertIn("denied path", unsafe.stderr) - (self.repo / "private.env").unlink() - - (self.repo / "feature.txt").write_text("implemented\n") - review_only = self.cli("submit", "--test", "true", check=False) - self.assertEqual(review_only.returncode, 2) - self.assertIn("rerun with --yes", review_only.stderr) - self.assertEqual(run(["git", "status", "--porcelain"], cwd=self.repo).stdout.strip(), "?? feature.txt") - - submitted = json.loads(self.cli("submit", "--test", "true", "--yes").stdout) - self.assertEqual(submitted["status"], "review") - self.assertEqual(submitted["pr"], 1) - self.assertIn("status/review", {x["name"] for x in self.state()["issues"][0]["labels"]}) - - not_merged = self.cli("complete", check=False) - self.assertEqual(not_merged.returncode, 2) - self.assertIn("not merged", not_merged.stderr) - - completed = json.loads(self.cli("complete", "--merge").stdout) - self.assertEqual(completed["status"], "done") - self.assertEqual(run(["git", "branch", "--show-current"], cwd=self.repo).stdout.strip(), "main") - self.assertEqual(run(["git", "status", "--porcelain"], cwd=self.repo).stdout.strip(), "") - issue = self.state()["issues"][0] - self.assertEqual(issue["state"], "closed") - self.assertIn("status/done", {x["name"] for x in issue["labels"]}) - self.assertFalse((self.repo / ".git" / "work-item-state.json").exists()) - - def test_dirty_tree_prevents_claim(self) -> None: - (self.repo / "unexpected.txt").write_text("dirty\n") - result = self.cli("claim", check=False) - self.assertEqual(result.returncode, 2) - self.assertIn("clean", result.stderr) - self.assertEqual(run(["git", "branch", "--show-current"], cwd=self.repo).stdout.strip(), "main") - - def test_block_requires_reason_and_records_state(self) -> None: - self.cli("claim") - result = self.cli("block", "--reason", "Provider fixture unavailable") - payload = json.loads(result.stdout) - self.assertEqual(payload["status"], "blocked") - issue = self.state()["issues"][0] - self.assertIn("status/blocked", {x["name"] for x in issue["labels"]}) - self.assertIn("Provider fixture unavailable", self.state()["comments"][-1]["body"]) - - -class InterpreterWrapperTests(unittest.TestCase): - def test_wrapper_selects_supported_python(self) -> None: - result = run( - [str(SOURCE_ROOT / "scripts" / "python"), "-c", "import sys; print(sys.version_info[:2])"], - cwd=SOURCE_ROOT, - ) - major, minor = eval(result.stdout.strip(), {"__builtins__": {}}) - self.assertEqual(major, 3) - self.assertGreaterEqual(minor, 12) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_core.py b/tests/test_core.py deleted file mode 100644 index 0c52c1e..0000000 --- a/tests/test_core.py +++ /dev/null @@ -1,227 +0,0 @@ -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, Story, WorkItemError, Workflow - - -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"})) - - -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() - - (self.repo / "secret.txt").write_text("api_key = abcdefghijklmnopqrstuvwxyz\n", 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_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()) - - -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() - - -if __name__ == "__main__": - unittest.main() diff --git a/work_item/.work-item.yml b/work_item/.work-item.yml new file mode 100644 index 0000000..a9bc586 --- /dev/null +++ b/work_item/.work-item.yml @@ -0,0 +1,47 @@ +repository: + slug: domverse/photoanalyzer + login: domverse + assignee: domverse + remote: origin + main_branch: main + +workflow: + branch_prefix: us + require_ci: false + required_tests: + - work_item/scripts/python -m unittest discover -s work_item/tests -v + +safety: + max_file_bytes: 5000000 + allow: + - tests/fixtures/** + deny: + - _IGNORE/** + - "**/_IGNORE/**" + - pictures/** + - photos/** + - _archive/** + - _todo/** + - data/** + - downloads/** + - "*.db" + - "*.db-*" + - "*.sqlite" + - "*.sqlite3" + - "*.log" + - "*.env" + - "*.pem" + - "*.key" + - "*.jpg" + - "*.jpeg" + - "*.png" + - "*.gif" + - "*.webp" + - "*.heic" + - "*.heif" + - "*.tif" + - "*.tiff" + - "*.mp4" + - "*.mov" + - "*.avi" + - "*.mkv" diff --git a/work_item/README.md b/work_item/README.md new file mode 100644 index 0000000..687fd6c --- /dev/null +++ b/work_item/README.md @@ -0,0 +1,227 @@ +# work-item + +Safe, deterministic Git and Gitea workflow automation for a single agent working an +ordered user-story backlog. + +`work-item` selects the first eligible story, creates its feature branch, runs the +configured automated tests, checks changes for unsafe content, pushes the commit, +opens a pull request, and closes the issue only after the merge is verified. + +## Highlights + +- deterministic numeric story selection with dependency checks; +- one active story at a time, with resume-safe local state; +- idempotent resume after blocked work or partially completed remote operations; +- mandatory automated tests before commit and push; +- configurable file, size, secret, and private-data safety checks; +- pull-request creation, optional CI gating, merge verification, and issue closure; +- TOML, YAML, dotenv, and process-environment configuration; +- no workflow state committed to the repository. + +## Requirements + +- Python 3.12 or newer; +- Git; +- [`tea`](https://gitea.com/gitea/tea), authenticated against the target Gitea + instance; +- PyYAML 6 or newer when using `.yml` or `.yaml` configuration. + +The repository must use story titles in this form: + +```text +US01-01 — Short imperative title +``` + +The two numeric groups define epic and story order. The helper recognizes these +status labels: + +```text +status/backlog status/ready status/in-progress +status/review status/blocked status/done +``` + +Dependencies must be configured through Gitea's issue-dependency feature. + +## Installation + +The directory is a self-contained Python subproject. From the host repository root: + +```bash +python -m pip install -e work_item +work-item --help +``` + +For a repository-local installation, copy the complete `work_item/` directory into the +target repository and run `work_item/scripts/work-item`. Its Python wrapper adds the +bundled `src/` directory to `PYTHONPATH`, prefers a compatible Conda base interpreter, +and otherwise uses `python3.13` or `python3.12`. + +```text +work_item/ +├── .work-item.yml # host-project configuration +├── README.md +├── pyproject.toml # independently installable package +├── scripts/ # repository-local launchers +├── src/work_item/ # implementation +└── tests/ # unit and fake-Gitea end-to-end tests +``` + +## Quick start + +Create one supported configuration file either in the repository root or inside the +`work_item/` directory. The repository root takes precedence; within each location, +automatic discovery uses this order: + +1. `.work-item.toml` +2. `.work-item.yml` +3. `.work-item.yaml` +4. `.work-item.env` + +Then run: + +```bash +work_item/scripts/work-item next +work_item/scripts/work-item claim + +# Implement the claimed story and its tests. +work_item/scripts/work-item status +work_item/scripts/work-item submit --test "pytest -q" + +# Review the displayed diff, then authorize Git/PR creation. +work_item/scripts/work-item submit --test "pytest -q" --yes + +# After review and successful CI: +work_item/scripts/work-item complete --merge +``` + +When installed, the shorter `work-item` command is equivalent. Use a non-default file +explicitly with `work-item --config path/to/config.yml next`. + +## Configuration + +### TOML + +```toml +[repository] +slug = "owner/project" +login = "my-tea-login" +assignee = "agent-user" +remote = "origin" +main_branch = "main" + +[workflow] +branch_prefix = "us" +require_ci = true +required_tests = ["pytest -q", "ruff check ."] + +[safety] +max_file_bytes = 5000000 +allow = ["tests/fixtures/**"] +deny = ["*.env", "*.pem", "data/**", "photos/**", "*.jpg"] +``` + +### YAML + +```yaml +repository: + slug: owner/project + login: my-tea-login + assignee: agent-user + remote: origin + main_branch: main + +workflow: + branch_prefix: us + require_ci: true + required_tests: + - pytest -q + - ruff check . + +safety: + max_file_bytes: 5000000 + allow: + - tests/fixtures/** + deny: + - "*.env" + - "*.pem" + - data/** + - photos/** +``` + +### dotenv + +Lists use JSON arrays so test commands and glob patterns remain unambiguous: + +```dotenv +WORK_ITEM_REPO_SLUG=owner/project +WORK_ITEM_LOGIN=my-tea-login +WORK_ITEM_ASSIGNEE=agent-user +WORK_ITEM_REMOTE=origin +WORK_ITEM_MAIN_BRANCH=main +WORK_ITEM_BRANCH_PREFIX=us +WORK_ITEM_REQUIRE_CI=true +WORK_ITEM_REQUIRED_TESTS='["pytest -q", "ruff check ."]' +WORK_ITEM_MAX_FILE_BYTES=5000000 +WORK_ITEM_ALLOW='[".work-item.env", "tests/fixtures/**"]' +WORK_ITEM_DENY='["*.env", "*.pem", "data/**", "photos/**"]' +``` + +The same `WORK_ITEM_*` variables may be exported by the calling process. Process +environment values override values from any configuration file, which is useful for +CI or per-machine `tea` login and assignee settings. + +Only repository slug, login, and assignee are required. Other fields use the defaults +shown above. + +## Commands + +| Command | Behavior | +|---|---| +| `next` | Shows the lowest-numbered unassigned story whose dependencies are closed. | +| `claim` | Claims the next story, or safely resumes the existing in-progress/blocked story. | +| `status` | Shows local workflow state, current branch, and changed files. | +| `submit` | Runs tests and safety checks; without `--yes`, stops after the review summary. | +| `complete` | Verifies CI/merge state, closes the issue, updates main, and removes the branch. | +| `block` | Records a precise blocking reason without discarding the branch or local work. | + +`--test` is repeatable. Configured `required_tests` are always included, and duplicate +commands are run only once. + +## Safety model + +Before submission, the helper: + +1. enumerates tracked and untracked changes; +2. rejects denied paths unless an allow pattern explicitly overrides them; +3. rejects files larger than `max_file_bytes`; +4. scans readable text for common credentials, authorization headers, and private keys; +5. runs `git diff --check` before staging and again against the staged diff; +6. runs every required and story-specific automated test; +7. requires a separate `--yes` invocation before commit and push. + +Keep real media, databases, runtime data, credentials, and generated artifacts in the +deny list. Allow only deterministic, non-private test fixtures. + +## Resume and recovery + +State is stored at `.git/work-item-state.json`, so it is local and never committed. + +- Re-running `claim` on the story branch preserves dirty implementation work. +- Re-running `claim` from another clean branch switches back to the claimed branch. +- Re-running `claim` after `block` returns the same story to `status/in-progress`. +- If push succeeds but PR creation or review labeling fails, `submit --yes` reuses the + pushed commit and existing PR instead of committing twice. +- A story in review cannot be replaced by a new claim; complete or block it first. + +## Development + +From this repository: + +```bash +ruff check --no-cache work_item/src work_item/tests +ruff format --check --no-cache work_item/src work_item/tests +work_item/scripts/python -m unittest discover -s work_item/tests -v +``` + +The end-to-end tests use temporary Git repositories and a deterministic fake `tea` +binary. They do not mutate a live Gitea project. diff --git a/work_item/pyproject.toml b/work_item/pyproject.toml new file mode 100644 index 0000000..704fdf3 --- /dev/null +++ b/work_item/pyproject.toml @@ -0,0 +1,20 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "gitea-work-item" +version = "0.1.0" +description = "Safe single-agent Git and Gitea backlog workflow automation" +readme = "README.md" +requires-python = ">=3.12" +dependencies = ["PyYAML>=6.0"] + +[project.scripts] +work-item = "work_item.core:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.ruff] +line-length = 100 diff --git a/scripts/python b/work_item/scripts/python similarity index 60% rename from scripts/python rename to work_item/scripts/python index 6bcbf38..8ceacab 100755 --- a/scripts/python +++ b/work_item/scripts/python @@ -1,6 +1,14 @@ #!/bin/sh set -eu +SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd) +WORK_ITEM_ROOT=$(CDPATH= cd -- "$SCRIPT_DIR/.." && pwd) +if [ -n "${PYTHONPATH:-}" ]; then + export PYTHONPATH="$WORK_ITEM_ROOT/src:$PYTHONPATH" +else + export PYTHONPATH="$WORK_ITEM_ROOT/src" +fi + if command -v conda >/dev/null 2>&1; then conda_version=$(conda run -n base python -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")' 2>/dev/null || true) case "$conda_version" in @@ -14,5 +22,5 @@ for python in python3.13 python3.12; do fi done -echo "photoanalyzer: Python 3.12 or newer is required" >&2 +echo "work-item: Python 3.12 or newer is required" >&2 exit 2 diff --git a/scripts/work-item b/work_item/scripts/work-item similarity index 100% rename from scripts/work-item rename to work_item/scripts/work-item diff --git a/work_item/__init__.py b/work_item/src/work_item/__init__.py similarity index 100% rename from work_item/__init__.py rename to work_item/src/work_item/__init__.py diff --git a/work_item/__main__.py b/work_item/src/work_item/__main__.py similarity index 100% rename from work_item/__main__.py rename to work_item/src/work_item/__main__.py diff --git a/work_item/core.py b/work_item/src/work_item/core.py similarity index 59% rename from work_item/core.py rename to work_item/src/work_item/core.py index 61d70b1..ed8f0ad 100644 --- a/work_item/core.py +++ b/work_item/src/work_item/core.py @@ -10,7 +10,7 @@ import subprocess import sys from dataclasses import dataclass from pathlib import Path -from typing import Any, Iterable, Sequence +from typing import Any, Iterable, Mapping, Sequence try: import tomllib @@ -19,8 +19,24 @@ except ModuleNotFoundError: # pragma: no cover - Python < 3.12 is unsupported STORY_RE = re.compile(r"^(US(?P\d{2})-(?P\d{2}))\s+—\s+(?P.+)$") +CONFIG_FILENAMES = (".work-item.toml", ".work-item.yml", ".work-item.yaml", ".work-item.env") +ENVIRONMENT_FIELDS = { + "WORK_ITEM_REPO_SLUG": ("repository", "slug"), + "WORK_ITEM_LOGIN": ("repository", "login"), + "WORK_ITEM_ASSIGNEE": ("repository", "assignee"), + "WORK_ITEM_REMOTE": ("repository", "remote"), + "WORK_ITEM_MAIN_BRANCH": ("repository", "main_branch"), + "WORK_ITEM_BRANCH_PREFIX": ("workflow", "branch_prefix"), + "WORK_ITEM_REQUIRE_CI": ("workflow", "require_ci"), + "WORK_ITEM_REQUIRED_TESTS": ("workflow", "required_tests"), + "WORK_ITEM_MAX_FILE_BYTES": ("safety", "max_file_bytes"), + "WORK_ITEM_DENY": ("safety", "deny"), + "WORK_ITEM_ALLOW": ("safety", "allow"), +} SECRET_PATTERNS = ( - re.compile(r"(?i)(api[_-]?key|access[_-]?token|client[_-]?secret|password)\s*[:=]\s*['\"]?[A-Za-z0-9_./+\-=]{12,}"), + re.compile( + r"(?i)(api[_-]?key|access[_-]?token|client[_-]?secret|password)\s*[:=]\s*['\"]?[A-Za-z0-9_./+\-=]{12,}" + ), re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"), re.compile(r"(?i)authorization:\s*(?:bearer|token)\s+[A-Za-z0-9._~+\-/=]{12,}"), ) @@ -30,6 +46,106 @@ class WorkItemError(RuntimeError): pass +def _merge_config(base: dict[str, Any], override: Mapping[str, Any]) -> dict[str, Any]: + result = dict(base) + for key, value in override.items(): + if isinstance(value, Mapping) and isinstance(result.get(key), Mapping): + result[key] = _merge_config(dict(result[key]), value) + else: + result[key] = value + return result + + +def _config_from_environment(environment: Mapping[str, str]) -> dict[str, Any]: + result: dict[str, Any] = {} + for variable, (section, key) in ENVIRONMENT_FIELDS.items(): + if variable in environment: + result.setdefault(section, {})[key] = environment[variable] + return result + + +def _dotenv_value(raw: str) -> str: + value = raw.strip() + if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: + if value[0] == '"': + try: + return json.loads(value) + except json.JSONDecodeError as error: + raise WorkItemError(f"Invalid quoted dotenv value: {raw}") from error + return value[1:-1] + comment = re.search(r"\s+#", value) + return value[: comment.start()].rstrip() if comment else value + + +def _load_dotenv(path: Path) -> dict[str, Any]: + environment: dict[str, str] = {} + for line_number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): + line = raw_line.strip() + if not line or line.startswith("#"): + continue + if line.startswith("export "): + line = line[7:].lstrip() + if "=" not in line: + raise WorkItemError(f"Invalid dotenv assignment at {path}:{line_number}") + key, value = line.split("=", 1) + key = key.strip() + if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", key): + raise WorkItemError(f"Invalid dotenv variable at {path}:{line_number}: {key}") + environment[key] = _dotenv_value(value) + return _config_from_environment(environment) + + +def _load_yaml(path: Path) -> dict[str, Any]: + try: + import yaml + except ModuleNotFoundError as error: # pragma: no cover - dependency declared by package + raise WorkItemError("YAML configuration requires PyYAML 6 or newer") from error + try: + raw = yaml.safe_load(path.read_text(encoding="utf-8")) + except yaml.YAMLError as error: + raise WorkItemError(f"Invalid workflow configuration {path}: {error}") from error + if raw is None: + return {} + if not isinstance(raw, dict): + raise WorkItemError(f"Workflow configuration must be a mapping: {path}") + return raw + + +def _as_bool(value: Any, field: str) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in {"1", "true", "yes", "on"}: + return True + if normalized in {"0", "false", "no", "off", ""}: + return False + raise WorkItemError(f"{field} must be true or false") + + +def _as_list(value: Any, field: str) -> tuple[str, ...]: + if value is None: + return () + if isinstance(value, (list, tuple)): + return tuple(str(item) for item in value) + if isinstance(value, str): + stripped = value.strip() + if not stripped: + return () + if stripped.startswith("["): + try: + parsed = json.loads(stripped) + except json.JSONDecodeError as error: + raise WorkItemError( + f"{field} must be a JSON array or newline-separated list" + ) from error + if not isinstance(parsed, list): + raise WorkItemError(f"{field} must be a JSON array or newline-separated list") + return tuple(str(item) for item in parsed) + return tuple(item.strip() for item in stripped.splitlines() if item.strip()) + raise WorkItemError(f"{field} must be a list") + + class Runner: def run( self, @@ -75,32 +191,56 @@ class Config: allowed_patterns: tuple[str, ...] = () @classmethod - def load(cls, path: Path) -> "Config": - if tomllib is None: - raise WorkItemError("Python 3.12 or newer is required") + def load(cls, path: Path, *, environment: Mapping[str, str] | None = None) -> "Config": if not path.is_file(): raise WorkItemError(f"Missing workflow configuration: {path}") - with path.open("rb") as handle: - raw = tomllib.load(handle) + suffix = path.suffix.lower() + if suffix == ".toml": + if tomllib is None: + raise WorkItemError("TOML configuration requires Python 3.11 or newer") + try: + with path.open("rb") as handle: + raw = tomllib.load(handle) + except tomllib.TOMLDecodeError as error: + raise WorkItemError(f"Invalid workflow configuration {path}: {error}") from error + elif suffix in {".yml", ".yaml"}: + raw = _load_yaml(path) + elif suffix == ".env": + raw = _load_dotenv(path) + else: + raise WorkItemError( + f"Unsupported workflow configuration format {suffix or '<none>'}; " + "use .toml, .yml, .yaml, or .env" + ) + active_environment = os.environ if environment is None else environment + raw = _merge_config(raw, _config_from_environment(active_environment)) repo = raw.get("repository", {}) workflow = raw.get("workflow", {}) safety = raw.get("safety", {}) + if not all(isinstance(section, Mapping) for section in (repo, workflow, safety)): + raise WorkItemError("repository, workflow, and safety configuration must be mappings") required = ("slug", "login", "assignee") missing = [key for key in required if not repo.get(key)] if missing: raise WorkItemError(f"Missing repository configuration: {', '.join(missing)}") + try: + max_file_bytes = int(safety.get("max_file_bytes", 5_000_000)) + except (TypeError, ValueError) as error: + raise WorkItemError("safety.max_file_bytes must be a positive integer") from error + if max_file_bytes <= 0: + raise WorkItemError("safety.max_file_bytes must be a positive integer") return cls( - repo_slug=repo["slug"], - login=repo["login"], - assignee=repo["assignee"], - remote=repo.get("remote", "origin"), - main_branch=repo.get("main_branch", "main"), - branch_prefix=workflow.get("branch_prefix", "us"), - require_ci=bool(workflow.get("require_ci", False)), - max_file_bytes=int(safety.get("max_file_bytes", 5_000_000)), - required_tests=tuple(workflow.get("required_tests", ())), - denied_patterns=tuple(safety.get("deny", ())), - allowed_patterns=tuple(safety.get("allow", ())), + repo_slug=str(repo["slug"]), + login=str(repo["login"]), + assignee=str(repo["assignee"]), + remote=str(repo.get("remote", "origin")), + main_branch=str(repo.get("main_branch", "main")), + branch_prefix=str(workflow.get("branch_prefix", "us")), + require_ci=_as_bool(workflow.get("require_ci", False), "workflow.require_ci"), + max_file_bytes=max_file_bytes, + required_tests=_as_list(workflow.get("required_tests"), "workflow.required_tests"), + denied_patterns=_as_list(safety.get("deny"), "safety.deny"), + allowed_patterns=_as_list(safety.get("allow"), "safety.allow"), ) @@ -161,7 +301,7 @@ class GitRepo: def ensure_clean(self) -> None: if self.git("status", "--porcelain=v1"): - raise WorkItemError("Working tree must be clean before claiming a story") + raise WorkItemError("Working tree must be clean for this workflow operation") def update_main(self) -> None: self.git("fetch", "--prune", self.config.remote) @@ -204,8 +344,12 @@ class GitRepo: violations: list[str] = [] for relative in paths: portable = relative.as_posix() - allowed = any(fnmatch.fnmatch(portable, pattern) for pattern in self.config.allowed_patterns) - denied = any(fnmatch.fnmatch(portable, pattern) for pattern in self.config.denied_patterns) + allowed = any( + fnmatch.fnmatch(portable, pattern) for pattern in self.config.allowed_patterns + ) + denied = any( + fnmatch.fnmatch(portable, pattern) for pattern in self.config.denied_patterns + ) if denied and not allowed: violations.append(f"denied path: {portable}") continue @@ -232,6 +376,7 @@ class GitRepo: staged = self.git("diff", "--cached", "--name-only") if not staged: raise WorkItemError("Nothing was staged") + self.git("diff", "--cached", "--check") subject = f"{story.story_id}: {story.title}" self.git("commit", "-m", subject) self.git("push", "--set-upstream", self.config.remote, branch) @@ -253,7 +398,14 @@ class Gitea: if data is not None: args.extend(("--data", json.dumps(data))) output = self.tea(*args) - return json.loads(output) if output else None + if not output: + return None + try: + return json.loads(output) + except json.JSONDecodeError as error: + raise WorkItemError( + f"Could not parse Gitea API response for {endpoint}: {output}" + ) from error @property def base(self) -> str: @@ -289,33 +441,50 @@ class Gitea: def claim(self, story: Story) -> None: self.tea( - "issues", "edit", str(story.number), - "--repo", self.config.repo_slug, - "--add-assignees", self.config.assignee, - "--add-labels", "status/in-progress", - "--remove-labels", "status/backlog,status/ready,status/blocked,status/review,status/done", + "issues", + "edit", + str(story.number), + "--repo", + self.config.repo_slug, + "--add-assignees", + self.config.assignee, + "--add-labels", + "status/in-progress", + "--remove-labels", + "status/backlog,status/ready,status/blocked,status/review,status/done", ) def comment(self, issue: int, body: str) -> None: self.tea("comments", "add", str(issue), body, "--repo", self.config.repo_slug) - def create_pr(self, story: Story, branch: str, tests: Sequence[str], commit: str) -> dict[str, Any]: - body = "\n".join(( - f"Implements #{story.number}", - "", - "Tests:", - *(f"- `{command}`" for command in tests), - "", - f"Commit: `{commit}`", - )) + def create_pr( + self, story: Story, branch: str, tests: Sequence[str], commit: str + ) -> dict[str, Any]: + body = "\n".join( + ( + f"Implements #{story.number}", + "", + "Tests:", + *(f"- `{command}`" for command in tests), + "", + f"Commit: `{commit}`", + ) + ) output = self.tea( - "pulls", "create", - "--repo", self.config.repo_slug, - "--head", branch, - "--base", self.config.main_branch, - "--title", f"{story.story_id}: {story.title}", - "--description", body, - "--output", "json", + "pulls", + "create", + "--repo", + self.config.repo_slug, + "--head", + branch, + "--base", + self.config.main_branch, + "--title", + f"{story.story_id}: {story.title}", + "--description", + body, + "--output", + "json", ) try: return json.loads(output) @@ -333,7 +502,9 @@ class Gitea: return self.api(f"{self.base}/pulls/{number}") def merge_pr(self, number: int) -> None: - self.tea("pulls", "merge", str(number), "--repo", self.config.repo_slug, "--style", "squash") + self.tea( + "pulls", "merge", str(number), "--repo", self.config.repo_slug, "--style", "squash" + ) def require_ci_success(self, pr: dict[str, Any]) -> None: if not self.config.require_ci: @@ -341,23 +512,41 @@ class Gitea: sha = pr["head"]["sha"] combined = self.api(f"{self.base}/commits/{sha}/status") if combined.get("state") != "success": - raise WorkItemError(f"Required CI is not successful: {combined.get('state', 'unknown')}") + raise WorkItemError( + f"Required CI is not successful: {combined.get('state', 'unknown')}" + ) def mark_review(self, issue: int, pr: dict[str, Any], tests: Sequence[str]) -> None: self.edit_labels( issue, add=("status/review",), - remove=("status/in-progress", "status/backlog", "status/ready", "status/blocked", "status/done"), + remove=( + "status/in-progress", + "status/backlog", + "status/ready", + "status/blocked", + "status/done", + ), ) number = pr.get("index") or pr.get("number") url = pr.get("url") or pr.get("html_url") or f"PR #{number}" - self.comment(issue, f"Submitted for review: {url}\n\nTests passed:\n" + "\n".join(f"- `{x}`" for x in tests)) + self.comment( + issue, + f"Submitted for review: {url}\n\nTests passed:\n" + + "\n".join(f"- `{x}`" for x in tests), + ) def mark_done(self, issue: int, pr_number: int) -> None: self.edit_labels( issue, add=("status/done",), - remove=("status/in-progress", "status/backlog", "status/ready", "status/blocked", "status/review"), + remove=( + "status/in-progress", + "status/backlog", + "status/ready", + "status/blocked", + "status/review", + ), ) self.comment(issue, f"Completed and merged via PR #{pr_number}.") self.tea("issues", "close", str(issue), "--repo", self.config.repo_slug) @@ -384,7 +573,10 @@ class Workflow: def load_state(self) -> dict[str, Any]: if not self.state_path.is_file(): raise WorkItemError("No story is currently claimed") - return json.loads(self.state_path.read_text(encoding="utf-8")) + try: + return json.loads(self.state_path.read_text(encoding="utf-8")) + except json.JSONDecodeError as error: + raise WorkItemError(f"Invalid workflow state: {self.state_path}") from error def save_state(self, state: dict[str, Any]) -> None: temporary = self.state_path.with_suffix(".tmp") @@ -400,9 +592,30 @@ class Workflow: def claim(self) -> dict[str, Any]: self.git.verify() - self.git.ensure_clean() if self.state_path.exists(): - raise WorkItemError("A story is already claimed; complete or block it first") + state = self.load_state() + status = state.get("status") + if status not in {"in-progress", "blocked"}: + raise WorkItemError( + f"Claimed story is {status!r}; finish its review/completion before claiming another" + ) + if self.git.current_branch() != state["branch"]: + self.git.ensure_clean() + self.git.git("switch", state["branch"]) + if status == "blocked": + self.gitea.edit_labels( + int(state["issue"]), + add=("status/in-progress",), + remove=("status/blocked", "status/ready", "status/review", "status/done"), + ) + self.gitea.comment( + int(state["issue"]), f"Resumed implementation on branch `{state['branch']}`." + ) + state.pop("reason", None) + state["status"] = "in-progress" + self.save_state(state) + return state + self.git.ensure_clean() self.git.update_main() story = self.gitea.next_story() branch = self.git.create_story_branch(story) @@ -426,17 +639,28 @@ class Workflow: def status(self) -> dict[str, Any]: self.git.verify() state = self.load_state() - return {**state, "current_branch": self.git.current_branch(), "changes": [p.as_posix() for p in self.git.changed_paths()]} + return { + **state, + "current_branch": self.git.current_branch(), + "changes": [p.as_posix() for p in self.git.changed_paths()], + } def submit(self, tests: Sequence[str], *, confirmed: bool) -> dict[str, Any]: self.git.verify() state = self.load_state() if self.git.current_branch() != state["branch"]: - raise WorkItemError(f"Expected branch {state['branch']}, found {self.git.current_branch()}") + raise WorkItemError( + f"Expected branch {state['branch']}, found {self.git.current_branch()}" + ) story = Story( - number=int(state["issue"]), story_id=state["story_id"], title=state["title"], - epic=int(state["story_id"][2:4]), sequence=int(state["story_id"][5:7]), - labels=frozenset(), assignees=(), raw={}, + number=int(state["issue"]), + story_id=state["story_id"], + title=state["title"], + epic=int(state["story_id"][2:4]), + sequence=int(state["story_id"][5:7]), + labels=frozenset(), + assignees=(), + raw={}, ) if state.get("status") == "pushed": commands = tuple(state["tests"]) @@ -465,9 +689,11 @@ class Workflow: if pr is None: pr = self.gitea.create_pr(story, state["branch"], commands, commit) pr_number = int(pr.get("index") or pr.get("number")) - state.update({"status": "review", "pr": pr_number, "commit": commit, "tests": list(commands)}) - self.save_state(state) + state.update( + {"status": "review", "pr": pr_number, "commit": commit, "tests": list(commands)} + ) self.gitea.mark_review(story.number, pr, commands) + self.save_state(state) return state def complete(self, *, merge: bool) -> dict[str, Any]: @@ -475,6 +701,7 @@ class Workflow: state = self.load_state() if state.get("status") != "review" or not state.get("pr"): raise WorkItemError("The claimed story has not been submitted for review") + self.git.ensure_clean() pr = self.gitea.pr(int(state["pr"])) self.gitea.require_ci_success(pr) if not pr.get("merged"): @@ -513,26 +740,56 @@ class Workflow: def find_root(start: Path) -> Path: result = subprocess.run( - ("git", "rev-parse", "--show-toplevel"), cwd=start, text=True, - stdout=subprocess.PIPE, stderr=subprocess.PIPE, + ("git", "rev-parse", "--show-toplevel"), + cwd=start, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, ) if result.returncode: - raise WorkItemError("This command must run inside the photoanalyzer Git repository") + raise WorkItemError("This command must run inside a Git repository") return Path(result.stdout.strip()).resolve() +def find_config(root: Path, requested: str | None) -> Path: + if requested: + path = Path(requested) + return path if path.is_absolute() else root / path + for directory in (root, root / "work_item"): + for filename in CONFIG_FILENAMES: + candidate = directory / filename + if candidate.is_file(): + return candidate + names = ", ".join(CONFIG_FILENAMES) + raise WorkItemError( + f"No workflow configuration found in {root} or {root / 'work_item'}; " + f"expected one of: {names}" + ) + + def parser() -> argparse.ArgumentParser: result = argparse.ArgumentParser(prog="work-item", description="Safe issue/Git workflow helper") - result.add_argument("--config", default=".work-item.toml") + result.add_argument( + "--config", + help="Configuration file; defaults to .work-item.toml/.yml/.yaml/.env discovery", + ) commands = result.add_subparsers(dest="command", required=True) commands.add_parser("next", help="Show the top eligible user story") commands.add_parser("claim", help="Claim the top story and create its branch") commands.add_parser("status", help="Show the current claim and Git state") submit = commands.add_parser("submit", help="Test, commit, push, and open a pull request") - submit.add_argument("--test", action="append", default=[], help="Additional automated test command; repeatable") - submit.add_argument("--yes", action="store_true", help="Confirm the reviewed diff may be committed") - complete = commands.add_parser("complete", help="Verify merge, close the issue, and return to main") - complete.add_argument("--merge", action="store_true", help="Merge the reviewed PR before completing") + submit.add_argument( + "--test", action="append", default=[], help="Additional automated test command; repeatable" + ) + submit.add_argument( + "--yes", action="store_true", help="Confirm the reviewed diff may be committed" + ) + complete = commands.add_parser( + "complete", help="Verify merge, close the issue, and return to main" + ) + complete.add_argument( + "--merge", action="store_true", help="Merge the reviewed PR before completing" + ) block = commands.add_parser("block", help="Mark the current story blocked") block.add_argument("--reason", required=True) return result @@ -542,9 +799,7 @@ def main(argv: Sequence[str] | None = None) -> int: args = parser().parse_args(argv) try: root = find_root(Path.cwd()) - config_path = Path(args.config) - if not config_path.is_absolute(): - config_path = root / config_path + config_path = find_config(root, args.config) workflow = Workflow(root, Config.load(config_path)) if args.command == "next": story = workflow.next() diff --git a/work_item/tests/test_cli_e2e.py b/work_item/tests/test_cli_e2e.py new file mode 100644 index 0000000..3f4c566 --- /dev/null +++ b/work_item/tests/test_cli_e2e.py @@ -0,0 +1,519 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +import textwrap +import unittest +from pathlib import Path + + +SOURCE_ROOT = Path(__file__).resolve().parents[1] +PACKAGE_ROOT = SOURCE_ROOT / "src" + + +def run(args, *, cwd: Path, env=None, check=True): + result = subprocess.run(args, cwd=cwd, env=env, text=True, capture_output=True) + if check and result.returncode: + raise AssertionError(f"{args}\nstdout={result.stdout}\nstderr={result.stderr}") + return result + + +FAKE_TEA = r"""#!/usr/bin/env python3 +import json, os, subprocess, sys +from pathlib import Path + +state_path = Path(os.environ["FAKE_TEA_STATE"]) +state = json.loads(state_path.read_text()) +args = sys.argv[1:] + +def save(): + state_path.write_text(json.dumps(state, indent=2, sort_keys=True) + "\n") + +def out(value): + print(json.dumps(value)) + +if args[0] == "api": + endpoint = args[1] + if endpoint.endswith("/issues?state=open&type=issues&limit=100"): + out([issue for issue in state["issues"] if issue["state"] == "open"]) + elif "/dependencies" in endpoint: + issue = int(endpoint.split("/issues/")[1].split("/")[0]) + blockers = state["dependencies"].get(str(issue), []) + out([next(x for x in state["issues"] if x["number"] == number) for number in blockers]) + elif endpoint.endswith("/pulls?state=open&limit=50"): + out([pr for pr in state["prs"].values() if not pr["merged"]]) + elif "/pulls/" in endpoint: + number = int(endpoint.rsplit("/", 1)[1]) + out(state["prs"][str(number)]) + elif "/commits/" in endpoint and endpoint.endswith("/status"): + out({"state": state.get("ci_state", "success")}) + else: + raise SystemExit(f"unsupported api: {endpoint}") +elif args[:2] == ["issues", "edit"]: + issue = next(x for x in state["issues"] if x["number"] == int(args[2])) + if "--add-assignees" in args: + issue["assignees"] = [{"login": args[args.index("--add-assignees") + 1]}] + labels = {x["name"] for x in issue.get("labels", [])} + if "--remove-labels" in args: + labels -= set(args[args.index("--remove-labels") + 1].split(",")) + if "--add-labels" in args: + labels |= set(args[args.index("--add-labels") + 1].split(",")) + issue["labels"] = [{"name": x} for x in sorted(labels)] + save(); out(issue) +elif args[:2] == ["comments", "add"]: + state["comments"].append({"issue": int(args[2]), "body": args[3]}) + save(); out(state["comments"][-1]) +elif args[:2] == ["pulls", "create"]: + number = len(state["prs"]) + 1 + branch = args[args.index("--head") + 1] + sha = subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip() + pr = {"number": number, "index": number, "url": f"https://example.test/pr/{number}", + "merged": False, "head": {"sha": sha, "ref": branch}} + state["prs"][str(number)] = pr + save(); out(pr) +elif args[:2] == ["pulls", "merge"]: + number = int(args[2]); pr = state["prs"][str(number)] + if state.get("merge_succeeds", True): + subprocess.check_call(["git", "push", "origin", f"{pr['head']['ref']}:main"]) + pr["merged"] = True + save(); out(pr) +elif args[:2] == ["issues", "close"]: + issue = next(x for x in state["issues"] if x["number"] == int(args[2])) + issue["state"] = "closed"; save(); out(issue) +else: + raise SystemExit("unsupported tea command: " + repr(args)) +""" + + +class CliEndToEndTests(unittest.TestCase): + def setUp(self) -> None: + self.temp = tempfile.TemporaryDirectory() + self.base = Path(self.temp.name) + remote_parent = self.base / "domverse" + remote_parent.mkdir() + seed = self.base / "seed" + seed.mkdir() + run(["git", "init", "-b", "main"], cwd=seed) + run(["git", "config", "user.name", "Test Agent"], cwd=seed) + run(["git", "config", "user.email", "agent@example.test"], cwd=seed) + (seed / "README.md").write_text("seed\n") + run(["git", "add", "README.md"], cwd=seed) + run(["git", "commit", "-m", "seed"], cwd=seed) + self.remote = remote_parent / "photoanalyzer.git" + run(["git", "clone", "--bare", str(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) + + config = textwrap.dedent("""\ + [repository] + slug = "domverse/photoanalyzer" + login = "fake" + assignee = "agent" + remote = "origin" + main_branch = "main" + [workflow] + branch_prefix = "us" + require_ci = true + required_tests = [] + [safety] + max_file_bytes = 100000 + allow = ["tests/fixtures/**"] + deny = ["_IGNORE/**", "**/_IGNORE/**", "pictures/**", "*.env", "*.jpg"] + """) + (self.repo / ".work-item.toml").write_text(config) + run(["git", "add", ".work-item.toml"], cwd=self.repo) + run(["git", "commit", "-m", "config"], cwd=self.repo) + run(["git", "push", "origin", "main"], cwd=self.repo) + + bin_dir = self.base / "bin" + bin_dir.mkdir() + tea = bin_dir / "tea" + tea.write_text(FAKE_TEA) + tea.chmod(0o755) + self.state_path = self.base / "tea-state.json" + self.state_path.write_text( + json.dumps( + { + "issues": [ + { + "number": 1, + "title": "US01-01 — Implement Safe Workflow", + "state": "open", + "labels": [{"name": "status/backlog"}], + "assignees": [], + } + ], + "dependencies": {"1": []}, + "comments": [], + "prs": {}, + } + ) + ) + self.env = os.environ.copy() + self.env["PATH"] = f"{bin_dir}{os.pathsep}{self.env['PATH']}" + self.env["FAKE_TEA_STATE"] = str(self.state_path) + self.env["PYTHONPATH"] = str(PACKAGE_ROOT) + + def tearDown(self) -> None: + self.temp.cleanup() + + def cli(self, *args, check=True): + return run( + [sys.executable, "-m", "work_item", *args], cwd=self.repo, env=self.env, check=check + ) + + def state(self): + return json.loads(self.state_path.read_text()) + + def save_state(self, state): + self.state_path.write_text(json.dumps(state)) + + def test_claim_submit_merge_complete_lifecycle(self) -> None: + next_result = self.cli("next") + self.assertEqual(json.loads(next_result.stdout)["story_id"], "US01-01") + + claim = json.loads(self.cli("claim").stdout) + self.assertEqual(claim["status"], "in-progress") + self.assertEqual( + run(["git", "branch", "--show-current"], cwd=self.repo).stdout.strip(), claim["branch"] + ) + issue = self.state()["issues"][0] + self.assertEqual(issue["assignees"], [{"login": "agent"}]) + self.assertIn({"name": "status/in-progress"}, issue["labels"]) + + (self.repo / "private.env").write_text("not-a-real-secret\n") + unsafe = self.cli("submit", "--test", "true", "--yes", check=False) + self.assertEqual(unsafe.returncode, 2) + self.assertIn("denied path", unsafe.stderr) + (self.repo / "private.env").unlink() + + (self.repo / "feature.txt").write_text("implemented\n") + review_only = self.cli("submit", "--test", "true", check=False) + self.assertEqual(review_only.returncode, 2) + self.assertIn("rerun with --yes", review_only.stderr) + self.assertEqual( + run(["git", "status", "--porcelain"], cwd=self.repo).stdout.strip(), "?? feature.txt" + ) + + submitted = json.loads(self.cli("submit", "--test", "true", "--yes").stdout) + self.assertEqual(submitted["status"], "review") + self.assertEqual(submitted["pr"], 1) + self.assertIn("status/review", {x["name"] for x in self.state()["issues"][0]["labels"]}) + + not_merged = self.cli("complete", check=False) + self.assertEqual(not_merged.returncode, 2) + self.assertIn("not merged", not_merged.stderr) + + completed = json.loads(self.cli("complete", "--merge").stdout) + self.assertEqual(completed["status"], "done") + self.assertEqual( + run(["git", "branch", "--show-current"], cwd=self.repo).stdout.strip(), "main" + ) + self.assertEqual(run(["git", "status", "--porcelain"], cwd=self.repo).stdout.strip(), "") + issue = self.state()["issues"][0] + self.assertEqual(issue["state"], "closed") + self.assertIn("status/done", {x["name"] for x in issue["labels"]}) + self.assertFalse((self.repo / ".git" / "work-item-state.json").exists()) + + def test_dirty_tree_prevents_claim(self) -> None: + (self.repo / "unexpected.txt").write_text("dirty\n") + result = self.cli("claim", check=False) + self.assertEqual(result.returncode, 2) + self.assertIn("clean", result.stderr) + self.assertEqual( + run(["git", "branch", "--show-current"], cwd=self.repo).stdout.strip(), "main" + ) + + def test_status_reports_changes_and_claim_resumes_dirty_work(self) -> None: + claimed = json.loads(self.cli("claim").stdout) + (self.repo / "work.txt").write_text("pending\n") + status = json.loads(self.cli("status").stdout) + self.assertEqual(status["branch"], claimed["branch"]) + self.assertEqual(status["current_branch"], claimed["branch"]) + self.assertEqual(status["changes"], ["work.txt"]) + resumed = json.loads(self.cli("claim").stdout) + self.assertEqual(resumed, claimed) + self.assertEqual((self.repo / "work.txt").read_text(), "pending\n") + + def test_claim_switches_back_to_claimed_branch_when_clean(self) -> None: + claimed = json.loads(self.cli("claim").stdout) + run(["git", "switch", "main"], cwd=self.repo) + resumed = json.loads(self.cli("claim").stdout) + self.assertEqual(resumed, claimed) + self.assertEqual( + run(["git", "branch", "--show-current"], cwd=self.repo).stdout.strip(), + claimed["branch"], + ) + + def test_next_selection_skips_ineligible_issues(self) -> None: + state = self.state() + state["issues"] = [ + {"number": 10, "title": "Maintenance", "state": "open", "labels": [], "assignees": []}, + { + "number": 1, + "title": "US01-01 — Busy", + "state": "open", + "labels": [{"name": "status/in-progress"}], + "assignees": [], + }, + { + "number": 2, + "title": "US01-02 — Assigned", + "state": "open", + "labels": [], + "assignees": [{"login": "someone"}], + }, + { + "number": 3, + "title": "US01-03 — Has dependency", + "state": "open", + "labels": [], + "assignees": [], + }, + { + "number": 4, + "title": "US01-04 — Ready", + "state": "open", + "labels": [{"name": "status/backlog"}], + "assignees": [], + }, + { + "number": 5, + "title": "US02-01 — Later epic", + "state": "open", + "labels": [], + "assignees": [], + }, + {"number": 99, "title": "Blocker", "state": "open", "labels": [], "assignees": []}, + ] + state["dependencies"] = {"3": [99], "4": [], "5": []} + self.save_state(state) + payload = json.loads(self.cli("next").stdout) + self.assertEqual(payload["story_id"], "US01-04") + + def test_next_fails_when_no_story_is_eligible(self) -> None: + state = self.state() + state["issues"][0]["labels"] = [{"name": "status/blocked"}] + self.save_state(state) + result = self.cli("next", check=False) + self.assertEqual(result.returncode, 2) + self.assertIn("No eligible", result.stderr) + + def test_commands_requiring_a_claim_fail_without_state(self) -> None: + cases = ( + ("status",), + ("submit", "--test", "true"), + ("complete",), + ("block", "--reason", "blocked"), + ) + for args in cases: + with self.subTest(command=args[0]): + result = self.cli(*args, check=False) + self.assertEqual(result.returncode, 2) + self.assertIn("No story is currently claimed", result.stderr) + + def test_submit_requires_changes_and_passing_tests(self) -> None: + self.cli("claim") + no_changes = self.cli("submit", "--test", "true", "--yes", check=False) + self.assertEqual(no_changes.returncode, 2) + self.assertIn("No changes", no_changes.stderr) + + (self.repo / "feature.txt").write_text("implemented\n") + failed_test = self.cli("submit", "--test", "false", "--yes", check=False) + self.assertEqual(failed_test.returncode, 2) + self.assertIn("Test command failed", failed_test.stderr) + self.assertEqual( + run(["git", "status", "--porcelain"], cwd=self.repo).stdout.strip(), "?? feature.txt" + ) + + def test_submit_rejects_wrong_branch(self) -> None: + claimed = json.loads(self.cli("claim").stdout) + run(["git", "switch", "main"], cwd=self.repo) + result = self.cli("submit", "--test", "true", check=False) + self.assertEqual(result.returncode, 2) + self.assertIn(f"Expected branch {claimed['branch']}", result.stderr) + + def test_complete_requires_review_and_successful_ci(self) -> None: + self.cli("claim") + before_review = self.cli("complete", "--merge", check=False) + self.assertEqual(before_review.returncode, 2) + self.assertIn("not been submitted", before_review.stderr) + + (self.repo / "feature.txt").write_text("implemented\n") + self.cli("submit", "--test", "true", "--yes") + state = self.state() + state["ci_state"] = "pending" + self.save_state(state) + pending = self.cli("complete", "--merge", check=False) + self.assertEqual(pending.returncode, 2) + self.assertIn("CI is not successful: pending", pending.stderr) + + def test_complete_detects_unverified_merge(self) -> None: + self.cli("claim") + (self.repo / "feature.txt").write_text("implemented\n") + self.cli("submit", "--test", "true", "--yes") + state = self.state() + state["merge_succeeds"] = False + self.save_state(state) + result = self.cli("complete", "--merge", check=False) + self.assertEqual(result.returncode, 2) + self.assertIn("merge could not be verified", result.stderr) + + def test_complete_rejects_new_dirty_work_after_submission(self) -> None: + self.cli("claim") + (self.repo / "feature.txt").write_text("implemented\n") + self.cli("submit", "--test", "true", "--yes") + (self.repo / "unexpected.txt").write_text("do not carry to main\n") + result = self.cli("complete", "--merge", check=False) + self.assertEqual(result.returncode, 2) + self.assertIn("Working tree must be clean", result.stderr) + self.assertTrue((self.repo / "unexpected.txt").is_file()) + + def test_whitespace_only_block_reason_is_rejected(self) -> None: + self.cli("claim") + result = self.cli("block", "--reason", " ", check=False) + self.assertEqual(result.returncode, 2) + self.assertIn("non-empty blocking reason", result.stderr) + + def test_blocked_story_can_be_resumed(self) -> None: + claimed = json.loads(self.cli("claim").stdout) + blocked = json.loads(self.cli("block", "--reason", "Waiting for a fixture").stdout) + self.assertEqual(blocked["status"], "blocked") + resumed = json.loads(self.cli("claim").stdout) + self.assertEqual(resumed["status"], "in-progress") + self.assertNotIn("reason", resumed) + issue = self.state()["issues"][0] + labels = {label["name"] for label in issue["labels"]} + self.assertIn("status/in-progress", labels) + self.assertNotIn("status/blocked", labels) + self.assertIn(claimed["branch"], self.state()["comments"][-1]["body"]) + + def test_review_story_cannot_be_reclaimed(self) -> None: + self.cli("claim") + (self.repo / "feature.txt").write_text("implemented\n") + self.cli("submit", "--test", "true", "--yes") + result = self.cli("claim", check=False) + self.assertEqual(result.returncode, 2) + self.assertIn("finish its review/completion", result.stderr) + + def test_command_works_from_repository_subdirectory(self) -> None: + subdirectory = self.repo / "nested" + subdirectory.mkdir() + result = run( + [sys.executable, "-m", "work_item", "next"], + cwd=subdirectory, + env=self.env, + ) + self.assertEqual(json.loads(result.stdout)["story_id"], "US01-01") + + def test_cli_discovers_yaml_configuration(self) -> None: + (self.repo / ".work-item.toml").unlink() + (self.repo / ".work-item.yml").write_text( + textwrap.dedent( + """\ + repository: + slug: domverse/photoanalyzer + login: fake + assignee: agent + remote: origin + main_branch: main + workflow: + branch_prefix: story + require_ci: false + required_tests: [] + safety: + max_file_bytes: 100000 + allow: ["tests/fixtures/**"] + deny: ["*.env", "*.jpg"] + """ + ), + encoding="utf-8", + ) + payload = json.loads(self.cli("next").stdout) + self.assertEqual(payload["story_id"], "US01-01") + + def test_cli_discovers_dotenv_and_honors_environment_override(self) -> None: + (self.repo / ".work-item.toml").unlink() + (self.repo / ".work-item.env").write_text( + textwrap.dedent( + """\ + WORK_ITEM_REPO_SLUG=wrong/repository + WORK_ITEM_LOGIN=fake + WORK_ITEM_ASSIGNEE=agent + WORK_ITEM_REQUIRED_TESTS=[] + WORK_ITEM_DENY=[] + WORK_ITEM_ALLOW=[] + """ + ), + encoding="utf-8", + ) + self.env["WORK_ITEM_REPO_SLUG"] = "domverse/photoanalyzer" + payload = json.loads(self.cli("next").stdout) + self.assertEqual(payload["story_id"], "US01-01") + + def test_block_requires_reason_and_records_state(self) -> None: + self.cli("claim") + result = self.cli("block", "--reason", "Provider fixture unavailable") + payload = json.loads(result.stdout) + self.assertEqual(payload["status"], "blocked") + issue = self.state()["issues"][0] + self.assertIn("status/blocked", {x["name"] for x in issue["labels"]}) + self.assertIn("Provider fixture unavailable", self.state()["comments"][-1]["body"]) + + +class InterpreterWrapperTests(unittest.TestCase): + def test_wrapper_selects_supported_python(self) -> None: + result = run( + [ + str(SOURCE_ROOT / "scripts" / "python"), + "-c", + "import sys; print(sys.version_info[:2])", + ], + cwd=SOURCE_ROOT, + ) + major, minor = eval(result.stdout.strip(), {"__builtins__": {}}) + self.assertEqual(major, 3) + self.assertGreaterEqual(minor, 12) + + def test_wrapper_falls_back_when_conda_python_is_too_old(self) -> None: + with tempfile.TemporaryDirectory() as directory: + bin_dir = Path(directory) + conda = bin_dir / "conda" + conda.write_text("#!/bin/sh\necho 3.11\n") + conda.chmod(0o755) + fallback = bin_dir / "python3.13" + fallback.symlink_to(sys.executable) + env = os.environ.copy() + env["PATH"] = f"{bin_dir}{os.pathsep}{env['PATH']}" + result = run( + [str(SOURCE_ROOT / "scripts" / "python"), "-c", "print('fallback')"], + cwd=SOURCE_ROOT, + env=env, + ) + self.assertEqual(result.stdout.strip(), "fallback") + + def test_wrapper_reports_when_no_supported_python_exists(self) -> None: + with tempfile.TemporaryDirectory() as directory: + bin_dir = Path(directory) + conda = bin_dir / "conda" + conda.write_text("#!/bin/sh\necho 3.11\n") + conda.chmod(0o755) + env = os.environ.copy() + env["PATH"] = f"{bin_dir}{os.pathsep}/bin:/usr/bin" + result = run( + [str(SOURCE_ROOT / "scripts" / "python"), "-c", "print('never')"], + cwd=SOURCE_ROOT, + env=env, + check=False, + ) + self.assertEqual(result.returncode, 2) + self.assertIn("Python 3.12 or newer is required", result.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/work_item/tests/test_core.py b/work_item/tests/test_core.py new file mode 100644 index 0000000..5dd2839 --- /dev/null +++ b/work_item/tests/test_core.py @@ -0,0 +1,526 @@ +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_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()