microdao-daarion/tests/test_risk_digest_attribution.py

"""
tests/test_risk_digest_attribution.py — Tests for "Likely Causes" section in digest.

Tests:
- digest includes "Likely Causes" section when attribution present
- causes listed with evidence
- section absent when no regressions
- attribution_summary in JSON
"""
import sys
import pytest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "services" / "router"))

from risk_digest import daily_digest, _build_markdown
from risk_engine import _builtin_defaults, _reload_policy


@pytest.fixture(autouse=True)
def reset_policy_cache():
    _reload_policy()
    yield
    _reload_policy()


@pytest.fixture
def policy():
    return _builtin_defaults()


def _report_with_attribution(
    service, score, band, delta_24h=None,
    causes=None, attribution_summary=None,
):
    from risk_engine import score_to_band, _builtin_defaults
    p = _builtin_defaults()
    b = band or score_to_band(score, p)
    attr = None
    if causes is not None:
        attr = {
            "service": service,
            "env": "prod",
            "causes": causes,
            "summary": attribution_summary or "Likely causes: test.",
            "llm_enrichment": {"enabled": False, "text": None},
        }
    trend = {
        "delta_24h": delta_24h,
        "delta_7d": None,
        "slope_per_day": None,
        "volatility": None,
        "regression": {"warn": delta_24h is not None and delta_24h >= 10,
                       "fail": delta_24h is not None and delta_24h >= 20},
    }
    return {
        "service": service,
        "env": "prod",
        "score": score,
        "band": b,
        "components": {},
        "reasons": [],
        "recommendations": [],
        "updated_at": "2026-02-23T00:00:00",
        "trend": trend,
        "attribution": attr,
    }


def _sample_causes():
    return [
        {"type": "deploy", "score": 30, "confidence": "medium",
         "evidence": ["deploy alerts: 2 in last 24h"]},
        {"type": "incident_storm", "score": 20, "confidence": "medium",
         "evidence": ["occurrences_60m=14", "escalations_24h=3"]},
    ]


# ─── Markdown section tests ───────────────────────────────────────────────────

class TestDigestAttributionSection:
    def test_likely_causes_section_present(self, policy):
        reports = [
            _report_with_attribution(
                "gateway", 75, "high", delta_24h=20,
                causes=_sample_causes(),
                attribution_summary="Likely causes: deploy activity + incident storm.",
            ),
        ]
        result = daily_digest(env="prod", service_reports=reports, policy=policy,
                               write_files=False)
        assert "Likely Causes" in result["markdown"]
        assert "gateway" in result["markdown"]

    def test_cause_type_in_markdown(self, policy):
        reports = [
            _report_with_attribution(
                "router", 65, "high", delta_24h=15,
                causes=[{"type": "deploy", "score": 30, "confidence": "medium",
                          "evidence": ["deploy alerts: 1 in last 24h"]}],
            ),
        ]
        result = daily_digest(env="prod", service_reports=reports, policy=policy,
                               write_files=False)
        assert "deploy" in result["markdown"]

    def test_evidence_in_markdown(self, policy):
        reports = [
            _report_with_attribution(
                "gateway", 80, "critical", delta_24h=25,
                causes=[{"type": "deploy", "score": 30, "confidence": "high",
                          "evidence": ["deploy alerts: 3 in last 24h", "last seen: 2026-02-23"]}],
            ),
        ]
        result = daily_digest(env="prod", service_reports=reports, policy=policy,
                               write_files=False)
        assert "deploy alerts" in result["markdown"]

    def test_likely_causes_section_absent_without_regression(self, policy):
        """No delta_24h > 0 → no Likely Causes section."""
        reports = [
            _report_with_attribution(
                "gateway", 30, "medium", delta_24h=None,
                causes=_sample_causes(),
            ),
        ]
        result = daily_digest(env="prod", service_reports=reports, policy=policy,
                               write_files=False)
        assert "Likely Causes" not in result["markdown"]

    def test_likely_causes_absent_when_no_attribution(self, policy):
        """Reports without attribution → section absent."""
        from risk_engine import score_to_band
        p = _builtin_defaults()
        reports = [{
            "service": "gateway", "env": "prod", "score": 80, "band": "high",
            "components": {}, "reasons": [], "recommendations": [],
            "updated_at": "2026-02-23T00:00:00",
            "trend": {"delta_24h": 20, "delta_7d": None,
                      "slope_per_day": None, "volatility": None,
                      "regression": {"warn": True, "fail": True}},
            "attribution": None,
        }]
        result = daily_digest(env="prod", service_reports=reports, policy=policy,
                               write_files=False)
        assert "Likely Causes" not in result["markdown"]

    def test_attribution_summary_in_json(self, policy):
        reports = [
            _report_with_attribution(
                "gateway", 75, "high", delta_24h=20,
                causes=_sample_causes(),
                attribution_summary="Likely causes: deploy activity + incident storm.",
            ),
        ]
        result = daily_digest(env="prod", service_reports=reports, policy=policy,
                               write_files=False)
        top_svcs = result["json_data"]["top_services"]
        gw = next((s for s in top_svcs if s["service"] == "gateway"), None)
        assert gw is not None
        assert gw.get("attribution_summary") == "Likely causes: deploy activity + incident storm."

    def test_top_causes_in_json(self, policy):
        causes = _sample_causes()
        reports = [
            _report_with_attribution(
                "gateway", 75, "high", delta_24h=20, causes=causes,
            ),
        ]
        result = daily_digest(env="prod", service_reports=reports, policy=policy,
                               write_files=False)
        gw = next(s for s in result["json_data"]["top_services"]
                  if s["service"] == "gateway")
        assert len(gw["top_causes"]) <= 2
        assert gw["top_causes"][0]["type"] == "deploy"

    def test_llm_text_appended_if_present(self, policy):
        causes = _sample_causes()
        attr = {
            "service": "gateway", "env": "prod",
            "causes": causes,
            "summary": "Likely causes: deploy.",
            "llm_enrichment": {"enabled": True, "text": "The deploy event likely caused instability."},
        }
        reports = [{
            "service": "gateway", "env": "prod", "score": 80, "band": "high",
            "components": {}, "reasons": [], "recommendations": [],
            "updated_at": "2026-02-23T00:00:00",
            "trend": {"delta_24h": 25, "delta_7d": None,
                      "slope_per_day": None, "volatility": None,
                      "regression": {"warn": True, "fail": True}},
            "attribution": attr,
        }]
        result = daily_digest(env="prod", service_reports=reports, policy=policy,
                               write_files=False)
        assert "The deploy event likely caused instability." in result["markdown"]