950d014707
The code path was gated on a single job id, so homegrown services could never produce a pull request. Escalated incidents on mapped repositories now additionally attempt a bounded patch proposal, and the filed issue links it. - hermes_code_flow.propose_for_incident: additive entry point invoked only from the escalation branch, so an auto-remediated failure never also gets a patch and a failed Hermes run never spends tokens on one - three gates before any HTTP call: code enabled, not the legacy demo job, and the job has a repo mapping; unmapped jobs make zero calls - never raises: a failed proposal cannot change the incident outcome or break the tick - issue body links the proposal when a pull request was opened - propose_code_fix added to the bounded action-label set The legacy demo-job short-circuit is untouched and all of its tests pass unchanged. 8 new tests; 480 pass in the hermes suite. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
496 lines
18 KiB
Python
496 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
from ariadne.services import hermes_incident_body as body
|
|
from ariadne.services import hermes_incident_issue as module
|
|
|
|
|
|
JOB = "metis"
|
|
INCIDENT_ID = f"{JOB}/12"
|
|
CLASSIFICATION = "dependency_resolution_failure"
|
|
TOKEN = "super-secret-token"
|
|
PROPOSAL_URL = "https://scm.example/pulls/8"
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, status_code: int, payload=None) -> None: # type: ignore[no-untyped-def]
|
|
self.status_code = status_code
|
|
self._payload = payload
|
|
|
|
def json(self): # type: ignore[no-untyped-def]
|
|
if isinstance(self._payload, Exception):
|
|
raise self._payload
|
|
return self._payload
|
|
|
|
|
|
class FakeStorage:
|
|
def __init__(self, explode: bool = False) -> None:
|
|
self.events: list[tuple[str, dict]] = []
|
|
self.explode = explode
|
|
|
|
def record_event(self, event_type, detail) -> None: # type: ignore[no-untyped-def]
|
|
if self.explode:
|
|
raise RuntimeError("storage is down")
|
|
self.events.append((event_type, detail))
|
|
|
|
|
|
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 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]
|
|
values = {
|
|
"gitea_base_url": "https://scm.example",
|
|
"gitea_token": TOKEN,
|
|
"owner": "bstein",
|
|
"repo": JOB,
|
|
"timeout_seconds": 7.5,
|
|
}
|
|
values.update(overrides)
|
|
return values
|
|
|
|
|
|
def _context(**overrides) -> dict: # type: ignore[no-untyped-def]
|
|
values = {
|
|
"incident_id": INCIDENT_ID,
|
|
"job": JOB,
|
|
"build_number": 12,
|
|
"build_url": "https://ci.example/job/metis/12/",
|
|
"classification": CLASSIFICATION,
|
|
"confidence": 0.91,
|
|
"first_failed_gate": "dependencies",
|
|
"reason": "no allowlisted action fits this failure",
|
|
"facts": [{"statement": "pip could not resolve urllib3", "source": "jenkins", "reference": "console"}],
|
|
"inferences": ["an upstream index published a broken constraint"],
|
|
"authorize_reason": "classification_not_actionable",
|
|
"run_id": "run-9",
|
|
}
|
|
values.update(overrides)
|
|
return values
|
|
|
|
|
|
def _issue(number: int, job: str = JOB, classification: str = CLASSIFICATION, incident: str = INCIDENT_ID) -> dict:
|
|
return {
|
|
"number": number,
|
|
"html_url": f"https://scm.example/bstein/{job}/issues/{number}",
|
|
"body": f"diagnosis\n\n{body.issue_marker(job, classification, incident)}",
|
|
}
|
|
|
|
|
|
def _settings(**overrides) -> SimpleNamespace: # type: ignore[no-untyped-def]
|
|
values = {
|
|
"hermes_issues_enabled": True,
|
|
"hermes_issue_repos": {JOB: ("bstein", JOB)},
|
|
"hermes_issue_dedupe_scope": "classification",
|
|
"hermes_issue_max_per_tick": 2,
|
|
"hermes_gitea_base_url": "https://scm.example",
|
|
"hermes_gitea_token": TOKEN,
|
|
}
|
|
values.update(overrides)
|
|
return SimpleNamespace(**values)
|
|
|
|
|
|
def _base(build_number: int = 12) -> dict:
|
|
return {"incident_id": f"{JOB}/{build_number}", "job": JOB, "build_number": build_number}
|
|
|
|
|
|
def _diagnosis(**overrides) -> dict: # type: ignore[no-untyped-def]
|
|
decision = SimpleNamespace(
|
|
classification=CLASSIFICATION,
|
|
confidence=0.91,
|
|
first_failed_gate="dependencies",
|
|
reason="the failure has no known signature",
|
|
facts=[SimpleNamespace(statement="pip failed", source="jenkins", reference="console")],
|
|
inferences=["upstream index broke"],
|
|
)
|
|
values = {
|
|
"bundle": {"jenkins": {"url": "https://ci.example/job/metis/12/"}},
|
|
"outcome": SimpleNamespace(decision=decision),
|
|
"authorize_reason": "classification_not_actionable",
|
|
"run_id": "run-9",
|
|
}
|
|
values.update(overrides)
|
|
return values
|
|
|
|
|
|
def test_marker_round_trips_through_the_body() -> None:
|
|
marker = body.issue_marker(JOB, CLASSIFICATION, INCIDENT_ID)
|
|
|
|
assert body.parse_issue_marker(f"text\n{marker}\nmore") == {
|
|
"job": JOB,
|
|
"classification": CLASSIFICATION,
|
|
"incident": INCIDENT_ID,
|
|
}
|
|
|
|
|
|
def test_marker_flattens_values_that_would_break_the_comment() -> None:
|
|
marker = body.issue_marker("job\nname", "a --> b", INCIDENT_ID)
|
|
|
|
assert "\n" not in marker
|
|
assert body.parse_issue_marker(marker) == {
|
|
"job": "job name",
|
|
"classification": "a b",
|
|
"incident": INCIDENT_ID,
|
|
}
|
|
|
|
|
|
def test_marker_parsing_tolerates_missing_and_non_string_bodies() -> None:
|
|
assert body.parse_issue_marker(None) is None
|
|
assert body.parse_issue_marker(12) is None
|
|
assert body.parse_issue_marker("a plain human issue") is None
|
|
|
|
|
|
def test_classification_scope_matches_a_different_incident_of_the_same_job(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [FakeResponse(200, [_issue(7, incident=f"{JOB}/9")])])
|
|
|
|
found = module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID)
|
|
|
|
assert found == {
|
|
"found": True,
|
|
"issue_number": 7,
|
|
"url": "https://scm.example/bstein/metis/issues/7",
|
|
"error": None,
|
|
}
|
|
method, url, kwargs = calls["requests"][0]
|
|
assert (method, url) == ("GET", "https://scm.example/api/v1/repos/bstein/metis/issues")
|
|
assert kwargs["params"] == {"state": "open", "limit": 50}
|
|
assert kwargs["headers"] == {"Authorization": f"token {TOKEN}"}
|
|
|
|
|
|
def test_classification_scope_ignores_other_jobs_and_classifications(monkeypatch) -> None:
|
|
_install_http(
|
|
monkeypatch,
|
|
[FakeResponse(200, [_issue(7, job="lesavka"), _issue(8, classification="flaky_test")])],
|
|
)
|
|
|
|
found = module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID)
|
|
|
|
assert found["found"] is False
|
|
assert found["error"] is None
|
|
|
|
|
|
def test_incident_scope_matches_only_the_same_incident(monkeypatch) -> None:
|
|
_install_http(
|
|
monkeypatch,
|
|
[FakeResponse(200, [_issue(7, incident=f"{JOB}/9"), _issue(9, incident=INCIDENT_ID)])],
|
|
)
|
|
|
|
found = module.find_open_incident_issue(
|
|
_cfg(dedupe_scope="incident"), JOB, CLASSIFICATION, INCIDENT_ID
|
|
)
|
|
|
|
assert found["found"] is True
|
|
assert found["issue_number"] == 9
|
|
|
|
|
|
def test_lookup_returns_the_lowest_numbered_match(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [FakeResponse(200, [_issue(31), _issue(12), _issue(20)])])
|
|
|
|
found = module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID)
|
|
|
|
assert found["issue_number"] == 12
|
|
|
|
|
|
def test_lookup_ignores_unmarked_and_malformed_issues(monkeypatch) -> None:
|
|
payload = [
|
|
{"number": 3, "body": "a human filed this"},
|
|
{"number": True, "body": _issue(4)["body"]},
|
|
{"number": "5", "body": _issue(5)["body"]},
|
|
"not an issue",
|
|
]
|
|
_install_http(monkeypatch, [FakeResponse(200, payload)])
|
|
|
|
assert module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID)["found"] is False
|
|
|
|
|
|
def test_lookup_fails_open_on_http_parse_and_transport_errors(monkeypatch) -> None:
|
|
_install_http(
|
|
monkeypatch,
|
|
[
|
|
FakeResponse(503, None),
|
|
FakeResponse(200, ValueError("no json")),
|
|
FakeResponse(200, {"issues": []}),
|
|
RuntimeError("connection reset"),
|
|
],
|
|
)
|
|
|
|
statuses = [
|
|
module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID) for _ in range(4)
|
|
]
|
|
|
|
assert [result["found"] for result in statuses] == [False, False, False, False]
|
|
assert statuses[0]["error"] == "open issue lookup http 503"
|
|
assert "parse failed" in statuses[1]["error"]
|
|
assert statuses[2]["error"] == "open issue payload is not a list"
|
|
assert "connection reset" in statuses[3]["error"]
|
|
|
|
|
|
def test_lookup_without_a_base_url_never_calls_gitea(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [])
|
|
|
|
found = module.find_open_incident_issue(_cfg(gitea_base_url=" "), JOB, CLASSIFICATION, INCIDENT_ID)
|
|
|
|
assert found == {"found": False, "issue_number": None, "url": None, "error": "gitea base url is empty"}
|
|
assert calls["requests"] == []
|
|
|
|
|
|
def test_create_posts_the_issue_and_returns_its_identity(monkeypatch) -> None:
|
|
payload = {"number": 41, "html_url": "https://scm.example/bstein/metis/issues/41"}
|
|
calls = _install_http(monkeypatch, [FakeResponse(201, payload)])
|
|
|
|
created = module.create_incident_issue(_cfg(), _context())
|
|
|
|
assert created == {"issue_number": 41, "url": payload["html_url"], "error": None}
|
|
method, url, kwargs = calls["requests"][0]
|
|
assert (method, url) == ("POST", "https://scm.example/api/v1/repos/bstein/metis/issues")
|
|
assert kwargs["headers"] == {"Authorization": f"token {TOKEN}"}
|
|
assert kwargs["json"]["title"] == f"[hermes] {JOB} #12: {CLASSIFICATION}"
|
|
|
|
|
|
def test_create_reports_http_errors_exceptions_and_missing_numbers(monkeypatch) -> None:
|
|
_install_http(
|
|
monkeypatch,
|
|
[FakeResponse(422, None), RuntimeError("tls handshake failed"), FakeResponse(201, {"html_url": "u"})],
|
|
)
|
|
|
|
results = [module.create_incident_issue(_cfg(), _context()) for _ in range(3)]
|
|
|
|
assert results[0] == {"issue_number": None, "url": None, "error": "issue create http 422"}
|
|
assert "tls handshake failed" in results[1]["error"]
|
|
assert results[2] == {"issue_number": None, "url": "u", "error": "issue create response had no number"}
|
|
|
|
|
|
def test_create_without_a_base_url_never_calls_gitea(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [])
|
|
|
|
created = module.create_incident_issue(_cfg(gitea_base_url=""), _context())
|
|
|
|
assert created["error"] == "gitea base url is empty"
|
|
assert calls["requests"] == []
|
|
|
|
|
|
def test_title_truncates_a_long_classification() -> None:
|
|
title = body.issue_title(_context(classification="x" * 400))
|
|
|
|
assert len(title) == 120
|
|
assert title.startswith(f"[hermes] {JOB} #12: ")
|
|
assert title.endswith("...")
|
|
|
|
|
|
def test_body_carries_the_marker_the_build_url_and_the_no_write_footer() -> None:
|
|
rendered = body.issue_body(_context())
|
|
|
|
assert rendered.rstrip().endswith(body.issue_marker(JOB, CLASSIFICATION, INCIDENT_ID))
|
|
assert "https://ci.example/job/metis/12/" in rendered
|
|
assert "Hermes has no write access to this repository" in rendered
|
|
assert "no files or infrastructure were changed" in rendered
|
|
assert "run `run-9`" in rendered
|
|
assert "/api/admin/audit/events" in rendered
|
|
assert "## Why a human is needed" in rendered
|
|
assert "classification_not_actionable" in rendered
|
|
assert "- **jenkins** — pip could not resolve urllib3 (`console`)" in rendered
|
|
|
|
|
|
def test_body_never_contains_the_gitea_token() -> None:
|
|
rendered = body.issue_body(_context(reason=f"failure while using {TOKEN[:4]}"))
|
|
|
|
assert TOKEN not in rendered
|
|
|
|
|
|
def test_body_caps_facts_inferences_and_statement_length() -> None:
|
|
rendered = body.issue_body(
|
|
_context(
|
|
facts=[{"statement": "s" * 500, "source": "jenkins", "reference": f"r{i}"} for i in range(20)],
|
|
inferences=[f"inference {i}" for i in range(12)],
|
|
)
|
|
)
|
|
|
|
assert rendered.count("- **jenkins**") == 10
|
|
assert rendered.count("- inference ") == 6
|
|
assert "s" * 500 not in rendered
|
|
assert "s" * 297 + "..." in rendered
|
|
|
|
|
|
def test_body_truncates_to_the_configured_cap_and_keeps_the_marker() -> None:
|
|
rendered = body.issue_body(_context(reason="r" * 20000), max_chars=1200)
|
|
|
|
assert len(rendered) <= 1200
|
|
assert "Truncated by Ariadne" in rendered
|
|
assert rendered.rstrip().endswith(body.issue_marker(JOB, CLASSIFICATION, INCIDENT_ID))
|
|
|
|
|
|
def test_a_proposed_fix_flows_from_the_diagnosis_into_the_issue_body() -> None:
|
|
proposal = {"branch": "hermes-repair/12", "pr_number": 8, "url": PROPOSAL_URL}
|
|
context = module.issue_context(_base(), _diagnosis(code_proposal=proposal))
|
|
assert context["code_proposal_url"] == PROPOSAL_URL
|
|
assert f"- Proposed fix awaiting review: {PROPOSAL_URL}" in body.issue_body(context)
|
|
assert module.issue_context(_base(), _diagnosis())["code_proposal_url"] == ""
|
|
assert module.issue_context(_base(), _diagnosis(code_proposal={}))["code_proposal_url"] == ""
|
|
assert "Proposed fix awaiting review" not in body.issue_body(_context())
|
|
|
|
|
|
def test_context_defaults_to_undiagnosed_when_no_decision_was_parsed() -> None:
|
|
context = module.issue_context(_base(), _diagnosis(outcome=None, run_id=None))
|
|
|
|
assert context["classification"] == "undiagnosed"
|
|
assert context["reason"] == "classification_not_actionable"
|
|
assert context["facts"] == []
|
|
assert context["build_url"] == "https://ci.example/job/metis/12/"
|
|
|
|
|
|
def test_maybe_file_issue_files_and_records_one_event(monkeypatch) -> None:
|
|
payload = {"number": 5, "html_url": "https://scm.example/bstein/metis/issues/5"}
|
|
calls = _install_http(monkeypatch, [FakeResponse(200, []), FakeResponse(201, payload)])
|
|
storage, tick = FakeStorage(), {}
|
|
|
|
detail = module.maybe_file_issue(storage, _settings(), _base(), _diagnosis(), tick)
|
|
|
|
assert detail == {
|
|
"incident_id": INCIDENT_ID,
|
|
"job": JOB,
|
|
"build_number": 12,
|
|
"classification": CLASSIFICATION,
|
|
"issue_number": 5,
|
|
"url": payload["html_url"],
|
|
"skipped": False,
|
|
"error": None,
|
|
}
|
|
assert storage.events == [(module.ISSUE_EVENT_TYPE, detail)]
|
|
assert tick == {"issues_filed": 1}
|
|
assert [call[0] for call in calls["requests"]] == ["GET", "POST"]
|
|
|
|
|
|
def test_maybe_file_issue_skips_when_an_open_issue_already_exists(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [FakeResponse(200, [_issue(7, incident=f"{JOB}/9")])])
|
|
storage, tick = FakeStorage(), {}
|
|
|
|
detail = module.maybe_file_issue(storage, _settings(), _base(), _diagnosis(), tick)
|
|
|
|
assert detail["skipped"] is True
|
|
assert detail["issue_number"] == 7
|
|
assert detail["error"] is None
|
|
assert [call[0] for call in calls["requests"]] == ["GET"]
|
|
assert tick == {}
|
|
|
|
|
|
def test_disabled_flag_files_nothing_and_calls_nothing(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [])
|
|
storage = FakeStorage()
|
|
|
|
assert module.maybe_file_issue(storage, _settings(hermes_issues_enabled=False), _base(), _diagnosis(), {}) is None
|
|
assert calls["requests"] == []
|
|
assert storage.events == []
|
|
|
|
|
|
def test_unmapped_job_files_nothing_and_calls_nothing(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [])
|
|
storage = FakeStorage()
|
|
|
|
result = module.maybe_file_issue(
|
|
storage, _settings(hermes_issue_repos={"other-job": ("bstein", "other")}), _base(), _diagnosis(), {}
|
|
)
|
|
|
|
assert result is None
|
|
assert calls["requests"] == []
|
|
assert storage.events == []
|
|
|
|
|
|
def test_max_per_tick_stops_a_burst_of_failures(monkeypatch) -> None:
|
|
payload = {"number": 5, "html_url": "u"}
|
|
calls = _install_http(
|
|
monkeypatch,
|
|
[FakeResponse(200, []), FakeResponse(201, payload), FakeResponse(200, []), FakeResponse(201, payload)],
|
|
)
|
|
storage, tick = FakeStorage(), {}
|
|
config = _settings(hermes_issue_max_per_tick=2)
|
|
|
|
results = [
|
|
module.maybe_file_issue(storage, config, _base(build), _diagnosis(), tick) for build in (12, 13, 14)
|
|
]
|
|
|
|
assert [result is None for result in results] == [False, False, True]
|
|
assert tick == {"issues_filed": 2}
|
|
assert len(calls["requests"]) == 4
|
|
|
|
|
|
def test_lookup_failure_fails_open_and_still_files(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [FakeResponse(500, None), FakeResponse(201, {"number": 6})])
|
|
storage, tick = FakeStorage(), {}
|
|
|
|
detail = module.maybe_file_issue(storage, _settings(), _base(), _diagnosis(), tick)
|
|
|
|
assert detail["skipped"] is False
|
|
assert detail["issue_number"] == 6
|
|
assert [call[0] for call in calls["requests"]] == ["GET", "POST"]
|
|
|
|
|
|
def test_create_failure_is_recorded_on_the_event(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [FakeResponse(200, []), FakeResponse(403, None)])
|
|
storage, tick = FakeStorage(), {}
|
|
|
|
detail = module.maybe_file_issue(storage, _settings(), _base(), _diagnosis(), tick)
|
|
|
|
assert detail["issue_number"] is None
|
|
assert detail["error"] == "issue create http 403"
|
|
assert detail["skipped"] is False
|
|
assert tick == {"issues_filed": 1}
|
|
|
|
|
|
def test_maybe_file_issue_never_raises(monkeypatch) -> None:
|
|
_install_http(monkeypatch, [FakeResponse(200, []), FakeResponse(201, {"number": 5})])
|
|
|
|
assert module.maybe_file_issue(FakeStorage(explode=True), _settings(), _base(), _diagnosis(), {}) is None
|
|
|
|
|
|
def test_repo_map_entries_accept_pairs_dicts_and_strings(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [FakeResponse(200, []), FakeResponse(201, {"number": 5})] * 2)
|
|
for repos in ({JOB: {"owner": "bstein", "repo": "metis"}}, {JOB: "bstein/metis"}):
|
|
module.maybe_file_issue(
|
|
FakeStorage(), _settings(hermes_issue_repos=repos), _base(), _diagnosis(), {}
|
|
)
|
|
|
|
assert [call[1] for call in calls["requests"]] == [
|
|
"https://scm.example/api/v1/repos/bstein/metis/issues"
|
|
] * 4
|
|
|
|
|
|
def test_broken_repo_map_entries_file_nothing(monkeypatch) -> None:
|
|
calls = _install_http(monkeypatch, [])
|
|
|
|
for repos in ({JOB: "bstein"}, {JOB: ("bstein", "")}, {JOB: 12}, {JOB: None}):
|
|
assert (
|
|
module.maybe_file_issue(
|
|
FakeStorage(), _settings(hermes_issue_repos=repos), _base(), _diagnosis(), {}
|
|
)
|
|
is None
|
|
)
|
|
assert calls["requests"] == []
|