d3a2c94f80
All checks were successful
Tests / Declarative: Post Actions passed: 1205
Two changes to how this reads and behaves on real service repositories. The fixture rules were stated to Hermes on every job, so it reasoned about them out loud and that reasoning was published verbatim into service issue trackers - ariadne/404 opened with 'The job is ariadne, not hermes-triage-demo, so the reserved demo fixture classification and repair action are forbidden'. That reads as though the system exists to serve a demonstration. Those rules are now appended only for the fixture job, so a real service is never told about them and cannot repeat them; the demo classification is unreachable elsewhere by construction rather than by instruction. The prompt also asks for language aimed at a maintainer who knows nothing about how triage is configured, and points at the structured test evidence first now that junit publishes it. The duplicate guard refused a proposal whenever any repair pull request was open, which meant one unreviewed fix blocked every later one across the repository. It now enforces a ceiling instead, ARIADNE_HERMES_CODE_MAX_OPEN_PROPOSALS, default 64. That is a review-capacity limit, not a correctness one: proposals are cheap to make and expensive to read. Auto-triage settings move to their own module; they had grown a section's worth and pushed settings_sections.py past its size budget. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
350 lines
14 KiB
Python
350 lines
14 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, 7, 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, 7, 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, 7, 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, 7, 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, 7, 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, 7, 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"] == []
|