247 lines
9.3 KiB
Python
247 lines
9.3 KiB
Python
# #region Test.NotificationProviders [C:3] [TYPE Module] [SEMANTICS test,notification,provider,smtp,telegram,slack]
|
|
# @BRIEF Tests for notifications/providers.py — SMTP, Telegram, Slack providers with mocked network.
|
|
# @RELATION BINDS_TO -> [providers]
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
import pytest
|
|
|
|
|
|
class TestSMTPProvider:
|
|
"""SMTPProvider — delivers notifications via SMTP."""
|
|
|
|
@pytest.fixture
|
|
def config(self):
|
|
return {
|
|
"host": "smtp.example.com",
|
|
"port": 587,
|
|
"username": "user",
|
|
"password": "pass",
|
|
"from_email": "noreply@example.com",
|
|
"use_tls": True,
|
|
}
|
|
|
|
@pytest.fixture
|
|
def provider(self, config):
|
|
from src.services.notifications.providers import SMTPProvider
|
|
return SMTPProvider(config)
|
|
|
|
def test_init(self, provider):
|
|
assert provider.host == "smtp.example.com"
|
|
assert provider.port == 587
|
|
assert provider.from_email == "noreply@example.com"
|
|
assert provider.use_tls is True
|
|
|
|
def test_init_defaults(self):
|
|
from src.services.notifications.providers import SMTPProvider
|
|
p = SMTPProvider({"host": "smtp.x.com"})
|
|
assert p.port == 587
|
|
assert p.use_tls is True
|
|
assert p.username is None
|
|
assert p.password is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_success(self, provider):
|
|
mock_smtp = AsyncMock()
|
|
mock_smtp.__aenter__.return_value = mock_smtp
|
|
|
|
with patch("src.services.notifications.providers.aiosmtplib.SMTP", return_value=mock_smtp):
|
|
result = await provider.send("admin@x.com", "Test", "Body")
|
|
assert result is True
|
|
mock_smtp.sendmail.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_failure_returns_false(self, provider):
|
|
mock_smtp = AsyncMock()
|
|
mock_smtp.__aenter__.side_effect = ConnectionError("SMTP timeout")
|
|
|
|
with patch("src.services.notifications.providers.aiosmtplib.SMTP", return_value=mock_smtp):
|
|
result = await provider.send("admin@x.com", "Test", "Body")
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_no_auth(self, config):
|
|
from src.services.notifications.providers import SMTPProvider
|
|
p = SMTPProvider({"host": "smtp.x.com", "username": None, "password": None})
|
|
mock_smtp = AsyncMock()
|
|
mock_smtp.__aenter__.return_value = mock_smtp
|
|
|
|
with patch("src.services.notifications.providers.aiosmtplib.SMTP", return_value=mock_smtp):
|
|
result = await p.send("admin@x.com", "Test", "Body")
|
|
assert result is True
|
|
mock_smtp.login.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_no_tls(self, config):
|
|
from src.services.notifications.providers import SMTPProvider
|
|
p = SMTPProvider({"host": "smtp.x.com", "use_tls": False})
|
|
mock_smtp = AsyncMock()
|
|
mock_smtp.__aenter__.return_value = mock_smtp
|
|
|
|
with patch("src.services.notifications.providers.aiosmtplib.SMTP", return_value=mock_smtp):
|
|
result = await p.send("admin@x.com", "Test", "Body")
|
|
assert result is True
|
|
mock_smtp.starttls.assert_not_called()
|
|
|
|
|
|
class TestTelegramProvider:
|
|
"""TelegramProvider — delivers via Telegram Bot API."""
|
|
|
|
@pytest.fixture
|
|
def config(self):
|
|
return {"bot_token": "123:ABC"}
|
|
|
|
@pytest.fixture
|
|
def provider(self, config):
|
|
from src.services.notifications.providers import TelegramProvider
|
|
return TelegramProvider(config)
|
|
|
|
def test_init(self, provider):
|
|
assert provider.bot_token == "123:ABC"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_success(self, provider):
|
|
mock_client = AsyncMock()
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status.return_value = None
|
|
mock_client.post.return_value = mock_response
|
|
|
|
with patch("src.services.notifications.providers._get_http_client",
|
|
return_value=mock_client):
|
|
result = await provider.send("@admin", "Alert", "DB issue")
|
|
assert result is True
|
|
mock_client.post.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_no_token_returns_false(self):
|
|
from src.services.notifications.providers import TelegramProvider
|
|
p = TelegramProvider({"bot_token": None})
|
|
result = await p.send("@admin", "Alert", "Body")
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_failure_returns_false(self, provider):
|
|
mock_client = AsyncMock()
|
|
mock_client.post.side_effect = ConnectionError("API timeout")
|
|
|
|
with patch("src.services.notifications.providers._get_http_client",
|
|
return_value=mock_client):
|
|
result = await provider.send("@admin", "Alert", "Body")
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_http_error_returns_false(self, provider):
|
|
mock_client = AsyncMock()
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status.side_effect = ConnectionError("HTTP 500")
|
|
mock_client.post.return_value = mock_response
|
|
|
|
with patch("src.services.notifications.providers._get_http_client",
|
|
return_value=mock_client):
|
|
result = await provider.send("@admin", "Alert", "Body")
|
|
assert result is False
|
|
|
|
|
|
class TestSlackProvider:
|
|
"""SlackProvider — delivers via Slack Webhooks."""
|
|
|
|
@pytest.fixture
|
|
def config(self):
|
|
return {"webhook_url": "https://hooks.slack.com/services/xxx"}
|
|
|
|
@pytest.fixture
|
|
def provider(self, config):
|
|
from src.services.notifications.providers import SlackProvider
|
|
return SlackProvider(config)
|
|
|
|
def test_init(self, provider):
|
|
assert provider.webhook_url == "https://hooks.slack.com/services/xxx"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_success(self, provider):
|
|
mock_client = AsyncMock()
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status.return_value = None
|
|
mock_client.post.return_value = mock_response
|
|
|
|
with patch("src.services.notifications.providers._get_http_client",
|
|
return_value=mock_client):
|
|
result = await provider.send("#alerts", "Alert", "DB issue")
|
|
assert result is True
|
|
mock_client.post.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_no_webhook_returns_false(self):
|
|
from src.services.notifications.providers import SlackProvider
|
|
p = SlackProvider({"webhook_url": None})
|
|
result = await p.send("#alerts", "Alert", "Body")
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_failure_returns_false(self, provider):
|
|
mock_client = AsyncMock()
|
|
mock_client.post.side_effect = ConnectionError("Webhook timeout")
|
|
|
|
with patch("src.services.notifications.providers._get_http_client",
|
|
return_value=mock_client):
|
|
result = await provider.send("#alerts", "Alert", "Body")
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_http_error_returns_false(self, provider):
|
|
mock_client = AsyncMock()
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status.side_effect = ConnectionError("HTTP 500")
|
|
mock_client.post.return_value = mock_response
|
|
|
|
with patch("src.services.notifications.providers._get_http_client",
|
|
return_value=mock_client):
|
|
result = await provider.send("#alerts", "Alert", "Body")
|
|
assert result is False
|
|
|
|
class TestGetHttpClient:
|
|
"""_get_http_client — singleton client creation (lines 41-46)."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_creates_and_reuses_client(self):
|
|
from src.services.notifications.providers import _get_http_client, _http_client
|
|
# Reset singleton
|
|
import src.services.notifications.providers as providers_mod
|
|
providers_mod._http_client = None
|
|
|
|
client1 = await _get_http_client()
|
|
assert client1 is not None
|
|
client2 = await _get_http_client()
|
|
assert client2 is client1 # same instance reused
|
|
|
|
# Cleanup
|
|
await client1.aclose()
|
|
providers_mod._http_client = None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_has_timeout(self):
|
|
from src.services.notifications.providers import _get_http_client
|
|
import src.services.notifications.providers as providers_mod
|
|
providers_mod._http_client = None
|
|
|
|
client = await _get_http_client()
|
|
assert client.timeout is not None
|
|
await client.aclose()
|
|
providers_mod._http_client = None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_abstract_send_has_pass(self):
|
|
"""Cover abstract method pass at line 72."""
|
|
from src.services.notifications.providers import NotificationProvider
|
|
# Patch __abstractmethods__ to bypass ABC instantiation check
|
|
import unittest.mock as mock
|
|
with mock.patch.object(NotificationProvider, '__abstractmethods__', set()):
|
|
p = NotificationProvider()
|
|
result = await p.send("target", "subject", "body", {})
|
|
assert result is None # pass returns None
|
|
# #endregion Test.NotificationProviders
|