""" tests/test_risk_digest.py — Unit tests for risk_digest.daily_digest. Tests: - JSON and markdown content generated correctly - Markdown clamped to max_chars - Top risky services and regressions included - Action list deterministic from report state """ import datetime import sys import tempfile from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "services" / "router")) from risk_digest import daily_digest, _build_action_list, _clamp 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(service, score, band=None, delta_24h=None, delta_7d=None, reg_warn=False, reg_fail=False, overdue_p1=0, slo_violations=0): from risk_engine import score_to_band, _builtin_defaults p = _builtin_defaults() b = band or score_to_band(score, p) return { "service": service, "env": "prod", "score": score, "band": b, "components": { "followups": {"P0": 0, "P1": overdue_p1, "other": 0, "points": overdue_p1 * 12}, "slo": {"violations": slo_violations, "points": slo_violations * 10}, }, "reasons": [], "recommendations": [], "updated_at": "2026-02-23T00:00:00", "trend": { "delta_24h": delta_24h, "delta_7d": delta_7d, "slope_per_day": None, "volatility": None, "regression": {"warn": reg_warn, "fail": reg_fail}, }, } # ─── _clamp ─────────────────────────────────────────────────────────────────── class TestClamp: def test_no_clamp_if_short(self): text = "hello world" assert _clamp(text, 100) == text def test_clamps_to_max_chars(self): text = "x" * 500 result = _clamp(text, 100) assert len(result) <= 200 # includes truncation notice assert "truncated" in result def test_exact_limit_not_clamped(self): text = "a" * 100 assert _clamp(text, 100) == text # ─── _build_action_list ─────────────────────────────────────────────────────── class TestActionList: def test_no_actions_for_low_risk(self): reports = [_report("svc", 5, band="low")] actions = _build_action_list(reports) assert actions == [] def test_critical_band_generates_oncall_action(self): reports = [_report("gateway", 90, band="critical")] actions = _build_action_list(reports) assert any("Critical" in a or "critical" in a.lower() for a in actions) def test_high_band_generates_coordinate_action(self): reports = [_report("router", 60, band="high")] actions = _build_action_list(reports) assert any("oncall" in a.lower() or "High" in a for a in actions) def test_regression_fail_generates_freeze_action(self): reports = [_report("gateway", 70, band="high", delta_24h=25, reg_fail=True)] actions = _build_action_list(reports) assert any("Regression" in a or "Freeze" in a for a in actions) def test_overdue_followups_action(self): reports = [_report("router", 40, band="medium", overdue_p1=2)] actions = _build_action_list(reports) assert any("follow-up" in a.lower() or "overdue" in a.lower() for a in actions) def test_slo_violation_action(self): reports = [_report("router", 40, band="medium", slo_violations=1)] actions = _build_action_list(reports) assert any("SLO" in a for a in actions) # ─── daily_digest (no file write) ──────────────────────────────────────────── class TestDailyDigest: def test_returns_json_and_markdown(self, policy): reports = [ _report("gateway", 80, band="high", delta_24h=15, reg_warn=True), _report("router", 30, band="medium"), ] result = daily_digest( env="prod", service_reports=reports, policy=policy, date_str="2026-02-23", write_files=False, ) assert result["date"] == "2026-02-23" assert result["env"] == "prod" assert "markdown" in result assert "json_data" in result assert result["json_path"] is None # not written def test_markdown_contains_service_name(self, policy): reports = [_report("gateway", 80, band="high")] result = daily_digest(env="prod", service_reports=reports, policy=policy, write_files=False) assert "gateway" in result["markdown"] def test_markdown_contains_band_summary(self, policy): reports = [ _report("a", 85, band="critical"), _report("b", 60, band="high"), ] result = daily_digest(env="prod", service_reports=reports, policy=policy, write_files=False) assert "critical" in result["markdown"].lower() assert "high" in result["markdown"].lower() def test_markdown_clamped_to_max_chars(self, policy): """max_chars in builtin_defaults is 8000; generate a large report.""" reports = [_report(f"service-{i:03d}", 80 - i, band="high") for i in range(50)] modified_policy = {**policy, "digest": {**policy.get("digest", {}), "markdown_max_chars": 200}} result = daily_digest(env="prod", service_reports=reports, policy=modified_policy, write_files=False) assert len(result["markdown"]) <= 400 # clamped + truncation note def test_top_regressions_in_json(self, policy): reports = [ _report("gateway", 80, delta_24h=25, reg_fail=True), _report("router", 50, delta_24h=5), # both > 0: both appear ] result = daily_digest(env="prod", service_reports=reports, policy=policy, write_files=False) top_reg = result["json_data"]["top_regressions"] # gateway has the highest delta assert top_reg[0]["service"] == "gateway" assert top_reg[0]["delta_24h"] == 25 # router also included (delta > 0) assert any(r["service"] == "router" for r in top_reg) def test_improving_services_in_json(self, policy): reports = [ _report("gateway", 30, delta_7d=-20), _report("router", 50, delta_7d=5), ] result = daily_digest(env="prod", service_reports=reports, policy=policy, write_files=False) improving = result["json_data"]["improving_services"] assert any(s["service"] == "gateway" for s in improving) def test_files_written_to_tempdir(self, policy, tmp_path): reports = [_report("gateway", 60, band="high")] result = daily_digest( env="prod", service_reports=reports, policy=policy, date_str="2026-02-23", output_dir=str(tmp_path), write_files=True, ) assert result["json_path"] is not None assert Path(result["json_path"]).exists() assert result["md_path"] is not None assert Path(result["md_path"]).exists() def test_empty_reports_does_not_crash(self, policy): result = daily_digest(env="prod", service_reports=[], policy=policy, write_files=False) assert result["date"] is not None assert "markdown" in result def test_top_n_limits_services(self, policy): reports = [_report(f"svc{i}", 100 - i) for i in range(15)] modified = {**policy, "digest": {**policy.get("digest", {}), "top_n": 5}} result = daily_digest(env="prod", service_reports=reports, policy=modified, write_files=False) assert len(result["json_data"]["top_services"]) == 5