test(p3.2): provider_version_alerter 單元測試(6 passed)
Some checks failed
CD Pipeline / build-and-deploy (push) Has been cancelled

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Your Name
2026-04-27 19:56:47 +08:00
parent ae5e33d254
commit b6e4e87e57

View File

@@ -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 dedupollama 未 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"] == []