136caa8477
All checks were successful
Tests / Declarative: Post Actions passed: 1200
Log evidence covered the demo namespace plus jenkins. That is right for the common case, since CI failures happen in Jenkins agent pods, but it means a build failure that correlates with the service itself being unhealthy carries no trace of the service at all. ARIADNE_HERMES_JOB_NAMESPACES maps a job to its own namespace, which is added alongside jenkins rather than replacing it. Unmapped jobs are unchanged, and a namespace already in the list is never duplicated. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
326 lines
12 KiB
Python
326 lines
12 KiB
Python
"""Shared fakes for the Hermes auto-triage orchestrator tests.
|
|
|
|
Not collected by pytest; imported by test_hermes_autotriage.py and
|
|
test_hermes_autotriage_retry.py so both drive the same fake Jenkins, storage,
|
|
and remediation executors.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
from ariadne.services import hermes_autotriage as module
|
|
from ariadne.services.hermes_agent_client import HermesRunResult
|
|
|
|
|
|
JOB = "hermes-triage-demo"
|
|
INCIDENT_ID = f"{JOB}/12"
|
|
INFRA_MARKER = "temporary failure in name resolution"
|
|
|
|
|
|
class FakeStorage:
|
|
def __init__(self) -> None:
|
|
self.events: list[dict] = []
|
|
|
|
def record_event(self, event_type, detail) -> None: # type: ignore[no-untyped-def]
|
|
self.events.append({"event_type": event_type, "detail": detail})
|
|
|
|
def list_events(self, limit=200, event_type=None): # type: ignore[no-untyped-def]
|
|
rows = [
|
|
dict(row)
|
|
for row in reversed(self.events)
|
|
if event_type is None or row["event_type"] == event_type
|
|
]
|
|
return rows[:limit]
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, payload) -> None: # type: ignore[no-untyped-def]
|
|
self.payload = payload
|
|
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
def json(self): # type: ignore[no-untyped-def]
|
|
return self.payload
|
|
|
|
|
|
def _settings(**overrides) -> SimpleNamespace: # type: ignore[no-untyped-def]
|
|
values = {
|
|
"hermes_autotriage_enabled": True,
|
|
"hermes_autotriage_job_allowlist": [JOB],
|
|
"hermes_autoremediation_enabled": True,
|
|
"hermes_allowed_actions": ["repair_demo_fixture"],
|
|
"hermes_action_classifications": {
|
|
"known_demo_fixture_failure": "repair_demo_fixture",
|
|
"transient_infra_failure": "retry_transient_infra",
|
|
},
|
|
"hermes_parameterized_jobs": [JOB],
|
|
"hermes_min_confidence": 0.85,
|
|
"hermes_hung_build_minutes": 45.0,
|
|
"hermes_max_branches": 5,
|
|
"hermes_job_namespaces": {},
|
|
"hermes_max_actions_per_incident": 1,
|
|
"hermes_api_url": "http://hermes:8642",
|
|
"hermes_api_key": "key",
|
|
"hermes_run_timeout_seconds": 420.0,
|
|
"hermes_demo_namespace": "hermes-triage-demo",
|
|
"hermes_demo_fixture_configmap": "hermes-triage-demo-fixture",
|
|
"hermes_code_enabled": False,
|
|
"hermes_code_job": "hermes-code-demo",
|
|
"hermes_code_owner": "bstein",
|
|
"hermes_code_repo": "hermes-code-demo",
|
|
"hermes_code_base_branch": "master",
|
|
"hermes_code_candidate_path": "src/discount.py",
|
|
"hermes_code_allowed_prefixes": ["src/"],
|
|
"hermes_code_allowed_suffixes": [".py"],
|
|
"hermes_code_max_patch_bytes": 4000,
|
|
"hermes_code_max_changed_lines": 20,
|
|
"hermes_gitea_base_url": "https://scm.example",
|
|
"hermes_gitea_token": "gitea-token",
|
|
"hermes_issues_enabled": False,
|
|
"hermes_issue_repos": {},
|
|
"hermes_issue_dedupe_scope": "classification",
|
|
"hermes_issue_max_per_tick": 2,
|
|
"jenkins_base_url": "https://ci.example",
|
|
"jenkins_api_user": "user",
|
|
"jenkins_api_token": "token",
|
|
"jenkins_api_timeout_sec": 5.0,
|
|
}
|
|
values.update(overrides)
|
|
return SimpleNamespace(**values)
|
|
|
|
|
|
def _build(number: int, result, job: str = JOB, **overrides): # type: ignore[no-untyped-def]
|
|
payload = {
|
|
"number": number,
|
|
"result": result,
|
|
"building": False,
|
|
"timestamp": 1720000000000,
|
|
"duration": 60000,
|
|
"url": f"https://ci.example/job/{job}/{number}/",
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
def _model_output(**overrides) -> str: # type: ignore[no-untyped-def]
|
|
payload = {
|
|
"incident_id": INCIDENT_ID,
|
|
"classification": "known_demo_fixture_failure",
|
|
"confidence": 0.95,
|
|
"facts": [
|
|
{"statement": "fixture-state-check failed", "source": "jenkins", "reference": "console"}
|
|
],
|
|
"inferences": [],
|
|
"first_failed_gate": "fixture-state-check",
|
|
"requested_action": {"type": "run_ariadne_job", "id": "repair_demo_fixture"},
|
|
"human_required": False,
|
|
"reason": "known fixture failure",
|
|
}
|
|
payload.update(overrides)
|
|
return json.dumps(payload)
|
|
|
|
|
|
def _retry_output(job: str = JOB, build_number: int = 12, **overrides) -> str: # type: ignore[no-untyped-def]
|
|
payload = {
|
|
"incident_id": f"{job}/{build_number}",
|
|
"classification": "transient_infra_failure",
|
|
"confidence": 0.95,
|
|
"facts": [
|
|
{"statement": "pip install hit a DNS failure", "source": "jenkins", "reference": "console"}
|
|
],
|
|
"inferences": [],
|
|
"first_failed_gate": "dependencies",
|
|
"requested_action": {"type": "run_ariadne_job", "id": "retry_transient_infra"},
|
|
"human_required": False,
|
|
"reason": "transient name resolution failure on the agent",
|
|
}
|
|
payload.update(overrides)
|
|
return json.dumps(payload)
|
|
|
|
|
|
def _run(status: str = "completed", output=None, error=None) -> HermesRunResult: # type: ignore[no-untyped-def]
|
|
return HermesRunResult(
|
|
status=status,
|
|
output=output,
|
|
run_id="run-1",
|
|
session_id="sess-1",
|
|
error=error,
|
|
duration_seconds=1.5,
|
|
denied_approvals=0,
|
|
)
|
|
|
|
|
|
def _install_jenkins(monkeypatch, calls, last_build, exc) -> None: # type: ignore[no-untyped-def]
|
|
class FakeClient:
|
|
def __init__(self, **kwargs) -> None: # type: ignore[no-untyped-def]
|
|
calls["client_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 get(self, url, params=None): # type: ignore[no-untyped-def]
|
|
calls["gets"].append((url, params))
|
|
if exc is not None:
|
|
raise exc
|
|
return FakeResponse({"lastBuild": last_build})
|
|
|
|
monkeypatch.setattr(module.hermes_jenkins_client.httpx, "Client", FakeClient)
|
|
|
|
|
|
def _prepare( # type: ignore[no-untyped-def] # noqa: PLR0913
|
|
monkeypatch,
|
|
*,
|
|
cfg=None,
|
|
last_build=None,
|
|
jenkins_exc=None,
|
|
run=None,
|
|
repair=None,
|
|
rebuild=None,
|
|
retry=None,
|
|
signature=True,
|
|
infra=None,
|
|
storage=None,
|
|
):
|
|
storage = storage if storage is not None else FakeStorage()
|
|
calls: dict = {"gets": [], "triage": [], "repairs": [], "rebuilds": [], "retries": []}
|
|
resolved = cfg if cfg is not None else _settings()
|
|
monkeypatch.setattr(module, "settings", resolved)
|
|
# The Jenkins transport holds its own settings reference, so patching
|
|
# only the orchestrator leaves it reading the real configuration.
|
|
monkeypatch.setattr(module.hermes_jenkins_client, "settings", resolved)
|
|
_install_jenkins(
|
|
monkeypatch, calls, last_build if last_build is not None else _build(12, "FAILURE"), jenkins_exc
|
|
)
|
|
monkeypatch.setattr(
|
|
module.hermes_evidence,
|
|
"collect_evidence",
|
|
lambda incident_id, job, build: {
|
|
"incident_id": incident_id,
|
|
"jenkins": {"job": job},
|
|
"log_evidence": {"records": []},
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
module.hermes_evidence, "evidence_has_signature", lambda bundle, incident_id: signature
|
|
)
|
|
monkeypatch.setattr(
|
|
module.hermes_infra_signals,
|
|
"has_transient_infra_signature",
|
|
lambda bundle: infra if infra is not None else (True, INFRA_MARKER),
|
|
)
|
|
|
|
def fake_run_triage(config, prompt): # type: ignore[no-untyped-def]
|
|
calls["triage"].append((config, prompt))
|
|
return run if run is not None else _run(output=_model_output())
|
|
|
|
def fake_execute_repair(repair_cfg, incident_id, build_number): # type: ignore[no-untyped-def]
|
|
calls["repairs"].append((repair_cfg, incident_id, build_number))
|
|
if repair is not None:
|
|
return repair
|
|
return {
|
|
"action": "configmap_patch",
|
|
"target": "hermes-triage-demo/hermes-triage-demo-fixture",
|
|
"succeeded": True,
|
|
"error": None,
|
|
}
|
|
|
|
def fake_trigger_rebuild(rebuild_cfg, job): # type: ignore[no-untyped-def]
|
|
calls["rebuilds"].append(job)
|
|
return rebuild if rebuild is not None else {"requested": True, "error": None}
|
|
|
|
def fake_retry_build(retry_cfg, job, parameterized): # type: ignore[no-untyped-def]
|
|
calls["retries"].append((job, parameterized))
|
|
return retry if retry is not None else {"requested": True, "error": None}
|
|
|
|
monkeypatch.setattr(module.hermes_agent_client, "run_triage", fake_run_triage)
|
|
monkeypatch.setattr(module.hermes_autotriage_repair, "execute_repair", fake_execute_repair)
|
|
monkeypatch.setattr(module.hermes_autotriage_repair, "trigger_rebuild", fake_trigger_rebuild)
|
|
monkeypatch.setattr(module.hermes_autotriage_repair, "retry_build", fake_retry_build)
|
|
return SimpleNamespace(storage=storage, calls=calls)
|
|
|
|
|
|
def _code_settings(**overrides) -> SimpleNamespace: # type: ignore[no-untyped-def]
|
|
values = {
|
|
"hermes_code_enabled": True,
|
|
"hermes_code_repos": f"{JOB}=bstein/{JOB}",
|
|
"hermes_issues_enabled": True,
|
|
"hermes_issue_repos": {JOB: ("bstein", JOB)},
|
|
}
|
|
values.update(overrides)
|
|
return _settings(**values)
|
|
|
|
|
|
def _pr_opened(**overrides) -> dict: # type: ignore[no-untyped-def]
|
|
payload = {
|
|
"status": "pr_opened",
|
|
"branch": "hermes-repair/12",
|
|
"pr_number": 8,
|
|
"url": "https://scm.example/pulls/8",
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
def _prepare_code(monkeypatch, *, result=None, error=None, cfg=None, **kwargs): # type: ignore[no-untyped-def]
|
|
env = _prepare(monkeypatch, cfg=cfg if cfg is not None else _code_settings(), **kwargs)
|
|
env.calls.update({"proposals": [], "gitea": [], "issues": []})
|
|
|
|
def fake_propose(storage, incident_id, job, build_number, bundle, hermes_cfg, code_cfg): # type: ignore[no-untyped-def] # noqa: PLR0913
|
|
env.calls["proposals"].append((incident_id, job, build_number, bundle, code_cfg))
|
|
if error is not None:
|
|
raise error
|
|
return result if result is not None else _pr_opened()
|
|
|
|
def fake_lookup(lookup_cfg): # type: ignore[no-untyped-def]
|
|
env.calls["gitea"].append(lookup_cfg)
|
|
return {"found": False, "error": None}
|
|
|
|
def fake_create(issue_cfg, context): # type: ignore[no-untyped-def]
|
|
env.calls["issues"].append(context)
|
|
return {"issue_number": 7, "url": "https://scm.example/issues/7", "error": None}
|
|
|
|
monkeypatch.setattr(module.hermes_code_flow, "propose_code_fix", fake_propose)
|
|
monkeypatch.setattr(module.hermes_code_flow.hermes_code_repair, "find_open_proposal", fake_lookup)
|
|
monkeypatch.setattr(
|
|
module.hermes_incident_issue,
|
|
"find_open_incident_issue",
|
|
lambda *args: {"found": False, "issue_number": None, "url": None, "error": None},
|
|
)
|
|
monkeypatch.setattr(module.hermes_incident_issue, "create_incident_issue", fake_create)
|
|
return env
|
|
|
|
|
|
def _events(storage: FakeStorage, event_type: str) -> list[dict]:
|
|
return [row["detail"] for row in storage.events if row["event_type"] == event_type]
|
|
|
|
|
|
def _statuses(storage: FakeStorage) -> list[str]:
|
|
return [detail["status"] for detail in _events(storage, module.INCIDENT_EVENT_TYPE)]
|
|
|
|
|
|
def _seed_incident( # type: ignore[no-untyped-def]
|
|
storage: FakeStorage, status: str, build_number: int = 12, as_json: bool = False, job: str = JOB
|
|
) -> None:
|
|
detail = {
|
|
"incident_id": f"{job}/{build_number}",
|
|
"job": job,
|
|
"build_number": build_number,
|
|
"status": status,
|
|
"phase": {},
|
|
}
|
|
storage.record_event(module.INCIDENT_EVENT_TYPE, json.dumps(detail) if as_json else detail)
|
|
|
|
|
|
def _gauge(build: str, status: str, job: str = JOB) -> float:
|
|
return module.HERMES_TRIAGE_INCIDENT.labels(jenkins_job=job, build=build, status=status)._value.get()
|
|
|
|
|
|
def _counter(action: str, result: str) -> float:
|
|
return module.HERMES_TRIAGE_ACTION_TOTAL.labels(action=action, result=result)._value.get()
|