628 lines
20 KiB
Python
628 lines
20 KiB
Python
import json
|
|
import sys
|
|
import threading
|
|
import time
|
|
from dataclasses import replace
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from src.release_engine import (
|
|
CommandResult,
|
|
EngineConfig,
|
|
GiteaClient,
|
|
Issue,
|
|
ReleaseEngine,
|
|
RunState,
|
|
ShellRunner,
|
|
StateStore,
|
|
_parse_args,
|
|
branch_name,
|
|
select_issue,
|
|
)
|
|
|
|
|
|
def test_shell_runner_treats_metacharacters_as_literal_arguments(tmp_path):
|
|
output = tmp_path / "argument.txt"
|
|
injected = tmp_path / "injected.txt"
|
|
payload = f"release title; touch {injected}"
|
|
|
|
result = ShellRunner().run(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
"from pathlib import Path; import sys; Path(sys.argv[1]).write_text(sys.argv[2])",
|
|
str(output),
|
|
payload,
|
|
],
|
|
tmp_path,
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
assert output.read_text() == payload
|
|
assert not injected.exists()
|
|
|
|
|
|
def test_shell_runner_times_out_and_terminates_spawned_children(tmp_path):
|
|
marker = tmp_path / "child-survived.txt"
|
|
child = (
|
|
"import time; from pathlib import Path; "
|
|
f"time.sleep(0.3); Path({str(marker)!r}).write_text('survived')"
|
|
)
|
|
parent = (
|
|
"import subprocess, sys, time; "
|
|
f"subprocess.Popen([sys.executable, '-c', {child!r}]); "
|
|
"time.sleep(10)"
|
|
)
|
|
|
|
started = time.monotonic()
|
|
result = ShellRunner().run(
|
|
[sys.executable, "-c", parent],
|
|
tmp_path,
|
|
timeout_seconds=0.05,
|
|
termination_grace_seconds=0.05,
|
|
)
|
|
elapsed = time.monotonic() - started
|
|
time.sleep(0.35)
|
|
|
|
assert result.timed_out is True
|
|
assert elapsed < 0.5
|
|
assert not marker.exists()
|
|
|
|
|
|
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 = []
|
|
self.environments = []
|
|
self.options = []
|
|
|
|
def run(self, command, cwd, env=None, **options):
|
|
self.commands.append(command)
|
|
self.environments.append(env)
|
|
self.options.append(options)
|
|
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-from-env",
|
|
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_empty_agent_command_fails_before_claim(tmp_path):
|
|
issue = Issue(14, "Engine", "body", "open", [], None)
|
|
client = FakeClient([issue])
|
|
cfg = EngineConfig(
|
|
repo="stackchain/stackchain-dashboard",
|
|
agent="timmy",
|
|
repo_path=tmp_path,
|
|
state_path=tmp_path / "state.json",
|
|
agent_command=" ",
|
|
test_command="pytest -q",
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="invalid agent command: command is empty"):
|
|
ReleaseEngine(cfg, client, FakeRunner([])).execute(issue_number=14)
|
|
|
|
assert client.claimed == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value"),
|
|
[
|
|
("agent_timeout_seconds", 0),
|
|
("test_timeout_seconds", -1),
|
|
("termination_grace_seconds", 0),
|
|
],
|
|
)
|
|
def test_non_positive_command_deadlines_fail_before_claim(tmp_path, field, value):
|
|
issue = Issue(14, "Engine", "body", "open", [], None)
|
|
client = FakeClient([issue])
|
|
cfg = replace(config(tmp_path), **{field: value})
|
|
|
|
with pytest.raises(ValueError, match=field.replace("_", " ")):
|
|
ReleaseEngine(cfg, client, FakeRunner([])).execute(issue_number=14)
|
|
|
|
assert client.claimed == []
|
|
|
|
|
|
def test_command_deadlines_are_configurable_from_cli(monkeypatch):
|
|
monkeypatch.setattr(sys, "argv", [
|
|
"release-engine",
|
|
"--repo", "stackchain/stackchain-dashboard",
|
|
"--agent-timeout", "120",
|
|
"--test-timeout", "45.5",
|
|
"--termination-grace", "2",
|
|
])
|
|
|
|
args = _parse_args()
|
|
|
|
assert args.agent_timeout == 120
|
|
assert args.test_timeout == 45.5
|
|
assert args.termination_grace == 2
|
|
|
|
|
|
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-from-env"],
|
|
]
|
|
assert client.created_prs == []
|
|
|
|
|
|
def test_agent_timeout_records_terminal_state_and_blocks_tests(tmp_path):
|
|
issue = Issue(14, "Engine", "body", "open", [], None)
|
|
client = FakeClient([issue])
|
|
runner = FakeRunner([
|
|
CommandResult(0, "", ""),
|
|
CommandResult(-9, "partial output", "stalled", timed_out=True),
|
|
])
|
|
cfg = config(tmp_path)
|
|
|
|
with pytest.raises(RuntimeError, match="coding agent timed out"):
|
|
ReleaseEngine(cfg, client, runner).execute(issue_number=14)
|
|
|
|
state = StateStore(cfg.state_path).load()
|
|
assert state is not None
|
|
assert state.status == "agent_timed_out"
|
|
assert "partial output" in state.evidence
|
|
assert runner.commands == [
|
|
["git", "checkout", "-B", "timmy/14-engine"],
|
|
["agent", "--issue-from-env"],
|
|
]
|
|
assert runner.options[1] == {
|
|
"timeout_seconds": cfg.agent_timeout_seconds,
|
|
"termination_grace_seconds": cfg.termination_grace_seconds,
|
|
}
|
|
assert client.created_prs == []
|
|
|
|
|
|
def test_issue_content_reaches_agent_only_through_environment(tmp_path):
|
|
title = "$(touch /tmp/title-injected); `id`"
|
|
body = "line one\n&& touch /tmp/body-injected > stolen"
|
|
issue = Issue(14, title, body, "open", [], None)
|
|
client = FakeClient([issue])
|
|
runner = FakeRunner([
|
|
CommandResult(0, "", ""),
|
|
CommandResult(1, "", "agent stopped"),
|
|
])
|
|
cfg = EngineConfig(
|
|
repo="stackchain/stackchain-dashboard",
|
|
agent="timmy",
|
|
repo_path=tmp_path,
|
|
state_path=tmp_path / "state.json",
|
|
agent_command="agent --work",
|
|
test_command="pytest -q",
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="coding agent failed"):
|
|
ReleaseEngine(cfg, client, runner).execute(issue_number=14)
|
|
|
|
assert runner.commands == [
|
|
["git", "checkout", "-B", "timmy/14-touch-tmp-title-injected-id"],
|
|
["agent", "--work"],
|
|
]
|
|
assert runner.environments[1]["RELEASE_ISSUE_TITLE"] == title
|
|
assert runner.environments[1]["RELEASE_ISSUE_BODY"] == body
|
|
|
|
|
|
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(0, "abc123\n", ""),
|
|
CommandResult(1, "1 failed", ""),
|
|
])
|
|
engine = ReleaseEngine(config(tmp_path), client, runner)
|
|
|
|
with pytest.raises(RuntimeError, match="tests failed"):
|
|
engine.execute(issue_number=14)
|
|
|
|
assert not any(command[:2] == ["git", "push"] for command in runner.commands)
|
|
assert client.created_prs == []
|
|
|
|
|
|
def test_test_timeout_records_terminal_state_and_blocks_push(tmp_path):
|
|
issue = Issue(14, "Engine", "body", "open", [], None)
|
|
client = FakeClient([issue])
|
|
runner = FakeRunner([
|
|
CommandResult(0, "", ""),
|
|
CommandResult(0, "implemented", ""),
|
|
CommandResult(0, "abc123\n", ""),
|
|
CommandResult(-15, "3 passed before stall", "", timed_out=True),
|
|
])
|
|
cfg = config(tmp_path)
|
|
|
|
with pytest.raises(RuntimeError, match="tests timed out"):
|
|
ReleaseEngine(cfg, client, runner).execute(issue_number=14)
|
|
|
|
state = StateStore(cfg.state_path).load()
|
|
assert state is not None
|
|
assert state.status == "tests_timed_out"
|
|
assert "3 passed before stall" in state.evidence
|
|
assert runner.options[3] == {
|
|
"timeout_seconds": cfg.test_timeout_seconds,
|
|
"termination_grace_seconds": cfg.termination_grace_seconds,
|
|
}
|
|
assert not any(command[:2] == ["git", "push"] for command in 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, "abc123\n", ""),
|
|
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] == ["git", "push", "-u", "origin", "timmy/14-engine-mvp"]
|
|
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()
|
|
|
|
|
|
def test_restart_from_agent_complete_skips_claim_and_coding(tmp_path):
|
|
issue = Issue(331, "Resume releases", "body", "open", [], "timmy")
|
|
client = FakeClient([issue])
|
|
cfg = config(tmp_path)
|
|
branch = "timmy/331-resume-releases"
|
|
cfg.state_path.write_text(json.dumps({
|
|
"schema_version": 1,
|
|
"repo": cfg.repo,
|
|
"issue": issue.number,
|
|
"branch": branch,
|
|
"status": "agent_complete",
|
|
"evidence": "",
|
|
"pr_number": None,
|
|
"pr_url": "",
|
|
"head_sha": "abc123",
|
|
}))
|
|
runner = FakeRunner([
|
|
CommandResult(0, "", ""),
|
|
CommandResult(0, "abc123\n", ""),
|
|
CommandResult(0, "", ""),
|
|
CommandResult(0, "8 passed", ""),
|
|
CommandResult(0, "", ""),
|
|
])
|
|
|
|
result = ReleaseEngine(cfg, client, runner).execute(issue_number=331)
|
|
|
|
assert result["tests"] == "passed"
|
|
assert client.claimed == []
|
|
assert ["agent", "--issue-from-env"] not in runner.commands
|
|
assert runner.commands[:3] == [
|
|
["git", "checkout", branch],
|
|
["git", "rev-parse", "HEAD"],
|
|
["git", "status", "--porcelain"],
|
|
]
|
|
|
|
|
|
def test_restart_from_tests_passed_skips_claim_coding_and_tests(tmp_path):
|
|
issue = Issue(331, "Resume releases", "body", "open", [], "timmy")
|
|
client = FakeClient([issue])
|
|
cfg = config(tmp_path)
|
|
branch = "timmy/331-resume-releases"
|
|
StateStore(cfg.state_path).save(RunState(
|
|
issue=issue.number,
|
|
branch=branch,
|
|
status="tests_passed",
|
|
evidence="604 passed",
|
|
repo=cfg.repo,
|
|
head_sha="abc123",
|
|
))
|
|
runner = FakeRunner([
|
|
CommandResult(0, "", ""),
|
|
CommandResult(0, "abc123\n", ""),
|
|
CommandResult(0, "", ""),
|
|
CommandResult(0, "", ""),
|
|
])
|
|
|
|
result = ReleaseEngine(cfg, client, runner).execute(issue_number=331)
|
|
|
|
assert result["tests"] == "passed"
|
|
assert client.claimed == []
|
|
assert ["agent", "--issue-from-env"] not in runner.commands
|
|
assert ["pytest", "-q"] not in runner.commands
|
|
assert runner.commands == [
|
|
["git", "checkout", branch],
|
|
["git", "rev-parse", "HEAD"],
|
|
["git", "status", "--porcelain"],
|
|
["git", "push", "-u", "origin", branch],
|
|
]
|
|
|
|
|
|
def test_restart_from_pushed_verifies_remote_and_only_opens_pr(tmp_path):
|
|
issue = Issue(331, "Resume releases", "body", "open", [], "timmy")
|
|
client = FakeClient([issue])
|
|
cfg = config(tmp_path)
|
|
branch = "timmy/331-resume-releases"
|
|
StateStore(cfg.state_path).save(RunState(
|
|
issue=issue.number,
|
|
branch=branch,
|
|
status="pushed",
|
|
evidence="604 passed",
|
|
repo=cfg.repo,
|
|
head_sha="abc123",
|
|
))
|
|
runner = FakeRunner([
|
|
CommandResult(0, "", ""),
|
|
CommandResult(0, "abc123\n", ""),
|
|
CommandResult(0, "", ""),
|
|
CommandResult(0, f"abc123\trefs/heads/{branch}\n", ""),
|
|
])
|
|
|
|
result = ReleaseEngine(cfg, client, runner).execute(issue_number=331)
|
|
|
|
assert result["pr_number"] == 44
|
|
assert client.claimed == []
|
|
assert len(client.created_prs) == 1
|
|
assert runner.commands == [
|
|
["git", "checkout", branch],
|
|
["git", "rev-parse", "HEAD"],
|
|
["git", "status", "--porcelain"],
|
|
["git", "ls-remote", "--heads", "origin", branch],
|
|
]
|
|
|
|
|
|
def test_fresh_run_persists_repository_and_commit_identity(tmp_path):
|
|
issue = Issue(331, "Resume releases", "body", "open", [], None)
|
|
client = FakeClient([issue])
|
|
cfg = config(tmp_path)
|
|
runner = FakeRunner([
|
|
CommandResult(0, "", ""),
|
|
CommandResult(0, "implemented", ""),
|
|
CommandResult(0, "abc123\n", ""),
|
|
CommandResult(0, "604 passed", ""),
|
|
CommandResult(0, "", ""),
|
|
])
|
|
|
|
ReleaseEngine(cfg, client, runner).execute(issue_number=331)
|
|
|
|
state = StateStore(cfg.state_path).load()
|
|
assert state is not None
|
|
assert state.status == "pr_opened"
|
|
assert state.schema_version == 1
|
|
assert state.repo == cfg.repo
|
|
assert state.head_sha == "abc123"
|
|
assert ["git", "rev-parse", "HEAD"] in runner.commands
|
|
|
|
|
|
def test_resume_fails_closed_when_checkpoint_worktree_is_dirty(tmp_path):
|
|
issue = Issue(331, "Resume releases", "body", "open", [], "timmy")
|
|
client = FakeClient([issue])
|
|
cfg = config(tmp_path)
|
|
branch = "timmy/331-resume-releases"
|
|
StateStore(cfg.state_path).save(RunState(
|
|
issue=issue.number,
|
|
branch=branch,
|
|
status="tests_passed",
|
|
evidence="604 passed",
|
|
repo=cfg.repo,
|
|
head_sha="abc123",
|
|
))
|
|
runner = FakeRunner([
|
|
CommandResult(0, "", ""),
|
|
CommandResult(0, "abc123\n", ""),
|
|
CommandResult(0, " M src/release_engine.py\n", ""),
|
|
CommandResult(0, "", ""),
|
|
])
|
|
|
|
with pytest.raises(RuntimeError, match="worktree is not clean"):
|
|
ReleaseEngine(cfg, client, runner).execute(issue_number=331)
|
|
|
|
assert not any(command[:2] == ["git", "push"] for command in runner.commands)
|
|
assert client.created_prs == []
|
|
|
|
|
|
def test_resume_fails_closed_when_checkpoint_targets_another_repository(tmp_path):
|
|
issue = Issue(331, "Resume releases", "body", "open", [], "timmy")
|
|
client = FakeClient([issue])
|
|
cfg = config(tmp_path)
|
|
StateStore(cfg.state_path).save(RunState(
|
|
issue=issue.number,
|
|
branch="timmy/331-resume-releases",
|
|
status="tests_passed",
|
|
repo="other/repository",
|
|
head_sha="abc123",
|
|
))
|
|
|
|
with pytest.raises(RuntimeError, match="checkpoint does not match"):
|
|
ReleaseEngine(cfg, client, FakeRunner([])).execute(issue_number=331)
|
|
|
|
assert client.created_prs == []
|
|
|
|
|
|
def test_resume_fails_closed_when_local_commit_changed(tmp_path):
|
|
issue = Issue(331, "Resume releases", "body", "open", [], "timmy")
|
|
client = FakeClient([issue])
|
|
cfg = config(tmp_path)
|
|
StateStore(cfg.state_path).save(RunState(
|
|
issue=issue.number,
|
|
branch="timmy/331-resume-releases",
|
|
status="tests_passed",
|
|
repo=cfg.repo,
|
|
head_sha="abc123",
|
|
))
|
|
runner = FakeRunner([
|
|
CommandResult(0, "", ""),
|
|
CommandResult(0, "different-sha\n", ""),
|
|
])
|
|
|
|
with pytest.raises(RuntimeError, match="commit does not match the local branch"):
|
|
ReleaseEngine(cfg, client, runner).execute(issue_number=331)
|
|
|
|
assert client.created_prs == []
|
|
|
|
|
|
def test_resume_fails_closed_when_remote_commit_changed(tmp_path):
|
|
issue = Issue(331, "Resume releases", "body", "open", [], "timmy")
|
|
client = FakeClient([issue])
|
|
cfg = config(tmp_path)
|
|
branch = "timmy/331-resume-releases"
|
|
StateStore(cfg.state_path).save(RunState(
|
|
issue=issue.number,
|
|
branch=branch,
|
|
status="pushed",
|
|
evidence="604 passed",
|
|
repo=cfg.repo,
|
|
head_sha="abc123",
|
|
))
|
|
runner = FakeRunner([
|
|
CommandResult(0, "", ""),
|
|
CommandResult(0, "abc123\n", ""),
|
|
CommandResult(0, "", ""),
|
|
CommandResult(0, f"different-sha\trefs/heads/{branch}\n", ""),
|
|
])
|
|
|
|
with pytest.raises(RuntimeError, match="commit does not match the remote branch"):
|
|
ReleaseEngine(cfg, client, runner).execute(issue_number=331)
|
|
|
|
assert client.created_prs == []
|