| 1 | """Notifier 单元测试。""" |
| 2 | |
| 3 | from typing import Any, Dict, List |
| 4 | |
| 5 | import pytest |
| 6 | |
| 7 | from utils.notifier import ( |
| 8 | BarkProvider, |
| 9 | Notifier, |
| 10 | TelegramProvider, |
| 11 | WebhookProvider, |
| 12 | build_notifier, |
| 13 | ) |
| 14 | |
| 15 | |
| 16 | class _FakeResponse: |
| 17 | def __init__(self, status: int = 200): |
| 18 | self.status = status |
| 19 | |
| 20 | async def __aenter__(self): |
| 21 | return self |
| 22 | |
| 23 | async def __aexit__(self, exc_type, exc, tb): |
| 24 | return False |
| 25 | |
| 26 | |
| 27 | class _FakeSession: |
| 28 | def __init__(self, status: int = 200): |
| 29 | self.status = status |
| 30 | self.calls: List[Dict[str, Any]] = [] |
| 31 | |
| 32 | def get(self, url, params=None): |
| 33 | self.calls.append({"method": "GET", "url": url, "params": params}) |
| 34 | return _FakeResponse(self.status) |
| 35 | |
| 36 | def post(self, url, json=None, headers=None): |
| 37 | self.calls.append({"method": "POST", "url": url, "json": json, "headers": headers}) |
| 38 | return _FakeResponse(self.status) |
| 39 | |
| 40 | |
| 41 | @pytest.mark.asyncio |
| 42 | async def test_bark_provider_sends_request(): |
| 43 | provider = BarkProvider({"url": "https://api.day.app/KEY", "sound": "bell"}) |
| 44 | session = _FakeSession() |
| 45 | ok = await provider.send(session, title="t", body="b", level="success") |
| 46 | assert ok is True |
| 47 | assert len(session.calls) == 1 |
| 48 | call = session.calls[0] |
| 49 | assert call["method"] == "GET" |
| 50 | assert call["url"].startswith("https://api.day.app/KEY/") |
| 51 | assert call["params"] == {"sound": "bell"} |
| 52 | |
| 53 | |
| 54 | @pytest.mark.asyncio |
| 55 | async def test_bark_provider_skips_without_url(): |
| 56 | provider = BarkProvider({}) |
| 57 | session = _FakeSession() |
| 58 | ok = await provider.send(session, title="t", body="b", level="info") |
| 59 | assert ok is False |
| 60 | assert session.calls == [] |
| 61 | |
| 62 | |
| 63 | @pytest.mark.asyncio |
| 64 | async def test_telegram_provider_uses_bot_api(): |
| 65 | provider = TelegramProvider({"bot_token": "abc", "chat_id": "42"}) |
| 66 | session = _FakeSession() |
| 67 | ok = await provider.send(session, title="t", body="b", level="info") |
| 68 | assert ok is True |
| 69 | call = session.calls[0] |
| 70 | assert call["url"] == "https://api.telegram.org/botabc/sendMessage" |
| 71 | assert call["json"]["chat_id"] == "42" |
| 72 | assert "t" in call["json"]["text"] |
| 73 | assert "b" in call["json"]["text"] |
| 74 | |
| 75 | |
| 76 | @pytest.mark.asyncio |
| 77 | async def test_webhook_provider_posts_json(): |
| 78 | provider = WebhookProvider( |
| 79 | { |
| 80 | "url": "https://hook.example/endpoint", |
| 81 | "headers": {"Authorization": "Bearer xyz"}, |
| 82 | "extra_body": {"source": "dy"}, |
| 83 | } |
| 84 | ) |
| 85 | session = _FakeSession() |
| 86 | ok = await provider.send(session, title="t", body="b", level="success") |
| 87 | assert ok is True |
| 88 | call = session.calls[0] |
| 89 | assert call["url"] == "https://hook.example/endpoint" |
| 90 | assert call["headers"] == {"Authorization": "Bearer xyz"} |
| 91 | assert call["json"]["title"] == "t" |
| 92 | assert call["json"]["body"] == "b" |
| 93 | assert call["json"]["level"] == "success" |
| 94 | assert call["json"]["source"] == "dy" |
| 95 | |
| 96 | |
| 97 | @pytest.mark.asyncio |
| 98 | async def test_webhook_provider_reports_failure_on_4xx(): |
| 99 | provider = WebhookProvider({"url": "https://hook.example"}) |
| 100 | session = _FakeSession(status=500) |
| 101 | ok = await provider.send(session, title="t", body="b", level="info") |
| 102 | assert ok is False |
| 103 | |
| 104 | |
| 105 | def test_build_notifier_disabled_returns_empty(): |
| 106 | notifier = build_notifier({"notifications": {"enabled": False, "providers": []}}) |
| 107 | assert notifier.enabled is False |
| 108 | |
| 109 | |
| 110 | def test_build_notifier_rejects_scalar_config(): |
| 111 | """用户误写 `notifications: on` 等 scalar 不应抛 AttributeError。""" |
| 112 | notifier = build_notifier({"notifications": "on"}) |
| 113 | assert notifier.enabled is False |
| 114 | notifier = build_notifier({"notifications": True}) |
| 115 | assert notifier.enabled is False |
| 116 | notifier = build_notifier({"notifications": 42}) |
| 117 | assert notifier.enabled is False |
| 118 | |
| 119 | |
| 120 | def test_build_notifier_ignores_unknown_provider(): |
| 121 | notifier = build_notifier( |
| 122 | { |
| 123 | "notifications": { |
| 124 | "enabled": True, |
| 125 | "providers": [ |
| 126 | {"type": "unknown"}, |
| 127 | {"type": "bark", "url": "https://api.day.app/KEY"}, |
| 128 | ], |
| 129 | } |
| 130 | } |
| 131 | ) |
| 132 | assert notifier.enabled is True |
| 133 | assert len(notifier.providers) == 1 |
| 134 | assert isinstance(notifier.providers[0], BarkProvider) |
| 135 | |
| 136 | |
| 137 | @pytest.mark.asyncio |
| 138 | async def test_notifier_respects_on_success_flag(): |
| 139 | # on_success=False 时 success 级别不应分发 |
| 140 | notifier = Notifier( |
| 141 | providers=[BarkProvider({"url": "https://api.day.app/KEY"})], |
| 142 | on_success=False, |
| 143 | on_failure=True, |
| 144 | ) |
| 145 | result = await notifier.send(title="t", body="b", level="success") |
| 146 | assert result == {} |
| 147 | |
| 148 | |
| 149 | @pytest.mark.asyncio |
| 150 | async def test_notifier_empty_providers_returns_empty(): |
| 151 | notifier = Notifier(providers=[]) |
| 152 | result = await notifier.send(title="t", body="b", level="info") |
| 153 | assert result == {} |
| 154 | assert notifier.enabled is False |
| 155 |