stackchain-dashboard/tests/test_release_engine.py

import json
from pathlib import Path

import pytest

from src.release_engine import (
    CommandResult,
    EngineConfig,
    Issue,
    ReleaseEngine,
    RunState,
    StateStore,
    branch_name,
    select_issue,
)


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()