chore: make work-item a reusable subproject
This commit is contained in:
519
work_item/tests/test_cli_e2e.py
Normal file
519
work_item/tests/test_cli_e2e.py
Normal file
@@ -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()
|
||||
526
work_item/tests/test_core.py
Normal file
526
work_item/tests/test_core.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user