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"] == "json" 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_latest_testing_triage_diagnosis_decodes_stored_json() -> None: storage = DummyStorage() 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] 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_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"