awoooi/apps/api/tests/test_adr100_slo_status_service.py

from __future__ import annotations

from typing import Any

import pytest

from src.services.adr100_slo_status_service import (
    Adr100SloStatusService,
    _build_verification_coverage_payload,
)


class _FakePrometheusResponse:
    def __init__(self, payload: dict[str, Any]) -> None:
        self._payload = payload

    def json(self) -> dict[str, Any]:
        return self._payload


class _FakePrometheusClient:
    def __init__(self, values: dict[str, str]) -> None:
        self.values = values

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc, tb):
        return False

    async def get(self, *args, **kwargs):  # noqa: ANN002, ANN003
        query = str(kwargs.get("params", {}).get("query", ""))
        value = self.values.get(query)
        if value is None:
            return _FakePrometheusResponse({
                "status": "success",
                "data": {"result": []},
            })
        return _FakePrometheusResponse({
            "status": "success",
            "data": {"result": [{"value": [1778756604, value]}]},
        })


async def _low_volume_coverage(self):  # noqa: ANN001
    return {
        "schema_version": "adr100_verification_coverage_v1",
        "source": "postgresql",
        "window": "24h",
        "status": "skipped_low_volume",
        "reason": "no_auto_repair_executions_24h",
        "evaluable": False,
        "total_auto": 0,
        "successful_auto": 0,
        "verified_auto": 0,
        "verified_success": 0,
        "verified_non_success": 0,
        "unverified_auto": 0,
        "coverage_rate": None,
        "verification_success_rate": None,
        "last_auto_at": None,
        "last_verified_auto_at": None,
        "last_verification_evidence_at": None,
        "latest_auto_age_seconds": None,
        "last_verified_auto_age_seconds": None,
        "recent_unverified": [],
    }


@pytest.mark.asyncio
async def test_fetch_report_marks_ratio_slos_low_volume(monkeypatch):
    values = {
        "sum(rate(automation_operation_log_total[5m]))": "0",
        'sum(rate(automation_operation_log_total{outcome="auto_executed"}[5m]))': "0",
        "sum(rate(approval_records_high_confidence_total[1h]))": "0",
        "max(knowledge_entries_created_24h) or max(sli:km_growth_rate:24h)": "24",
    }

    monkeypatch.setattr(
        "httpx.AsyncClient",
        lambda *args, **kwargs: _FakePrometheusClient(values),
    )
    monkeypatch.setattr(
        Adr100SloStatusService,
        "_fetch_verification_coverage",
        _low_volume_coverage,
    )

    report = await Adr100SloStatusService().fetch_report()

    by_name = {metric["name"]: metric for metric in report["metrics"]}
    assert by_name["decision_accuracy"]["status"] == "skipped_low_volume"
    assert by_name["decision_accuracy"]["evaluable"] is False
    assert by_name["confidence_calibration"]["status"] == "skipped_low_volume"
    assert by_name["km_growth_rate"]["status"] == "ok"
    assert by_name["km_growth_rate"]["value"] == 24
    assert report["overall_status"] == "partial"
    assert report["overall_compliance"] == 1.0


@pytest.mark.asyncio
async def test_fetch_report_classifies_hard_red_line_violation(monkeypatch):
    values = {
        "sum(rate(automation_operation_log_total[5m]))": "0.02",
        "sli:autonomy_rate:5m": "0.5",
        'sum(rate(automation_operation_log_total{outcome="auto_executed"}[5m]))': "0",
        "sum(rate(approval_records_high_confidence_total[1h]))": "0",
        "max(knowledge_entries_created_24h) or max(sli:km_growth_rate:24h)": "3",
    }

    monkeypatch.setattr(
        "httpx.AsyncClient",
        lambda *args, **kwargs: _FakePrometheusClient(values),
    )
    monkeypatch.setattr(
        Adr100SloStatusService,
        "_fetch_verification_coverage",
        _low_volume_coverage,
    )

    report = await Adr100SloStatusService().fetch_report()

    by_name = {metric["name"]: metric for metric in report["metrics"]}
    assert by_name["autonomy_rate"]["status"] == "violated"
    assert by_name["autonomy_rate"]["sample_count"] == 6
    assert by_name["km_growth_rate"]["status"] == "violated"
    assert report["overall_status"] == "violated"


def test_verification_coverage_payload_flags_backlog():
    payload = _build_verification_coverage_payload(
        {
            "total_auto": 7,
            "successful_auto": 6,
            "verified_auto": 5,
            "verified_success": 4,
            "verified_non_success": 1,
            "unverified_auto": 2,
            "last_auto_at": None,
            "last_verified_auto_at": None,
            "last_verification_evidence_at": None,
            "latest_auto_age_seconds": 90,
            "last_verified_auto_age_seconds": 120,
        },
        [
            {
                "id": "are-1",
                "incident_id": "INC-1",
                "success": True,
                "created_at": None,
            },
        ],
    )

    assert payload["status"] == "warning"
    assert payload["reason"] == "verification_backlog_present"
    assert payload["coverage_rate"] == pytest.approx(5 / 7)
    assert payload["verification_success_rate"] == pytest.approx(4 / 5)
    assert payload["recent_unverified"][0]["incident_id"] == "INC-1"


def test_verification_coverage_payload_skips_when_no_auto_repair():
    payload = _build_verification_coverage_payload(
        {
            "total_auto": 0,
            "successful_auto": 0,
            "verified_auto": 0,
            "verified_success": 0,
            "verified_non_success": 0,
            "unverified_auto": 0,
        },
        [],
    )

    assert payload["status"] == "skipped_low_volume"
    assert payload["reason"] == "no_auto_repair_executions_24h"
    assert payload["evaluable"] is False