import json import threading from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path import pytest from src.release_engine import ( CommandResult, EngineConfig, GiteaClient, Issue, ReleaseEngine, RunState, StateStore, branch_name, select_issue, ) def test_stale_token_error_explains_how_to_restore_cron_authentication(): class UnauthorizedHandler(BaseHTTPRequestHandler): def do_GET(self): self.send_response(401) self.end_headers() self.wfile.write(b'{"message":"user does not exist [uid: 0, name: ]"}') def log_message(self, format, *args): pass server = HTTPServer(("127.0.0.1", 0), UnauthorizedHandler) thread = threading.Thread(target=server.handle_request) thread.start() try: client = GiteaClient(f"http://127.0.0.1:{server.server_port}", "stale-token") with pytest.raises(RuntimeError, match="refresh the cron GITEA_TOKEN"): client.list_open_issues("stackchain/stackchain-dashboard") finally: thread.join(timeout=2) server.server_close() class FakeClient: def __init__(self, issues, verified_assignee="timmy", existing_pr=None): self.issues = issues self.verified_assignee = verified_assignee self.existing_pr = existing_pr self.claimed = [] self.created_prs = [] def list_open_issues(self, repo): return self.issues def claim_issue(self, repo, number, assignee): self.claimed.append((repo, number, assignee)) def get_issue(self, repo, number): issue = next(i for i in self.issues if i.number == number) return Issue(**{**issue.__dict__, "assignee": self.verified_assignee}) def find_open_pr(self, repo, head): return self.existing_pr def create_pr(self, repo, head, base, title, body): self.created_prs.append((repo, head, base, title, body)) return {"number": 44, "html_url": "https://forge/pr/44"} class FakeRunner: def __init__(self, results): self.results = list(results) self.commands = [] def run(self, command, cwd, env=None): self.commands.append(command) return self.results.pop(0) @pytest.fixture def issues(): return [ Issue(2, "P2 task", "body", "open", ["P2"], None), Issue(3, "Owned P0", "body", "open", ["P0"], "other"), Issue(4, "Unowned P0", "body", "open", ["P0"], None), ] def config(tmp_path): return EngineConfig( repo="stackchain/stackchain-dashboard", agent="timmy", repo_path=tmp_path, state_path=tmp_path / "state.json", agent_command="agent --issue {issue_number}", test_command="pytest -q", ) def test_selection_prioritizes_p0_and_skips_other_assignees(issues): assert select_issue(issues, agent="timmy").number == 4 def test_explicit_issue_must_be_eligible(issues): with pytest.raises(ValueError, match="assigned to other"): select_issue(issues, agent="timmy", issue_number=3) def test_branch_name_is_reproducible_and_bounded(): assert branch_name("Timmy", Issue(14, "Autonomous Issue-to-Release Engine!", "", "open", [], None)) == "timmy/14-autonomous-issue-to-release-engine" def test_claim_verification_blocks_execution(tmp_path): issue = Issue(14, "Engine", "body", "open", [], None) client = FakeClient([issue], verified_assignee="other") runner = FakeRunner([]) engine = ReleaseEngine(config(tmp_path), client, runner) with pytest.raises(RuntimeError, match="claim verification failed"): engine.execute(issue_number=14) assert runner.commands == [] def test_agent_failure_blocks_tests_and_pr(tmp_path): issue = Issue(14, "Engine", "body", "open", [], None) client = FakeClient([issue]) runner = FakeRunner([ CommandResult(0, "", ""), CommandResult(1, "", "agent failed"), ]) engine = ReleaseEngine(config(tmp_path), client, runner) with pytest.raises(RuntimeError, match="coding agent failed"): engine.execute(issue_number=14) assert runner.commands == [ "git checkout -B timmy/14-engine", "agent --issue 14", ] assert client.created_prs == [] def test_failed_tests_block_push_and_pr(tmp_path): issue = Issue(14, "Engine", "body", "open", [], None) client = FakeClient([issue]) runner = FakeRunner([ CommandResult(0, "", ""), CommandResult(0, "changed", ""), CommandResult(1, "1 failed", ""), ]) engine = ReleaseEngine(config(tmp_path), client, runner) with pytest.raises(RuntimeError, match="tests failed"): engine.execute(issue_number=14) assert "git push" not in " ".join(runner.commands) assert client.created_prs == [] def test_passing_run_pushes_and_opens_linked_pr(tmp_path): issue = Issue(14, "Engine MVP", "body", "open", ["P1"], None) client = FakeClient([issue]) runner = FakeRunner([ CommandResult(0, "", ""), CommandResult(0, "implemented", ""), CommandResult(0, "8 passed", ""), CommandResult(0, "", ""), ]) engine = ReleaseEngine(config(tmp_path), client, runner) result = engine.execute(issue_number=14) assert result["pr_number"] == 44 assert runner.commands[-1].startswith("git push -u origin") assert len(client.created_prs) == 1 body = client.created_prs[0][4] assert "Closes #14" in body assert "pytest -q" in body assert "8 passed" in body def test_dry_run_does_not_claim_or_execute(tmp_path): issue = Issue(14, "Engine MVP", "body", "open", [], None) client = FakeClient([issue]) runner = FakeRunner([]) engine = ReleaseEngine(config(tmp_path), client, runner) result = engine.execute(issue_number=14, dry_run=True) assert result["dry_run"] is True assert result["issue"] == 14 assert client.claimed == [] assert runner.commands == [] def test_existing_pr_is_reused_without_agent_execution(tmp_path): issue = Issue(14, "Engine MVP", "body", "open", [], "timmy") client = FakeClient([issue], existing_pr={"number": 9, "html_url": "https://forge/pr/9"}) runner = FakeRunner([]) engine = ReleaseEngine(config(tmp_path), client, runner) result = engine.execute(issue_number=14) assert result["pr_number"] == 9 assert runner.commands == [] assert client.created_prs == [] def test_state_store_round_trip_is_atomic(tmp_path): store = StateStore(tmp_path / "state.json") expected = RunState(issue=14, branch="timmy/14-engine", status="tests_passed", evidence="8 passed") store.save(expected) assert store.load() == expected assert not (tmp_path / "state.json.tmp").exists()