""" tests/test_release_check_risk_watch.py — Unit tests for risk_watch release gate. Tests: - warn mode: gate passes but adds recommendations when score >= warn_at - strict mode: gate fails when score >= fail_at for p0_services - non-fatal error: skipped gracefully, never blocks release """ import asyncio import pytest import sys from pathlib import Path from unittest.mock import AsyncMock, MagicMock sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "services" / "router")) # ─── Helpers ───────────────────────────────────────────────────────────────── def _make_risk_report(service, score, band=None, reasons=None, recs=None, warn_at=50, fail_at=80): """Build a minimal RiskReport dict matching risk_engine output.""" 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, "thresholds": {"warn_at": warn_at, "fail_at": fail_at}, "components": { "open_incidents": {"P0": 0, "P1": 0, "points": 0}, "recurrence": {"points": 0}, "followups": {"points": 0}, "slo": {"violations": 0, "points": 0}, "alerts_loop": {"violations": 0, "points": 0}, "escalations": {"count_24h": 0, "points": 0}, }, "reasons": reasons or [], "recommendations": recs or [], "updated_at": "2026-02-23T00:00:00", } def _make_tool_manager(score, service="gateway", warn_at=50, fail_at=80, fail_execute=False): """Stub ToolManager that returns a pre-built RiskReport.""" tm = MagicMock() if fail_execute: tm.execute_tool = AsyncMock(side_effect=RuntimeError("connection timeout")) else: result = MagicMock() result.success = True result.error = None result.result = _make_risk_report(service, score, warn_at=warn_at, fail_at=fail_at) tm.execute_tool = AsyncMock(return_value=result) return tm # ─── Import the helper directly ────────────────────────────────────────────── from release_check_runner import _run_risk_watch # ─── Warn mode tests ───────────────────────────────────────────────────────── class TestRiskWatchWarnMode: def test_score_below_warn_at_is_clean(self): """Score 30 < warn_at 50 — gate passes, no recommendations.""" tm = _make_tool_manager(score=30) ok, gate = asyncio.run( _run_risk_watch(tm, "ops", service_name="gateway", env="prod") ) assert ok is True assert gate["status"] == "pass" assert not gate.get("skipped") assert gate.get("recommendations", []) == [] def test_score_at_warn_at_adds_recommendation(self): """Score 50 == warn_at 50 — passes but includes recommendations.""" tm = _make_tool_manager(score=50, service="gateway") ok, gate = asyncio.run( _run_risk_watch(tm, "ops", service_name="gateway", env="prod") ) assert ok is True assert gate["status"] == "pass" assert len(gate.get("recommendations", [])) > 0 def test_score_above_warn_at_still_passes_in_warn_mode(self): """In warn mode the gate always passes (overall_pass is controlled by caller).""" tm = _make_tool_manager(score=75) ok, gate = asyncio.run( _run_risk_watch(tm, "ops", service_name="gateway", env="prod") ) # _run_risk_watch itself always returns ok=True; caller drives strict logic assert ok is True assert gate["score"] == 75 assert gate["band"] in ("high", "critical") def test_warn_threshold_override(self): """Caller can override warn_at via parameter.""" tm = _make_tool_manager(score=40, warn_at=30) ok, gate = asyncio.run( _run_risk_watch(tm, "ops", service_name="gateway", env="prod", warn_at=30) ) assert gate["effective_warn_at"] == 30 assert gate["score"] == 40 # >= 30, so recommendations should fire # ─── Strict mode tests ─────────────────────────────────────────────────────── class TestRiskWatchStrictMode: def test_score_above_fail_at_should_be_caught_by_caller(self): """ _run_risk_watch returns the gate data; the caller (release_check_runner) applies strict-mode logic. Verify effective_fail_at is correct. """ tm = _make_tool_manager(score=85, fail_at=80, service="gateway") ok, gate = asyncio.run( _run_risk_watch(tm, "ops", service_name="gateway", env="prod") ) assert gate["score"] == 85 assert gate["effective_fail_at"] == 80 # caller would check: score >= effective_fail_at → block in strict mode assert gate["score"] >= gate["effective_fail_at"] def test_fail_threshold_override(self): """Caller-supplied fail_at overrides policy value.""" tm = _make_tool_manager(score=70, fail_at=80, service="gateway") ok, gate = asyncio.run( _run_risk_watch(tm, "ops", service_name="gateway", env="staging", fail_at=65) ) assert gate["effective_fail_at"] == 65 # override in effect assert gate["score"] >= gate["effective_fail_at"] def test_score_below_fail_at_is_safe(self): tm = _make_tool_manager(score=60, fail_at=80, service="gateway") ok, gate = asyncio.run( _run_risk_watch(tm, "ops", service_name="gateway", env="staging") ) assert gate["score"] < gate["effective_fail_at"] # would not block # ─── Non-fatal error tests ──────────────────────────────────────────────────── class TestRiskWatchNonFatal: def test_tool_error_returns_skip(self): """When risk_engine_tool raises, gate is skipped and ok=True.""" tm = _make_tool_manager(score=0, fail_execute=True) ok, gate = asyncio.run( _run_risk_watch(tm, "ops", service_name="gateway", env="prod") ) assert ok is True assert gate.get("skipped") is True assert gate["status"] == "pass" def test_tool_failure_result_returns_skip(self): """When tool result.success=False, gate is skipped.""" tm = MagicMock() result = MagicMock() result.success = False result.error = "tool unavailable" result.result = None tm.execute_tool = AsyncMock(return_value=result) ok, gate = asyncio.run( _run_risk_watch(tm, "ops", service_name="gateway", env="prod") ) assert ok is True assert gate.get("skipped") is True def test_no_service_name_returns_skip(self): """Missing service_name → skip, no calls made.""" tm = MagicMock() tm.execute_tool = AsyncMock() ok, gate = asyncio.run( _run_risk_watch(tm, "ops", service_name="", env="prod") ) assert ok is True assert gate.get("skipped") is True tm.execute_tool.assert_not_called()