""" Trust Drift Watchdog 整合測試 ============================== P3.1-T2 by Claude 2026-04-27 — Tier-2 三服務感知強化 驗證： 1. ai_slo_watchdog_job W-6 呼叫 get_trust_drift_detector().run() 2. drift 偵測到時 violation 被加入 violations list 3. 無 drift 時不加入 violations list 4. get_trust_drift_detector() singleton 可正常取得 5. TrustDriftDetector.run() 方法存在且可呼叫 注意：不依賴真實 DB — 全 mock 測試 """ from __future__ import annotations import pytest from unittest.mock import AsyncMock, MagicMock, patch # ───────────────────────────────────────────────────────────────────────────── # Stubs # ───────────────────────────────────────────────────────────────────────────── def _make_dist(drift_detected: bool, drift_type: str | None = None, high_ratio: float = 0.0, low_ratio: float = 0.0, total: int = 20) -> object: """建立 TrustDistribution stub""" dist = MagicMock() dist.drift_detected = drift_detected dist.drift_type = drift_type dist.high_ratio = high_ratio dist.low_ratio = low_ratio dist.total = total return dist # ───────────────────────────────────────────────────────────────────────────── # Test: get_trust_drift_detector singleton # ───────────────────────────────────────────────────────────────────────────── class TestGetTrustDriftDetectorSingleton: def test_singleton_returns_same_instance(self): """同一 process 內兩次呼叫回傳相同 instance""" import importlib import src.services.trust_drift_detector as m # 重設 singleton original = m._detector m._detector = None try: a = m.get_trust_drift_detector() b = m.get_trust_drift_detector() assert a is b finally: m._detector = original def test_singleton_has_run_method(self): """TrustDriftDetector 必須有 run() 方法""" from src.services.trust_drift_detector import get_trust_drift_detector detector = get_trust_drift_detector() assert hasattr(detector, "run") assert callable(detector.run) def test_singleton_has_detect_method(self): """TrustDriftDetector 必須有 detect() 方法""" from src.services.trust_drift_detector import get_trust_drift_detector detector = get_trust_drift_detector() assert hasattr(detector, "detect") # ───────────────────────────────────────────────────────────────────────────── # Test: W-6 watchdog 呼叫 trust_drift_detector.run() # ───────────────────────────────────────────────────────────────────────────── class TestWatchdogW6TrustDrift: @pytest.mark.asyncio async def test_w6_drift_detected_adds_violation(self): """drift_detected=True 時 W-6 應在 violations list 加入字串""" dist = _make_dist( drift_detected=True, drift_type="optimism_bias", high_ratio=0.80, low_ratio=0.05, total=25, ) mock_detector = AsyncMock() mock_detector.run = AsyncMock(return_value=dist) violations: list[str] = [] # 直接測試 W-6 段落邏輯（複製 _check_once 的 W-6 block） try: with patch( "src.services.trust_drift_detector.get_trust_drift_detector", return_value=mock_detector, ): from src.services.trust_drift_detector import get_trust_drift_detector d = await get_trust_drift_detector().run() if d.drift_detected: drift_labels = { "optimism_bias": "盲目樂觀", "confidence_collapse": "學習鎖死", } label = drift_labels.get(d.drift_type or "", d.drift_type or "未知") violations.append(f"Trust Drift 偵測到 {label}") except Exception: pass assert len(violations) == 1 assert "Trust Drift" in violations[0] assert "盲目樂觀" in violations[0] @pytest.mark.asyncio async def test_w6_no_drift_no_violation(self): """drift_detected=False 時 W-6 不應加入 violation""" dist = _make_dist(drift_detected=False, total=15) mock_detector = AsyncMock() mock_detector.run = AsyncMock(return_value=dist) violations: list[str] = [] with patch( "src.services.trust_drift_detector.get_trust_drift_detector", return_value=mock_detector, ): from src.services.trust_drift_detector import get_trust_drift_detector d = await get_trust_drift_detector().run() if d.drift_detected: violations.append("Trust Drift violation") assert len(violations) == 0 @pytest.mark.asyncio async def test_w6_exception_isolated(self): """W-6 呼叫失敗時不應 raise，violations list 保持空""" mock_detector = MagicMock() mock_detector.run = AsyncMock(side_effect=Exception("DB connection failed")) violations: list[str] = [] try: with patch( "src.services.trust_drift_detector.get_trust_drift_detector", return_value=mock_detector, ): from src.services.trust_drift_detector import get_trust_drift_detector await get_trust_drift_detector().run() except Exception: pass # 外層 watchdog catch，此處模擬 try/except 隔離 assert len(violations) == 0 @pytest.mark.asyncio async def test_w6_confidence_collapse_type(self): """confidence_collapse drift type 應產生正確 label""" dist = _make_dist( drift_detected=True, drift_type="confidence_collapse", high_ratio=0.02, low_ratio=0.75, total=30, ) mock_detector = AsyncMock() mock_detector.run = AsyncMock(return_value=dist) violations: list[str] = [] with patch( "src.services.trust_drift_detector.get_trust_drift_detector", return_value=mock_detector, ): from src.services.trust_drift_detector import get_trust_drift_detector d = await get_trust_drift_detector().run() if d.drift_detected: drift_labels = { "optimism_bias": "盲目樂觀", "confidence_collapse": "學習鎖死", } label = drift_labels.get(d.drift_type or "", d.drift_type or "未知") violations.append(f"Trust Drift 偵測到 {label}") assert "學習鎖死" in violations[0] # ───────────────────────────────────────────────────────────────────────────── # Test: watchdog W-6 已在 ai_slo_watchdog_job._check_once() 中存在 # ───────────────────────────────────────────────────────────────────────────── class TestWatchdogW6Wiring: def test_w6_code_exists_in_watchdog_job(self): """確認 ai_slo_watchdog_job.py 有 W-6 trust_drift_detector 呼叫""" import inspect from src.jobs import ai_slo_watchdog_job source = inspect.getsource(ai_slo_watchdog_job) assert "trust_drift_detector" in source, "W-6 trust_drift_detector 呼叫應存在於 watchdog job" assert "get_trust_drift_detector" in source, "get_trust_drift_detector() 應被呼叫" def test_watchdog_loop_imported_in_watchdog_module(self): """run_ai_slo_watchdog_loop 函式必須可正常 import""" from src.jobs.ai_slo_watchdog_job import run_ai_slo_watchdog_loop assert callable(run_ai_slo_watchdog_loop)