372 lines
15 KiB
Python
372 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from ariadne.services import testing_triage
|
|
from ariadne.services import testing_triage_diagnosis
|
|
|
|
|
|
class DummyStorage:
|
|
def __init__(self) -> None:
|
|
self.events: list[tuple[str, dict]] = []
|
|
|
|
def record_event(self, event_type: str, detail: dict) -> None:
|
|
self.events.append((event_type, detail))
|
|
|
|
def list_events(self, limit: int = 1, event_type: str | None = None): # type: ignore[no-untyped-def]
|
|
matching = [
|
|
{"detail": detail}
|
|
for stored_type, detail in self.events
|
|
if event_type is None or stored_type == event_type
|
|
]
|
|
return matching[-limit:][::-1]
|
|
|
|
|
|
class SettingsStub:
|
|
def __init__(self, **overrides) -> None: # type: ignore[no-untyped-def]
|
|
self.testing_triage_model_url = ""
|
|
self.testing_triage_model = "qwen2.5:7b-instruct-q4_0"
|
|
self.testing_triage_model_timeout_sec = 1.0
|
|
for key, value in overrides.items():
|
|
setattr(self, key, value)
|
|
|
|
|
|
def test_run_testing_triage_stores_diagnosis_when_model_enabled(monkeypatch) -> None:
|
|
storage = DummyStorage()
|
|
bundle = {"summary": {"status": "needs_attention", "problem_count": 1, "failed_suites": ["ariadne"]}}
|
|
diagnosis = {"kind": "testing_triage_diagnosis", "status": "needs_attention"}
|
|
|
|
monkeypatch.setattr(testing_triage, "model_diagnosis_enabled", lambda: True)
|
|
monkeypatch.setattr(testing_triage, "collect_testing_triage", lambda _storage: bundle)
|
|
monkeypatch.setattr(testing_triage, "diagnose_testing_triage", lambda _bundle: diagnosis)
|
|
|
|
summary = testing_triage.run_testing_triage(storage) # type: ignore[arg-type]
|
|
latest = testing_triage.latest_testing_triage_diagnosis(storage) # type: ignore[arg-type]
|
|
|
|
assert summary.status == "needs_attention"
|
|
assert [event[0] for event in storage.events] == [
|
|
testing_triage.TRIAGE_EVENT_TYPE,
|
|
testing_triage.TRIAGE_DIAGNOSIS_EVENT_TYPE,
|
|
]
|
|
assert latest == diagnosis
|
|
|
|
|
|
def test_diagnose_testing_triage_calls_local_ollama(monkeypatch) -> None:
|
|
captured = {}
|
|
model_response = json.dumps(
|
|
{
|
|
"headline": "Ariadne has one failing suite.",
|
|
"root_cause": "Jenkins reported a failed ariadne run.",
|
|
"blast_radius": "ariadne",
|
|
"confidence": "medium",
|
|
"needs_human": True,
|
|
"next_actions": ["Inspect the failed Jenkins build log."],
|
|
"evidence_refs": ["summary.failed_suites=ariadne"],
|
|
}
|
|
)
|
|
|
|
class FakeResponse:
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
def json(self): # type: ignore[no-untyped-def]
|
|
return {"response": model_response}
|
|
|
|
class FakeClient:
|
|
def __init__(self, *, timeout) -> None: # type: ignore[no-untyped-def]
|
|
captured["timeout"] = timeout
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args) -> None: # type: ignore[no-untyped-def]
|
|
return None
|
|
|
|
def post(self, url, json=None): # type: ignore[no-untyped-def]
|
|
captured["url"] = url
|
|
captured["request"] = json
|
|
return FakeResponse()
|
|
|
|
monkeypatch.setattr(
|
|
testing_triage_diagnosis,
|
|
"settings",
|
|
SettingsStub(
|
|
testing_triage_model_url="http://ollama/",
|
|
testing_triage_model="tiny-model",
|
|
testing_triage_model_timeout_sec=3.0,
|
|
),
|
|
)
|
|
monkeypatch.setattr(testing_triage_diagnosis.httpx, "Client", FakeClient)
|
|
|
|
diagnosis = testing_triage_diagnosis.diagnose_testing_triage(
|
|
{
|
|
"kind": "testing_triage_bundle",
|
|
"generated_at": "now",
|
|
"summary": {"status": "needs_attention", "problem_count": 1, "failed_suites": ["ariadne"]},
|
|
"evidence": {"jenkins": {"failed_builds": [{"job": "ariadne"}]}},
|
|
"unknowns": [],
|
|
}
|
|
)
|
|
|
|
assert captured["url"] == "http://ollama/api/generate"
|
|
assert captured["timeout"] == 3.0
|
|
assert captured["request"]["model"] == "tiny-model"
|
|
assert captured["request"]["format"]["additionalProperties"] is False
|
|
assert "headline" in captured["request"]["format"]["required"]
|
|
assert "Do not include keys named pipelines" in captured["request"]["prompt"]
|
|
assert diagnosis["kind"] == "testing_triage_diagnosis"
|
|
assert diagnosis["status"] == "needs_attention"
|
|
assert diagnosis["diagnosis"]["confidence"] == "medium"
|
|
assert diagnosis["diagnosis"]["next_actions"] == ["Inspect the failed Jenkins build log."]
|
|
|
|
|
|
def test_diagnose_testing_triage_handles_disabled_and_bad_json(monkeypatch) -> None:
|
|
monkeypatch.setattr(testing_triage_diagnosis, "settings", SettingsStub(testing_triage_model_url=""))
|
|
disabled = testing_triage_diagnosis.diagnose_testing_triage({"summary": {"status": "ok", "problem_count": 0}})
|
|
assert disabled["status"] == "unavailable"
|
|
assert disabled["diagnosis"]["root_cause"] == "model_url_not_configured"
|
|
|
|
class BadResponse:
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
def json(self): # type: ignore[no-untyped-def]
|
|
return {"response": "not json"}
|
|
|
|
class BadClient:
|
|
def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def]
|
|
return None
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args) -> None: # type: ignore[no-untyped-def]
|
|
return None
|
|
|
|
def post(self, url, json=None): # type: ignore[no-untyped-def]
|
|
return BadResponse()
|
|
|
|
monkeypatch.setattr(testing_triage_diagnosis, "settings", SettingsStub(testing_triage_model_url="http://ollama"))
|
|
monkeypatch.setattr(testing_triage_diagnosis.httpx, "Client", BadClient)
|
|
diagnosis = testing_triage_diagnosis.diagnose_testing_triage({"summary": {"status": "ok", "problem_count": 0}})
|
|
|
|
assert diagnosis["status"] == "ok"
|
|
assert "model_json_parse_failed" in diagnosis["unknowns"][0]
|
|
|
|
|
|
def test_diagnose_testing_triage_handles_empty_model_response(monkeypatch) -> None:
|
|
class EmptyResponse:
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
def json(self): # type: ignore[no-untyped-def]
|
|
return {"response": ""}
|
|
|
|
class EmptyClient:
|
|
def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def]
|
|
return None
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args) -> None: # type: ignore[no-untyped-def]
|
|
return None
|
|
|
|
def post(self, url, json=None): # type: ignore[no-untyped-def]
|
|
return EmptyResponse()
|
|
|
|
monkeypatch.setattr(testing_triage_diagnosis, "settings", SettingsStub(testing_triage_model_url="http://ollama"))
|
|
monkeypatch.setattr(testing_triage_diagnosis.httpx, "Client", EmptyClient)
|
|
diagnosis = testing_triage_diagnosis.diagnose_testing_triage({"summary": {"status": "ok", "problem_count": 0}})
|
|
|
|
assert diagnosis["status"] == "ok"
|
|
assert "empty_model_response" in diagnosis["unknowns"]
|
|
|
|
|
|
def test_latest_testing_triage_diagnosis_decodes_stored_json() -> None:
|
|
storage = DummyStorage()
|
|
assert testing_triage_diagnosis.latest_testing_triage_diagnosis(storage) is None # type: ignore[arg-type]
|
|
payload = {"kind": "testing_triage_diagnosis", "status": "ok"}
|
|
storage.events.append((testing_triage_diagnosis.TRIAGE_DIAGNOSIS_EVENT_TYPE, json.dumps(payload)))
|
|
|
|
assert testing_triage_diagnosis.latest_testing_triage_diagnosis(storage) == payload # type: ignore[arg-type]
|
|
|
|
storage.events.append((testing_triage_diagnosis.TRIAGE_DIAGNOSIS_EVENT_TYPE, "not json"))
|
|
assert testing_triage_diagnosis.latest_testing_triage_diagnosis(storage) is None # type: ignore[arg-type]
|
|
|
|
storage.events.append((testing_triage_diagnosis.TRIAGE_DIAGNOSIS_EVENT_TYPE, json.dumps(["not", "a", "dict"])))
|
|
assert testing_triage_diagnosis.latest_testing_triage_diagnosis(storage) is None # type: ignore[arg-type]
|
|
|
|
storage.events.append((testing_triage_diagnosis.TRIAGE_DIAGNOSIS_EVENT_TYPE, 7))
|
|
assert testing_triage_diagnosis.latest_testing_triage_diagnosis(storage) is None # type: ignore[arg-type]
|
|
|
|
|
|
def test_model_configuration_helpers_normalize_settings(monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
testing_triage_diagnosis,
|
|
"settings",
|
|
SettingsStub(
|
|
testing_triage_model_url=" http://ollama.local/ ",
|
|
testing_triage_model="",
|
|
testing_triage_model_timeout_sec=0,
|
|
),
|
|
)
|
|
|
|
assert testing_triage_diagnosis.model_diagnosis_enabled()
|
|
assert testing_triage_diagnosis._model_url() == "http://ollama.local" # noqa: SLF001
|
|
assert testing_triage_diagnosis._model_name() == "" # noqa: SLF001
|
|
assert testing_triage_diagnosis._model_timeout() == 180.0 # noqa: SLF001
|
|
|
|
|
|
def test_diagnose_testing_triage_handles_model_request_failure(monkeypatch) -> None:
|
|
class FailingClient:
|
|
def __init__(self, *args, **kwargs) -> None: # type: ignore[no-untyped-def]
|
|
return None
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args) -> None: # type: ignore[no-untyped-def]
|
|
return None
|
|
|
|
def post(self, url, json=None): # type: ignore[no-untyped-def]
|
|
raise RuntimeError("ollama unavailable")
|
|
|
|
monkeypatch.setattr(testing_triage_diagnosis, "settings", SettingsStub(testing_triage_model_url="http://ollama"))
|
|
monkeypatch.setattr(testing_triage_diagnosis.httpx, "Client", FailingClient)
|
|
|
|
diagnosis = testing_triage_diagnosis.diagnose_testing_triage(
|
|
{"generated_at": "evidence-time", "summary": {"status": "warning", "problem_count": 2}}
|
|
)
|
|
|
|
assert diagnosis["status"] == "unavailable"
|
|
assert diagnosis["evidence_generated_at"] == "evidence-time"
|
|
assert "model_request_failed" in diagnosis["diagnosis"]["root_cause"]
|
|
|
|
|
|
def test_prompt_and_evidence_payload_use_safe_defaults(monkeypatch) -> None:
|
|
monkeypatch.setattr(testing_triage_diagnosis, "_MAX_MODEL_EVIDENCE_CHARS", 120)
|
|
prompt = testing_triage_diagnosis._diagnosis_prompt( # noqa: SLF001
|
|
{
|
|
"kind": "testing_triage_bundle",
|
|
"summary": "bad summary",
|
|
"evidence": {"large": "x" * 300},
|
|
"unknowns": "bad unknowns",
|
|
}
|
|
)
|
|
evidence = testing_triage_diagnosis._model_evidence_payload( # noqa: SLF001
|
|
{"summary": "bad summary", "evidence": "bad evidence", "unknowns": "bad unknowns"}
|
|
)
|
|
|
|
assert prompt.endswith("[truncated]")
|
|
assert evidence["summary"] == {}
|
|
assert evidence["evidence"] == {}
|
|
assert evidence["unknowns"] == []
|
|
|
|
|
|
def test_diagnosis_from_model_coerces_fallback_values(monkeypatch) -> None:
|
|
monkeypatch.setattr(testing_triage_diagnosis, "settings", SettingsStub(testing_triage_model="triage-model"))
|
|
diagnosis = testing_triage_diagnosis._diagnosis_from_model( # noqa: SLF001
|
|
{
|
|
"generated_at": "bundle-time",
|
|
"summary": {"status": "ok", "problem_count": 0, "failed_suites": []},
|
|
"unknowns": ["existing_unknown"],
|
|
},
|
|
{
|
|
"headline": "",
|
|
"root_cause": None,
|
|
"confidence": "certain",
|
|
"needs_human": "false",
|
|
"next_actions": "Verify current dashboard state.",
|
|
"evidence_refs": ["suite", "", 7],
|
|
},
|
|
"raw response",
|
|
"parse warning",
|
|
)
|
|
|
|
assert diagnosis["status"] == "ok"
|
|
assert diagnosis["model"] == "triage-model"
|
|
assert diagnosis["diagnosis"]["headline"] == "Testing triage needs review."
|
|
assert diagnosis["diagnosis"]["confidence"] == "low"
|
|
assert diagnosis["diagnosis"]["needs_human"] is False
|
|
assert diagnosis["diagnosis"]["next_actions"] == ["Verify current dashboard state."]
|
|
assert diagnosis["diagnosis"]["evidence_refs"] == ["suite", "7"]
|
|
assert diagnosis["unknowns"] == ["existing_unknown", "parse warning"]
|
|
|
|
|
|
def test_diagnosis_from_model_rejects_non_english_and_out_of_scope_jobs(monkeypatch) -> None:
|
|
monkeypatch.setattr(testing_triage_diagnosis, "settings", SettingsStub(testing_triage_model="triage-model"))
|
|
diagnosis = testing_triage_diagnosis._diagnosis_from_model( # noqa: SLF001
|
|
{
|
|
"generated_at": "bundle-time",
|
|
"summary": {
|
|
"status": "needs_attention",
|
|
"problem_count": 2,
|
|
"failed_suites": ["titan_iac"],
|
|
},
|
|
"unknowns": [],
|
|
},
|
|
{
|
|
"headline": "多项目持续集成状态更新",
|
|
"root_cause": "arcanagon is stale and titan_iac failed.",
|
|
"blast_radius": "arcanagon and titan_iac",
|
|
"confidence": "high",
|
|
"needs_human": True,
|
|
"next_actions": [
|
|
"检查 titan_iac logs.",
|
|
"Review arcanagon logs.",
|
|
"Review titan_iac logs.",
|
|
],
|
|
"evidence_refs": [
|
|
"https://ci.bstein.dev/job/arcanagon/1/consoleText",
|
|
"https://ci.bstein.dev/job/titan-iac/463/consoleText",
|
|
],
|
|
},
|
|
"raw response",
|
|
None,
|
|
)
|
|
|
|
assert diagnosis["diagnosis"]["headline"] == "Testing triage needs review."
|
|
assert diagnosis["diagnosis"]["root_cause"] == "Evidence is insufficient for a confident root cause."
|
|
assert diagnosis["diagnosis"]["blast_radius"] == "titan_iac"
|
|
assert diagnosis["diagnosis"]["next_actions"] == ["Review titan_iac logs."]
|
|
assert diagnosis["diagnosis"]["evidence_refs"] == ["https://ci.bstein.dev/job/titan-iac/463/consoleText"]
|
|
assert "model_headline_non_english" in diagnosis["unknowns"]
|
|
assert "model_root_cause_out_of_scope" in diagnosis["unknowns"]
|
|
assert "model_evidence_refs_out_of_scope" in diagnosis["unknowns"]
|
|
|
|
|
|
def test_diagnosis_helpers_cover_boolean_and_defaults() -> None:
|
|
assert testing_triage_diagnosis._bool_value("yes", False) is True # noqa: SLF001
|
|
assert testing_triage_diagnosis._bool_value("no", True) is False # noqa: SLF001
|
|
assert testing_triage_diagnosis._bool_value("maybe", True) is True # noqa: SLF001
|
|
assert testing_triage_diagnosis._allowed_suite_jobs({"failed_suites": ["", "bstein_home", "data_prepper"]}) == { # noqa: SLF001
|
|
"bstein_home",
|
|
"bstein-home",
|
|
"bstein-dev-home",
|
|
"data_prepper",
|
|
"data-prepper",
|
|
}
|
|
assert testing_triage_diagnosis._default_next_actions({"problem_count": 0}) == [ # noqa: SLF001
|
|
"No action required unless a fresh bundle changes the status."
|
|
]
|
|
assert testing_triage_diagnosis._default_next_actions({"problem_count": 1}) == [ # noqa: SLF001
|
|
"Review the evidence bundle sections with non-empty problem lists.",
|
|
"Check the named Jenkins build logs and Flux Kustomizations before changing manifests.",
|
|
]
|
|
|
|
|
|
def test_safe_evidence_refs_rejects_non_ascii_refs() -> None:
|
|
unknowns = []
|
|
refs = testing_triage_diagnosis._safe_evidence_refs(["référence"], {}, unknowns) # noqa: SLF001
|
|
assert refs == []
|
|
assert unknowns == ["model_evidence_refs_non_english"]
|
|
|
|
|
|
def test_default_evidence_refs_include_failed_suites() -> None:
|
|
refs = testing_triage_diagnosis._default_evidence_refs( # noqa: SLF001
|
|
{"status": "needs_attention", "problem_count": 3, "failed_suites": ["a", "b", "c", "d", "e", "f", "g"]}
|
|
)
|
|
|
|
assert refs[-1] == "summary.failed_suites=a,b,c,d,e,f"
|