215 lines
6.6 KiB
Python
215 lines
6.6 KiB
Python
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()
|