2cb5d50fa8
Hermes proposes a minimal anchored patch; Ariadne validates it structurally and opens a pull request for human review. Nothing merges automatically and Hermes never holds Git credentials or executes anything. - hermes_code_patch: response parsing + patch validation (path prefix/suffix allowlist, size and changed-line caps, exact-single-occurrence anchor, replacement-differs) and exact application - hermes_code_repair: Gitea contents-API client (fetch, branch push with hermes-repair/<build> prefix and base-branch refusal, PR creation) - hermes_code_flow: proposal orchestration + audit event, no patch bodies or tokens recorded - hermes_autotriage: code-path branch for the configured repo job; a PR records human_required/code_fix_proposed, never auto-resolution, and is kept out of the fixture-action accounting - settings: ARIADNE_HERMES_CODE_* configuration, disabled by default 74 new tests; 214 pass in the hermes suite. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
492 lines
19 KiB
Python
492 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
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"
|
|
|
|
|
|
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_min_confidence": 0.85,
|
|
"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_repair_image": "busybox:1.37",
|
|
"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",
|
|
"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, **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 _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.httpx, "Client", FakeClient)
|
|
|
|
|
|
def _prepare( # type: ignore[no-untyped-def]
|
|
monkeypatch,
|
|
*,
|
|
cfg=None,
|
|
last_build=None,
|
|
jenkins_exc=None,
|
|
run=None,
|
|
repair=None,
|
|
rebuild=None,
|
|
signature=True,
|
|
):
|
|
storage = FakeStorage()
|
|
calls: dict = {"gets": [], "triage": [], "repairs": [], "rebuilds": []}
|
|
monkeypatch.setattr(module, "settings", cfg if cfg is not None else _settings())
|
|
_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
|
|
)
|
|
|
|
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 {"job_name": "hermes-demo-repair-12", "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}
|
|
|
|
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)
|
|
return SimpleNamespace(storage=storage, calls=calls)
|
|
|
|
|
|
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(storage: FakeStorage, status: str, build_number: int = 12, as_json: bool = False) -> 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) -> 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()
|
|
|
|
|
|
def test_disabled_tick(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch, cfg=_settings(hermes_autotriage_enabled=False))
|
|
assert module.run_hermes_autotriage(env.storage) == {"status": "disabled"}
|
|
assert env.storage.events == []
|
|
assert env.calls["gets"] == []
|
|
|
|
|
|
def test_healthy_tick_without_incidents(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch, last_build=_build(13, "SUCCESS"))
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["status"] == "ok"
|
|
assert summary["jobs"][JOB] == {"status": "healthy", "resolved": []}
|
|
assert env.storage.events == []
|
|
assert env.calls["triage"] == []
|
|
|
|
|
|
def test_success_resolves_only_older_awaiting_rebuild(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch, last_build=_build(13, "SUCCESS"))
|
|
_seed_incident(env.storage, "awaiting_rebuild", build_number=12)
|
|
_seed_incident(env.storage, "awaiting_rebuild", build_number=13)
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["jobs"][JOB]["resolved"] == [INCIDENT_ID]
|
|
resolved = _events(env.storage, module.INCIDENT_EVENT_TYPE)[-1]
|
|
assert resolved["incident_id"] == INCIDENT_ID
|
|
assert resolved["status"] == "resolved"
|
|
assert resolved["phase"] == {"resolved_by_build": 13}
|
|
assert _gauge("12", "resolved") == 1.0
|
|
assert _gauge("12", "human_required") == 0.0
|
|
assert module.HERMES_TRIAGE_LAST_SUCCESS_TS._value.get() > 0
|
|
|
|
|
|
def test_new_failure_full_happy_path(monkeypatch) -> None:
|
|
success_before = _counter("repair_demo_fixture", "success")
|
|
env = _prepare(monkeypatch)
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
job_summary = summary["jobs"][JOB]
|
|
assert job_summary["status"] == "awaiting_rebuild"
|
|
assert job_summary["repair_job"] == "hermes-demo-repair-12"
|
|
assert _statuses(env.storage) == ["detected", "diagnosed", "repairing", "awaiting_rebuild"]
|
|
actions = _events(env.storage, module.ACTION_EVENT_TYPE)
|
|
assert [action["result"] for action in actions] == ["requested", "accepted", "executed"]
|
|
assert all(action["action"] == "repair_demo_fixture" for action in actions)
|
|
diagnosis = _events(env.storage, module.DIAGNOSIS_EVENT_TYPE)[0]
|
|
assert diagnosis["authorized"] is True
|
|
assert diagnosis["authorize_reason"] == "authorized"
|
|
assert diagnosis["run"] == {
|
|
"status": "completed",
|
|
"run_id": "run-1",
|
|
"session_id": "sess-1",
|
|
"error": None,
|
|
"duration_seconds": 1.5,
|
|
"denied_approvals": 0,
|
|
}
|
|
assert diagnosis["outcome"]["classification"] == "known_demo_fixture_failure"
|
|
assert env.calls["repairs"] == [
|
|
(
|
|
{
|
|
"namespace": "hermes-triage-demo",
|
|
"fixture_configmap": "hermes-triage-demo-fixture",
|
|
"image": "busybox:1.37",
|
|
},
|
|
INCIDENT_ID,
|
|
12,
|
|
)
|
|
]
|
|
assert env.calls["rebuilds"] == [JOB]
|
|
assert _counter("repair_demo_fixture", "success") == success_before + 1.0
|
|
assert _gauge("12", "awaiting_rebuild") == 1.0
|
|
assert _gauge("12", "detected") == 0.0
|
|
for phase in ("evidence", "diagnosis", "repair", "total"):
|
|
assert module.HERMES_TRIAGE_DURATION_SECONDS.labels(phase=phase)._value.get() >= 0.0
|
|
|
|
|
|
def test_prompt_is_frozen_shape(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch)
|
|
module.run_hermes_autotriage(env.storage)
|
|
config, prompt = env.calls["triage"][0]
|
|
assert config == {
|
|
"base_url": "http://hermes:8642",
|
|
"api_key": "key",
|
|
"total_timeout_seconds": 420.0,
|
|
}
|
|
assert prompt.startswith("Use $triage-titan-test-failures.\n")
|
|
assert f"Analyze incident {INCIDENT_ID}." in prompt
|
|
assert f'"<must equal {INCIDENT_ID}>"' in prompt
|
|
assert "You are diagnosing only; you do not execute anything." in prompt
|
|
assert "Set human_required to false when the evidence matches" in prompt
|
|
assert "Do not perform mutations.\n\nBundle:\n" in prompt
|
|
assert prompt.rstrip().endswith('"log_evidence":{"records":[]}}')
|
|
|
|
|
|
def test_observe_mode_requires_human_without_actions(monkeypatch) -> None:
|
|
rejected_before = _counter("repair_demo_fixture", "rejected")
|
|
env = _prepare(monkeypatch, cfg=_settings(hermes_autoremediation_enabled=False))
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["jobs"][JOB] == {
|
|
"status": "human_required",
|
|
"incident_id": INCIDENT_ID,
|
|
"reason": "autoremediation_disabled",
|
|
}
|
|
assert _statuses(env.storage) == ["detected", "diagnosed", "human_required"]
|
|
assert _events(env.storage, module.ACTION_EVENT_TYPE) == []
|
|
assert env.calls["repairs"] == []
|
|
assert env.calls["rebuilds"] == []
|
|
assert _counter("repair_demo_fixture", "rejected") == rejected_before + 1.0
|
|
|
|
|
|
def test_known_incident_is_deduped(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch)
|
|
_seed_incident(env.storage, "human_required")
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["jobs"][JOB] == {"status": "deduped", "incident_id": INCIDENT_ID}
|
|
assert len(env.storage.events) == 1
|
|
assert env.calls["triage"] == []
|
|
|
|
|
|
def test_incident_state_reads_json_string_detail(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch)
|
|
_seed_incident(env.storage, "resolved", as_json=True)
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["jobs"][JOB]["status"] == "deduped"
|
|
|
|
|
|
def test_failed_rebuild_marks_both_incidents(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch, last_build=_build(13, "FAILURE"))
|
|
_seed_incident(env.storage, "awaiting_rebuild", build_number=12)
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["jobs"][JOB] == {
|
|
"status": "rebuild_failed",
|
|
"incident_id": f"{JOB}/13",
|
|
"failed_incident": INCIDENT_ID,
|
|
}
|
|
details = _events(env.storage, module.INCIDENT_EVENT_TYPE)[1:]
|
|
assert [(d["incident_id"], d["status"]) for d in details] == [
|
|
(INCIDENT_ID, "failed"),
|
|
(f"{JOB}/13", "human_required"),
|
|
]
|
|
assert details[1]["phase"] == {"reason": "repair rebuild failed"}
|
|
assert env.calls["triage"] == []
|
|
assert env.calls["repairs"] == []
|
|
|
|
|
|
@pytest.mark.parametrize("status", ["timeout", "lost", "error", "failed", "cancelled"])
|
|
def test_unfinished_hermes_run_requires_human(monkeypatch, status) -> None:
|
|
env = _prepare(monkeypatch, run=_run(status=status, error="boom"))
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["jobs"][JOB]["reason"] == f"hermes_run_{status}"
|
|
assert _statuses(env.storage) == ["detected", "human_required"]
|
|
diagnosis = _events(env.storage, module.DIAGNOSIS_EVENT_TYPE)[0]
|
|
assert diagnosis["run"]["status"] == status
|
|
assert diagnosis["outcome"] is None
|
|
assert diagnosis["authorized"] is False
|
|
assert env.calls["repairs"] == []
|
|
|
|
|
|
def test_invalid_response_requires_human(monkeypatch) -> None:
|
|
rejected_before = _counter("unknown", "rejected")
|
|
env = _prepare(monkeypatch, run=_run(output="no json here"))
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["jobs"][JOB]["reason"].startswith("response_invalid")
|
|
assert _statuses(env.storage) == ["detected", "human_required"]
|
|
assert _counter("unknown", "rejected") == rejected_before + 1.0
|
|
|
|
|
|
def test_model_human_required_is_rejected(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch, run=_run(output=_model_output(human_required=True)))
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["jobs"][JOB]["reason"] == "human_required"
|
|
assert _statuses(env.storage) == ["detected", "diagnosed", "human_required"]
|
|
assert env.calls["repairs"] == []
|
|
|
|
|
|
def test_missing_signature_is_rejected(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch, signature=False)
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["jobs"][JOB]["reason"] == "evidence_signature_missing"
|
|
assert _statuses(env.storage) == ["detected", "diagnosed", "human_required"]
|
|
assert env.calls["repairs"] == []
|
|
|
|
|
|
def test_non_allowlisted_action_is_rejected(monkeypatch) -> None:
|
|
rejected_before = _counter("unknown", "rejected")
|
|
output = _model_output(requested_action={"type": "run_ariadne_job", "id": "other_action"})
|
|
env = _prepare(monkeypatch, run=_run(output=output))
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["jobs"][JOB]["reason"].startswith("action_not_allowlisted")
|
|
assert _counter("unknown", "rejected") == rejected_before + 1.0
|
|
|
|
|
|
def test_prior_action_blocks_second_action(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch)
|
|
_seed_incident(env.storage, "detected")
|
|
env.storage.record_event(
|
|
module.ACTION_EVENT_TYPE,
|
|
{"incident_id": INCIDENT_ID, "action": "repair_demo_fixture", "result": "requested"},
|
|
)
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["jobs"][JOB]["reason"] == "max_actions_reached"
|
|
assert env.calls["repairs"] == []
|
|
|
|
|
|
def test_repair_failure_marks_failed_and_human(monkeypatch) -> None:
|
|
failed_before = _counter("repair_demo_fixture", "failed")
|
|
env = _prepare(
|
|
monkeypatch,
|
|
repair={"job_name": "hermes-demo-repair-12", "succeeded": False, "error": "repair job failed"},
|
|
)
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["jobs"][JOB] == {
|
|
"status": "failed",
|
|
"incident_id": INCIDENT_ID,
|
|
"reason": "repair job failed",
|
|
}
|
|
assert _statuses(env.storage) == ["detected", "diagnosed", "repairing", "failed"]
|
|
actions = _events(env.storage, module.ACTION_EVENT_TYPE)
|
|
assert [action["result"] for action in actions] == ["requested", "accepted", "failed"]
|
|
assert env.calls["rebuilds"] == []
|
|
assert _counter("repair_demo_fixture", "failed") == failed_before + 1.0
|
|
assert _gauge("12", "failed") == 1.0
|
|
assert _gauge("12", "human_required") == 1.0
|
|
|
|
|
|
def test_rebuild_trigger_failure_marks_failed(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch, rebuild={"requested": False, "error": "rebuild http 500"})
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["jobs"][JOB]["status"] == "failed"
|
|
assert summary["jobs"][JOB]["reason"] == "rebuild http 500"
|
|
actions = _events(env.storage, module.ACTION_EVENT_TYPE)
|
|
assert [action["result"] for action in actions] == ["requested", "accepted", "failed"]
|
|
|
|
|
|
def test_jenkins_fetch_failure_skips_job(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch, jenkins_exc=RuntimeError("boom"))
|
|
summary = module.run_hermes_autotriage(env.storage)
|
|
assert summary["jobs"][JOB] == {"status": "skipped"}
|
|
assert env.storage.events == []
|
|
|
|
|
|
def test_building_build_is_skipped(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch, last_build=_build(12, None, building=True))
|
|
assert module.run_hermes_autotriage(env.storage)["jobs"][JOB] == {"status": "skipped"}
|
|
assert env.storage.events == []
|
|
|
|
|
|
def test_empty_jenkins_base_url_skips(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch, cfg=_settings(jenkins_base_url=""))
|
|
assert module.run_hermes_autotriage(env.storage)["jobs"][JOB] == {"status": "skipped"}
|
|
assert env.calls["gets"] == []
|
|
|
|
|
|
def test_non_terminal_result_is_ignored(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch, last_build=_build(12, "ABORTED"))
|
|
assert module.run_hermes_autotriage(env.storage)["jobs"][JOB] == {
|
|
"status": "ignored",
|
|
"result": "ABORTED",
|
|
}
|
|
assert env.storage.events == []
|
|
|
|
|
|
def test_event_detail_tolerates_bad_payloads() -> None:
|
|
assert module._event_detail({"detail": "not-json"}) is None
|
|
assert module._event_detail({"detail": "[1,2]"}) is None
|
|
assert module._event_detail({"detail": 5}) is None
|
|
assert module._event_detail("not-a-row") is None
|
|
assert module._int_value("not-a-number") == 0
|
|
|
|
|
|
def test_jenkins_request_uses_basic_auth_and_tree(monkeypatch) -> None:
|
|
env = _prepare(monkeypatch, last_build=_build(13, "SUCCESS"))
|
|
module.run_hermes_autotriage(env.storage)
|
|
assert env.calls["client_kwargs"]["auth"] == ("user", "token")
|
|
url, params = env.calls["gets"][0]
|
|
assert url == f"https://ci.example/job/{JOB}/api/json"
|
|
assert params == {"tree": "lastBuild[number,result,building,timestamp,duration,url]"}
|