ariadne/tests/test_testing_triage_diagnosis.py

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]