fe794b8ffe
All checks were successful
Tests / Declarative: Post Actions passed: 1378
A run id in a pull request is a string to copy; a link is a page to open. The console reopens a finished run at /chat?resume=<id> - the prompt Hermes was given, the evidence bundle it read, the skill it invoked, and the JSON it returned, all on one page. That page is the whole answer to "who decided this". Making the reviewer reconstruct the URL from a bare id is the difference between evidence someone looks at and evidence someone takes on trust. run_url is shared rather than formatted at each call site, so the pull request and the demo monitor cannot drift into pointing at different pages. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
401 lines
16 KiB
Python
401 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
|
|
from ariadne.services import hermes_code_repair as module
|
|
from ariadne.services.hermes_code_patch import ProposedPatch
|
|
|
|
|
|
INCIDENT_ID = "hermes-code-demo/7"
|
|
BRANCH = "hermes-repair/7"
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, status_code: int, payload=None, text: str = "") -> None: # type: ignore[no-untyped-def]
|
|
self.status_code = status_code
|
|
self._payload = payload
|
|
self.text = text
|
|
|
|
def json(self): # type: ignore[no-untyped-def]
|
|
if self._payload is None:
|
|
raise ValueError("no json body")
|
|
return self._payload
|
|
|
|
|
|
def _install_http(monkeypatch, responses=None) -> dict: # type: ignore[no-untyped-def]
|
|
calls: dict = {"requests": [], "kwargs": None}
|
|
queue = list(responses or [])
|
|
|
|
class FakeClient:
|
|
def __init__(self, **kwargs) -> None: # type: ignore[no-untyped-def]
|
|
calls["kwargs"] = kwargs
|
|
|
|
def __enter__(self): # type: ignore[no-untyped-def]
|
|
return self
|
|
|
|
def __exit__(self, *args) -> None: # type: ignore[no-untyped-def]
|
|
return None
|
|
|
|
def _next(self, method, url, kwargs): # type: ignore[no-untyped-def]
|
|
calls["requests"].append((method, url, kwargs))
|
|
item = queue.pop(0)
|
|
if isinstance(item, Exception):
|
|
raise item
|
|
return item
|
|
|
|
def get(self, url, **kwargs): # type: ignore[no-untyped-def]
|
|
return self._next("GET", url, kwargs)
|
|
|
|
def put(self, url, **kwargs): # type: ignore[no-untyped-def]
|
|
return self._next("PUT", url, kwargs)
|
|
|
|
def post(self, url, **kwargs): # type: ignore[no-untyped-def]
|
|
return self._next("POST", url, kwargs)
|
|
|
|
monkeypatch.setattr(module.httpx, "Client", FakeClient)
|
|
return calls
|
|
|
|
|
|
def _cfg(**overrides) -> dict: # type: ignore[no-untyped-def]
|
|
base = {
|
|
"gitea_base_url": "https://scm.example",
|
|
"gitea_token": "secret-token",
|
|
"owner": "bstein",
|
|
"repo": "hermes-code-demo",
|
|
"base_branch": "master",
|
|
"timeout_seconds": 7.5,
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
def _patch(**overrides) -> ProposedPatch: # type: ignore[no-untyped-def]
|
|
values = {
|
|
"path": "src/discount.py",
|
|
"original": "return price * 0.5",
|
|
"replacement": "return price * 0.9",
|
|
"rationale": "restore the intended discount",
|
|
}
|
|
values.update(overrides)
|
|
return ProposedPatch(**values)
|
|
|
|
|
|
def _push(monkeypatch, responses): # type: ignore[no-untyped-def]
|
|
calls = _install_http(monkeypatch, responses)
|
|
result = module.push_branch(_cfg(), INCIDENT_ID, 7, _patch(), "patched contents\n")
|
|
return calls, result
|
|
|
|
|
|
def test_fetch_file_success(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [FakeResponse(200, text="file body")])
|
|
contents, error = module.fetch_file(_cfg(), "src/discount.py")
|
|
assert (contents, error) == ("file body", None)
|
|
method, url, kwargs = calls["requests"][0]
|
|
assert method == "GET"
|
|
assert url == "https://scm.example/api/v1/repos/bstein/hermes-code-demo/raw/src/discount.py"
|
|
assert kwargs["params"] == {"ref": "master"}
|
|
assert kwargs["headers"] == {"Authorization": "token secret-token"}
|
|
assert calls["kwargs"] == {"timeout": 7.5}
|
|
|
|
|
|
def test_fetch_file_http_error(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [FakeResponse(404)])
|
|
assert module.fetch_file(_cfg(), "src/discount.py") == (None, "file fetch http 404")
|
|
|
|
|
|
def test_fetch_file_request_exception(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [RuntimeError("connect refused")])
|
|
contents, error = module.fetch_file(_cfg(), "src/discount.py")
|
|
assert contents is None
|
|
assert error == "file fetch failed: connect refused"
|
|
|
|
|
|
def test_fetch_file_without_base_url(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [])
|
|
assert module.fetch_file(_cfg(gitea_base_url=""), "x") == (None, "gitea base url is empty")
|
|
assert calls["requests"] == []
|
|
|
|
|
|
def test_fetch_file_uses_default_timeout(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [FakeResponse(200, text="ok")])
|
|
module.fetch_file(_cfg(timeout_seconds="bad"), "src/discount.py")
|
|
assert calls["kwargs"] == {"timeout": 15.0}
|
|
|
|
|
|
def _pull(number: int, head: str = BRANCH, base: str = "master") -> dict:
|
|
return {
|
|
"number": number,
|
|
"html_url": f"https://scm.example/pulls/{number}",
|
|
"head": {"ref": head},
|
|
"base": {"ref": base},
|
|
}
|
|
|
|
|
|
def _none_found(error=None) -> dict: # type: ignore[no-untyped-def]
|
|
return {"found": False, "open_count": 0, "pr_number": None, "url": None, "branch": None, "error": error}
|
|
|
|
|
|
def test_find_open_proposal_matches_repair_branch(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [FakeResponse(200, [_pull(2)])])
|
|
result = module.find_open_proposal(_cfg())
|
|
assert result == {
|
|
"found": True,
|
|
"open_count": 1,
|
|
"pr_number": 2,
|
|
"url": "https://scm.example/pulls/2",
|
|
"branch": BRANCH,
|
|
"error": None,
|
|
}
|
|
method, url, kwargs = calls["requests"][0]
|
|
assert (method, url) == ("GET", "https://scm.example/api/v1/repos/bstein/hermes-code-demo/pulls")
|
|
assert kwargs["params"] == {"state": "open", "limit": 50}
|
|
assert kwargs["headers"] == {"Authorization": "token secret-token"}
|
|
|
|
|
|
def test_find_open_proposal_ignores_other_branch_prefixes(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [FakeResponse(200, [_pull(3, head="feature/thing"), _pull(4, head="renovate/x")])])
|
|
assert module.find_open_proposal(_cfg()) == _none_found()
|
|
|
|
|
|
def test_find_open_proposal_ignores_wrong_base_branch(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [FakeResponse(200, [_pull(3, base="develop")])])
|
|
assert module.find_open_proposal(_cfg()) == _none_found()
|
|
|
|
|
|
def test_find_open_proposal_returns_lowest_numbered_match(monkeypatch) -> None:
|
|
payload = [
|
|
_pull(9, head="hermes-repair/9"),
|
|
_pull(2, head="hermes-repair/4"),
|
|
_pull(1, head="other/1"),
|
|
_pull(5, head="hermes-repair/5"),
|
|
]
|
|
_install_http(monkeypatch, [FakeResponse(200, payload)])
|
|
result = module.find_open_proposal(_cfg())
|
|
assert (result["found"], result["pr_number"], result["branch"]) == (True, 2, "hermes-repair/4")
|
|
|
|
|
|
def test_find_open_proposal_skips_malformed_entries(monkeypatch) -> None:
|
|
payload = ["not a dict", {"number": 1}, {"number": True, "head": {"ref": BRANCH}, "base": {"ref": "master"}}]
|
|
_install_http(monkeypatch, [FakeResponse(200, payload)])
|
|
assert module.find_open_proposal(_cfg()) == _none_found()
|
|
|
|
|
|
def test_find_open_proposal_empty_list(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [FakeResponse(200, [])])
|
|
assert module.find_open_proposal(_cfg()) == _none_found()
|
|
|
|
|
|
def test_find_open_proposal_http_error_fails_open(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [FakeResponse(503)])
|
|
assert module.find_open_proposal(_cfg()) == _none_found("open proposal lookup http 503")
|
|
|
|
|
|
def test_find_open_proposal_malformed_json_fails_open(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [FakeResponse(200)])
|
|
result = module.find_open_proposal(_cfg())
|
|
assert result["found"] is False
|
|
assert result["error"] == "open proposal parse failed: no json body"
|
|
|
|
|
|
def test_find_open_proposal_non_list_payload_fails_open(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [FakeResponse(200, {"message": "nope"})])
|
|
assert module.find_open_proposal(_cfg()) == _none_found("open proposal payload is not a list")
|
|
|
|
|
|
def test_find_open_proposal_request_exception_fails_open(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [RuntimeError("connection reset")])
|
|
assert module.find_open_proposal(_cfg()) == _none_found("open proposal lookup failed: connection reset")
|
|
|
|
|
|
def test_find_open_proposal_without_base_url(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [])
|
|
assert module.find_open_proposal(_cfg(gitea_base_url="")) == _none_found("gitea base url is empty")
|
|
assert calls["requests"] == []
|
|
|
|
|
|
def test_push_branch_success_with_new_branch_payload(monkeypatch) -> None:
|
|
calls, result = _push(monkeypatch, [FakeResponse(200, {"sha": "abc123"}), FakeResponse(201, {})])
|
|
assert result == {"branch": BRANCH, "committed": True, "error": None}
|
|
get_method, get_url, get_kwargs = calls["requests"][0]
|
|
assert (get_method, get_kwargs["params"]) == ("GET", {"ref": "master"})
|
|
assert get_url == "https://scm.example/api/v1/repos/bstein/hermes-code-demo/contents/src/discount.py"
|
|
put_method, put_url, put_kwargs = calls["requests"][1]
|
|
assert (put_method, put_url) == ("PUT", get_url)
|
|
body = put_kwargs["json"]
|
|
assert body["branch"] == "master"
|
|
assert body["new_branch"] == BRANCH
|
|
assert body["sha"] == "abc123"
|
|
assert body["message"] == f"fix(hermes): restore the intended discount (incident {INCIDENT_ID})"
|
|
assert base64.b64decode(body["content"]).decode() == "patched contents\n"
|
|
assert body["author"] == {"name": "Hermes Agent", "email": "hermes@bstein.dev"}
|
|
assert body["committer"] == {"name": "Hermes Agent", "email": "hermes@bstein.dev"}
|
|
|
|
|
|
def test_push_branch_falls_back_to_branch_only_payload(monkeypatch) -> None:
|
|
calls, result = _push(
|
|
monkeypatch,
|
|
[FakeResponse(200, {"sha": "abc123"}), FakeResponse(422, {}), FakeResponse(201, {})],
|
|
)
|
|
assert result == {"branch": BRANCH, "committed": True, "error": None}
|
|
fallback_body = calls["requests"][2][2]["json"]
|
|
assert fallback_body["branch"] == BRANCH
|
|
assert "new_branch" not in fallback_body
|
|
|
|
|
|
def test_push_branch_reports_both_failed_payload_shapes(monkeypatch) -> None:
|
|
_, result = _push(
|
|
monkeypatch,
|
|
[FakeResponse(200, {"sha": "abc123"}), FakeResponse(404, {}), FakeResponse(422, {})],
|
|
)
|
|
assert result["committed"] is False
|
|
assert result["error"] == "commit http 404 then fallback http 422"
|
|
|
|
|
|
def test_push_branch_non_retryable_commit_error(monkeypatch) -> None:
|
|
calls, result = _push(monkeypatch, [FakeResponse(200, {"sha": "abc123"}), FakeResponse(500, {})])
|
|
assert result["error"] == "commit http 500"
|
|
assert len(calls["requests"]) == 2
|
|
|
|
|
|
def test_push_branch_sha_read_error(monkeypatch) -> None:
|
|
_, result = _push(monkeypatch, [FakeResponse(404)])
|
|
assert result == {"branch": BRANCH, "committed": False, "error": "file sha http 404"}
|
|
|
|
|
|
def test_push_branch_sha_missing(monkeypatch) -> None:
|
|
_, result = _push(monkeypatch, [FakeResponse(200, {})])
|
|
assert result["error"] == "file sha missing from contents response"
|
|
|
|
|
|
def test_push_branch_request_exception(monkeypatch) -> None:
|
|
_, result = _push(monkeypatch, [RuntimeError("boom")])
|
|
assert result == {"branch": BRANCH, "committed": False, "error": "branch push failed: boom"}
|
|
|
|
|
|
def test_push_branch_refuses_protected_branch_names(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [])
|
|
for name in ("master", "main", " "):
|
|
monkeypatch.setattr(module, "_branch_name", lambda build_number, name=name: name)
|
|
result = module.push_branch(_cfg(), INCIDENT_ID, 7, _patch(), "contents")
|
|
assert result["committed"] is False
|
|
assert result["error"] == f"refusing branch {name!r}"
|
|
assert calls["requests"] == []
|
|
|
|
|
|
def test_push_branch_without_base_url(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [])
|
|
result = module.push_branch(_cfg(gitea_base_url=""), INCIDENT_ID, 7, _patch(), "contents")
|
|
assert result == {"branch": BRANCH, "committed": False, "error": "gitea base url is empty"}
|
|
assert calls["requests"] == []
|
|
|
|
|
|
def test_open_pull_request_created(monkeypatch) -> None:
|
|
calls = _install_http(
|
|
monkeypatch,
|
|
[FakeResponse(201, {"number": 5, "html_url": "https://scm.example/pulls/5"})],
|
|
)
|
|
result = module.open_pull_request(_cfg(), INCIDENT_ID, "run_x", BRANCH, _patch(), "root cause analysis")
|
|
assert result == {"pr_number": 5, "url": "https://scm.example/pulls/5", "error": None}
|
|
method, url, kwargs = calls["requests"][0]
|
|
assert (method, url) == ("POST", "https://scm.example/api/v1/repos/bstein/hermes-code-demo/pulls")
|
|
payload = kwargs["json"]
|
|
assert payload["head"] == BRANCH
|
|
assert payload["base"] == "master"
|
|
assert payload["title"] == f"fix(hermes): repair {INCIDENT_ID}"
|
|
body = payload["body"]
|
|
assert INCIDENT_ID in body
|
|
assert "root cause analysis" in body
|
|
assert "restore the intended discount" in body
|
|
assert "`src/discount.py`" in body
|
|
assert "Proposed by Hermes; validated and pushed by Ariadne; requires human review — no automatic merge." in body
|
|
assert "secret-token" not in json.dumps(payload)
|
|
|
|
|
|
def test_open_pull_request_conflict_returns_existing(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [FakeResponse(409, {"number": 9, "html_url": "https://scm.example/pulls/9"})])
|
|
result = module.open_pull_request(_cfg(), INCIDENT_ID, "run_x", BRANCH, _patch(), "analysis")
|
|
assert result == {"pr_number": 9, "url": "https://scm.example/pulls/9", "error": None}
|
|
|
|
|
|
def test_open_pull_request_conflict_without_payload(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [FakeResponse(409)])
|
|
result = module.open_pull_request(_cfg(), INCIDENT_ID, "run_x", BRANCH, _patch(), "analysis")
|
|
assert result == {"pr_number": None, "url": None, "error": "pull request already exists"}
|
|
|
|
|
|
def test_open_pull_request_http_error(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [FakeResponse(500)])
|
|
result = module.open_pull_request(_cfg(), INCIDENT_ID, "run_x", BRANCH, _patch(), "analysis")
|
|
assert result == {"pr_number": None, "url": None, "error": "pull request http 500"}
|
|
|
|
|
|
def test_open_pull_request_created_with_bad_payload(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [FakeResponse(201, {"number": True})])
|
|
result = module.open_pull_request(_cfg(), INCIDENT_ID, "run_x", BRANCH, _patch(), "analysis")
|
|
assert result == {"pr_number": None, "url": None, "error": None}
|
|
|
|
|
|
def test_open_pull_request_request_exception(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [RuntimeError("down")])
|
|
result = module.open_pull_request(_cfg(), INCIDENT_ID, "run_x", BRANCH, _patch(), "analysis")
|
|
assert result == {"pr_number": None, "url": None, "error": "pull request failed: down"}
|
|
|
|
|
|
def test_open_pull_request_without_base_url(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [])
|
|
result = module.open_pull_request(_cfg(gitea_base_url=" "), INCIDENT_ID, 7, BRANCH, _patch(), "a")
|
|
assert result == {"pr_number": None, "url": None, "error": "gitea base url is empty"}
|
|
assert calls["requests"] == []
|
|
|
|
|
|
def test_the_pull_request_names_the_hermes_run(monkeypatch) -> None:
|
|
""""Proposed by Hermes" is a claim; the run id is the receipt."""
|
|
|
|
calls = _install_http(monkeypatch, [FakeResponse(201, {"number": 5, "html_url": "u"})])
|
|
module.open_pull_request(
|
|
{**_cfg(), "hermes_ui_url": "https://agent.bstein.dev/"},
|
|
INCIDENT_ID,
|
|
"run_a5af87af",
|
|
BRANCH,
|
|
_patch(),
|
|
"analysis",
|
|
)
|
|
|
|
body = calls["requests"][0][2]["json"]["body"]
|
|
assert (
|
|
"**Hermes run:** [run_a5af87af]"
|
|
"(https://agent.bstein.dev/chat?resume=run_a5af87af)"
|
|
) in body
|
|
assert "the tools it called" in body
|
|
|
|
|
|
def test_the_pull_request_omits_the_link_when_no_ui_is_configured(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [FakeResponse(201, {"number": 5, "html_url": "u"})])
|
|
module.open_pull_request(_cfg(), INCIDENT_ID, "run_x", BRANCH, _patch(), "analysis")
|
|
|
|
body = calls["requests"][0][2]["json"]["body"]
|
|
assert "**Hermes run:** `run_x`" in body
|
|
assert "http" not in body.split("**Hermes run:**")[1]
|
|
|
|
|
|
def test_the_pull_request_stays_readable_without_a_run_id(monkeypatch) -> None:
|
|
"""A proposal with no run is still worth opening; it just claims less."""
|
|
|
|
calls = _install_http(monkeypatch, [FakeResponse(201, {"number": 5, "html_url": "u"})])
|
|
module.open_pull_request(_cfg(), INCIDENT_ID, "", BRANCH, _patch(), "analysis")
|
|
|
|
body = calls["requests"][0][2]["json"]["body"]
|
|
assert "Hermes run" not in body
|
|
assert "requires human review" in body
|
|
|
|
|
|
def test_the_run_link_reopens_the_run_in_the_hermes_console() -> None:
|
|
"""One helper, so the pull request and the demo monitor cannot diverge."""
|
|
|
|
assert module.run_url("https://agent.bstein.dev/", "run_x") == (
|
|
"https://agent.bstein.dev/chat?resume=run_x"
|
|
)
|
|
assert module.run_url("", "run_x") == ""
|
|
assert module.run_url("https://agent.bstein.dev", " ") == ""
|