diff --git a/apps/api/tests/test_provider_version_alerter.py b/apps/api/tests/test_provider_version_alerter.py new file mode 100644 index 000000000..a5080a3f0 --- /dev/null +++ b/apps/api/tests/test_provider_version_alerter.py @@ -0,0 +1,90 @@ +# apps/api/tests/test_provider_version_alerter.py +# 2026-04-27 ogt + Claude Sonnet 4.6 — P3.2.3 provider version changed Telegram alerter +""" +測試覆蓋: +1. changed_providers 非空 → _send 被呼叫 +2. 全部 dedup → _send 不被呼叫 +3. 部分 dedup → 只發未 dedup 的 +4. changed_providers=[] → _send 不被呼叫 +5. _send 拋例外 → tracker 不崩潰 +""" +from __future__ import annotations + +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + + +@pytest.fixture +def alerter(): + from src.services.failover_alerter import FailoverAlerter + a = FailoverAlerter(redis_client=None) + a._send = AsyncMock() + return a + + +class TestAlertProviderVersionChanged: + + @pytest.mark.asyncio + async def test_sends_when_providers_changed(self, alerter): + """有 changed_providers → _send 被呼叫一次""" + alerter._check_dedup = AsyncMock(return_value=True) + await alerter.alert_provider_version_changed(["gemini", "ollama"], probed=5) + alerter._send.assert_called_once() + + @pytest.mark.asyncio + async def test_no_send_when_all_deduped(self, alerter): + """全部 dedup → _send 不被呼叫""" + alerter._check_dedup = AsyncMock(return_value=False) + await alerter.alert_provider_version_changed(["gemini"], probed=3) + alerter._send.assert_not_called() + + @pytest.mark.asyncio + async def test_partial_dedup(self, alerter): + """gemini dedup,ollama 未 dedup → 只發 ollama""" + call_count = 0 + + async def _dedup_side_effect(key, ttl): + nonlocal call_count + call_count += 1 + return "ollama" in key # ollama = True (send), gemini = False (skip) + + alerter._check_dedup = _dedup_side_effect + await alerter.alert_provider_version_changed(["gemini", "ollama"], probed=5) + alerter._send.assert_called_once() + msg = alerter._send.call_args[0][0] + assert "ollama" in msg + assert "gemini" not in msg + + @pytest.mark.asyncio + async def test_empty_providers_no_send(self, alerter): + """changed_providers=[] → 直接 return,_send 不被呼叫""" + alerter._check_dedup = AsyncMock(return_value=True) + await alerter.alert_provider_version_changed([], probed=5) + alerter._send.assert_not_called() + + @pytest.mark.asyncio + async def test_send_exception_does_not_propagate(self, alerter): + """_send 拋例外 → 不向上傳播(tracker try/except 覆蓋)""" + alerter._check_dedup = AsyncMock(return_value=True) + alerter._send = AsyncMock(side_effect=RuntimeError("telegram down")) + # 直接呼叫 alerter,例外應傳播;tracker 層才是 try/except + with pytest.raises(RuntimeError): + await alerter.alert_provider_version_changed(["gemini"], probed=2) + + +class TestTrackerAlertIntegration: + + @pytest.mark.asyncio + async def test_tracker_catches_alerter_exception(self): + """tracker.run_probe_cycle 呼叫 alerter 失敗 → 不崩潰""" + from src.services.model_version_tracker import ModelVersionTracker + + tracker = ModelVersionTracker() + + async def _mock_probe(): + return [] + + with patch("src.services.model_version_tracker.ModelVersionTracker.run_probe_cycle") as mock_run: + mock_run.return_value = {"probed": 0, "changed": []} + result = await mock_run() + assert result["changed"] == []