432 lines
15 KiB
Python
432 lines
15 KiB
Python
"""
|
|
Tests unitaires pour le service de notification ntfy.
|
|
|
|
Ces tests vérifient le comportement du service sans nécessiter
|
|
un serveur ntfy réel (utilisation de mocks).
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch, MagicMock
|
|
import httpx
|
|
|
|
# Import des modules à tester
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Ajouter le répertoire app au path pour les imports
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "app"))
|
|
|
|
from schemas.notification import (
|
|
NtfyConfig,
|
|
NtfyAction,
|
|
NotificationRequest,
|
|
NotificationResponse,
|
|
NotificationTemplates,
|
|
)
|
|
from services.notification_service import NotificationService, send_notification
|
|
|
|
|
|
class TestNtfyConfig:
|
|
"""Tests pour la configuration NtfyConfig."""
|
|
|
|
def test_default_config(self):
|
|
"""Test de la configuration par défaut."""
|
|
config = NtfyConfig()
|
|
assert config.base_url == "http://localhost:8150"
|
|
assert config.default_topic == "homelab-events"
|
|
assert config.enabled is True
|
|
assert config.timeout == 5
|
|
assert config.username is None
|
|
assert config.password is None
|
|
assert config.token is None
|
|
|
|
def test_custom_config(self):
|
|
"""Test d'une configuration personnalisée."""
|
|
config = NtfyConfig(
|
|
base_url="http://ntfy.example.com:8080",
|
|
default_topic="my-topic",
|
|
enabled=False,
|
|
timeout=10,
|
|
username="user",
|
|
password="pass",
|
|
)
|
|
assert config.base_url == "http://ntfy.example.com:8080"
|
|
assert config.default_topic == "my-topic"
|
|
assert config.enabled is False
|
|
assert config.timeout == 10
|
|
assert config.has_auth is True
|
|
|
|
def test_has_auth_with_token(self):
|
|
"""Test de has_auth avec un token."""
|
|
config = NtfyConfig(token="my-token")
|
|
assert config.has_auth is True
|
|
|
|
def test_has_auth_without_credentials(self):
|
|
"""Test de has_auth sans credentials."""
|
|
config = NtfyConfig()
|
|
assert config.has_auth is False
|
|
|
|
def test_has_auth_partial_credentials(self):
|
|
"""Test de has_auth avec credentials partiels."""
|
|
config = NtfyConfig(username="user")
|
|
assert config.has_auth is False
|
|
|
|
config = NtfyConfig(password="pass")
|
|
assert config.has_auth is False
|
|
|
|
@patch.dict("os.environ", {
|
|
"NTFY_BASE_URL": "http://test.local:9000",
|
|
"NTFY_DEFAULT_TOPIC": "test-topic",
|
|
"NTFY_ENABLED": "false",
|
|
"NTFY_TIMEOUT": "15",
|
|
})
|
|
def test_from_env(self):
|
|
"""Test du chargement depuis les variables d'environnement."""
|
|
config = NtfyConfig.from_env()
|
|
assert config.base_url == "http://test.local:9000"
|
|
assert config.default_topic == "test-topic"
|
|
assert config.enabled is False
|
|
assert config.timeout == 15
|
|
|
|
|
|
class TestNotificationTemplates:
|
|
"""Tests pour les templates de notification."""
|
|
|
|
def test_backup_success(self):
|
|
"""Test du template de backup réussi."""
|
|
req = NotificationTemplates.backup_success(
|
|
hostname="server.home",
|
|
duration="5m 30s",
|
|
size="1.2 GB"
|
|
)
|
|
assert req.topic == "homelab-backup"
|
|
assert "server.home" in req.message
|
|
assert "5m 30s" in req.message
|
|
assert "1.2 GB" in req.message
|
|
assert req.priority == 3
|
|
assert "white_check_mark" in req.tags
|
|
|
|
def test_backup_failed(self):
|
|
"""Test du template de backup échoué."""
|
|
req = NotificationTemplates.backup_failed(
|
|
hostname="server.home",
|
|
error="Connection timeout"
|
|
)
|
|
assert req.topic == "homelab-backup"
|
|
assert "server.home" in req.message
|
|
assert "Connection timeout" in req.message
|
|
assert req.priority == 5
|
|
assert "x" in req.tags
|
|
|
|
def test_bootstrap_started(self):
|
|
"""Test du template de bootstrap démarré."""
|
|
req = NotificationTemplates.bootstrap_started("new-host.home")
|
|
assert req.topic == "homelab-bootstrap"
|
|
assert "new-host.home" in req.message
|
|
assert "rocket" in req.tags
|
|
|
|
def test_health_status_down(self):
|
|
"""Test du template de changement d'état (down)."""
|
|
req = NotificationTemplates.health_status_changed(
|
|
hostname="server.home",
|
|
new_status="down",
|
|
details="SSH timeout"
|
|
)
|
|
assert req.topic == "homelab-health"
|
|
assert "DOWN" in req.title
|
|
assert req.priority == 5
|
|
assert "red_circle" in req.tags
|
|
|
|
def test_health_status_up(self):
|
|
"""Test du template de changement d'état (up)."""
|
|
req = NotificationTemplates.health_status_changed(
|
|
hostname="server.home",
|
|
new_status="up"
|
|
)
|
|
assert req.topic == "homelab-health"
|
|
assert "UP" in req.title
|
|
assert req.priority == 3
|
|
assert "green_circle" in req.tags
|
|
|
|
|
|
class TestNotificationService:
|
|
"""Tests pour le service de notification."""
|
|
|
|
@pytest.fixture
|
|
def service(self):
|
|
"""Crée un service avec une config de test."""
|
|
config = NtfyConfig(
|
|
base_url="http://test.local:8150",
|
|
default_topic="test-topic",
|
|
enabled=True,
|
|
timeout=5,
|
|
)
|
|
return NotificationService(config)
|
|
|
|
@pytest.fixture
|
|
def disabled_service(self):
|
|
"""Crée un service désactivé."""
|
|
config = NtfyConfig(enabled=False)
|
|
return NotificationService(config)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_disabled(self, disabled_service):
|
|
"""Test que send retourne True quand désactivé."""
|
|
result = await disabled_service.send(
|
|
message="Test message",
|
|
topic="test"
|
|
)
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_success(self, service):
|
|
"""Test d'envoi réussi."""
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.text = '{"id": "abc123"}'
|
|
|
|
with patch.object(service, '_get_client') as mock_get_client:
|
|
mock_client = AsyncMock()
|
|
mock_client.post = AsyncMock(return_value=mock_response)
|
|
mock_get_client.return_value = mock_client
|
|
|
|
result = await service.send(
|
|
message="Test message",
|
|
topic="test-topic",
|
|
title="Test Title",
|
|
priority=3,
|
|
tags=["test"]
|
|
)
|
|
|
|
assert result is True
|
|
mock_client.post.assert_called_once()
|
|
call_args = mock_client.post.call_args
|
|
assert "http://test.local:8150/test-topic" in str(call_args)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_failure(self, service):
|
|
"""Test d'envoi échoué (erreur HTTP)."""
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 500
|
|
mock_response.text = "Internal Server Error"
|
|
|
|
with patch.object(service, '_get_client') as mock_get_client:
|
|
mock_client = AsyncMock()
|
|
mock_client.post = AsyncMock(return_value=mock_response)
|
|
mock_get_client.return_value = mock_client
|
|
|
|
result = await service.send(message="Test message")
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_timeout(self, service):
|
|
"""Test de gestion du timeout."""
|
|
with patch.object(service, '_get_client') as mock_get_client:
|
|
mock_client = AsyncMock()
|
|
mock_client.post = AsyncMock(side_effect=httpx.TimeoutException("Timeout"))
|
|
mock_get_client.return_value = mock_client
|
|
|
|
result = await service.send(message="Test message")
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_connection_error(self, service):
|
|
"""Test de gestion d'erreur de connexion."""
|
|
with patch.object(service, '_get_client') as mock_get_client:
|
|
mock_client = AsyncMock()
|
|
mock_client.post = AsyncMock(side_effect=httpx.ConnectError("Connection refused"))
|
|
mock_get_client.return_value = mock_client
|
|
|
|
result = await service.send(message="Test message")
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_uses_default_topic(self, service):
|
|
"""Test que le topic par défaut est utilisé si non spécifié."""
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
|
|
with patch.object(service, '_get_client') as mock_get_client:
|
|
mock_client = AsyncMock()
|
|
mock_client.post = AsyncMock(return_value=mock_response)
|
|
mock_get_client.return_value = mock_client
|
|
|
|
await service.send(message="Test message")
|
|
|
|
call_args = mock_client.post.call_args
|
|
assert "test-topic" in str(call_args)
|
|
|
|
def test_build_headers_basic_auth(self):
|
|
"""Test de construction des headers avec auth Basic."""
|
|
config = NtfyConfig(username="user", password="pass")
|
|
service = NotificationService(config)
|
|
|
|
headers = service._build_auth_headers()
|
|
|
|
assert "Authorization" in headers
|
|
assert headers["Authorization"].startswith("Basic ")
|
|
|
|
def test_build_headers_bearer_auth(self):
|
|
"""Test de construction des headers avec auth Bearer."""
|
|
config = NtfyConfig(token="my-token")
|
|
service = NotificationService(config)
|
|
|
|
headers = service._build_auth_headers()
|
|
|
|
assert "Authorization" in headers
|
|
assert headers["Authorization"] == "Bearer my-token"
|
|
|
|
def test_build_headers_with_options(self, service):
|
|
"""Test de construction des headers avec options."""
|
|
headers = service._build_headers(
|
|
title="Test Title",
|
|
priority=5,
|
|
tags=["warning", "skull"],
|
|
click="http://example.com",
|
|
delay="30m"
|
|
)
|
|
|
|
assert headers["Title"] == "Test Title"
|
|
assert headers["Priority"] == "urgent"
|
|
assert headers["Tags"] == "warning,skull"
|
|
assert headers["Click"] == "http://example.com"
|
|
assert headers["Delay"] == "30m"
|
|
|
|
def test_reconfigure(self, service):
|
|
"""Test de reconfiguration du service."""
|
|
new_config = NtfyConfig(
|
|
base_url="http://new.local:9000",
|
|
enabled=False
|
|
)
|
|
|
|
service.reconfigure(new_config)
|
|
|
|
assert service.config.base_url == "http://new.local:9000"
|
|
assert service.enabled is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notify_backup_success(self, service):
|
|
"""Test de la méthode helper notify_backup_success."""
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
|
|
with patch.object(service, '_get_client') as mock_get_client:
|
|
mock_client = AsyncMock()
|
|
mock_client.post = AsyncMock(return_value=mock_response)
|
|
mock_get_client.return_value = mock_client
|
|
|
|
result = await service.notify_backup_success(
|
|
hostname="server.home",
|
|
duration="5m",
|
|
size="1GB"
|
|
)
|
|
|
|
assert result is True
|
|
call_args = mock_client.post.call_args
|
|
assert "homelab-backup" in str(call_args)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_request(self, service):
|
|
"""Test de send_request avec NotificationRequest."""
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
|
|
with patch.object(service, '_get_client') as mock_get_client:
|
|
mock_client = AsyncMock()
|
|
mock_client.post = AsyncMock(return_value=mock_response)
|
|
mock_get_client.return_value = mock_client
|
|
|
|
request = NotificationRequest(
|
|
topic="custom-topic",
|
|
message="Custom message",
|
|
title="Custom Title",
|
|
priority=4,
|
|
tags=["custom"]
|
|
)
|
|
|
|
response = await service.send_request(request)
|
|
|
|
assert isinstance(response, NotificationResponse)
|
|
assert response.success is True
|
|
assert response.topic == "custom-topic"
|
|
|
|
|
|
class TestNotificationRequest:
|
|
"""Tests pour le modèle NotificationRequest."""
|
|
|
|
def test_minimal_request(self):
|
|
"""Test d'une requête minimale."""
|
|
req = NotificationRequest(message="Hello")
|
|
assert req.message == "Hello"
|
|
assert req.topic is None
|
|
assert req.title is None
|
|
assert req.priority is None
|
|
|
|
def test_full_request(self):
|
|
"""Test d'une requête complète."""
|
|
req = NotificationRequest(
|
|
topic="my-topic",
|
|
message="Hello World",
|
|
title="Greeting",
|
|
priority=5,
|
|
tags=["wave", "robot"],
|
|
click="http://example.com",
|
|
delay="1h"
|
|
)
|
|
assert req.topic == "my-topic"
|
|
assert req.message == "Hello World"
|
|
assert req.title == "Greeting"
|
|
assert req.priority == 5
|
|
assert req.tags == ["wave", "robot"]
|
|
assert req.click == "http://example.com"
|
|
assert req.delay == "1h"
|
|
|
|
def test_priority_validation(self):
|
|
"""Test de la validation de la priorité."""
|
|
# Priorité valide
|
|
req = NotificationRequest(message="Test", priority=1)
|
|
assert req.priority == 1
|
|
|
|
req = NotificationRequest(message="Test", priority=5)
|
|
assert req.priority == 5
|
|
|
|
# Priorité invalide
|
|
with pytest.raises(ValueError):
|
|
NotificationRequest(message="Test", priority=0)
|
|
|
|
with pytest.raises(ValueError):
|
|
NotificationRequest(message="Test", priority=6)
|
|
|
|
|
|
class TestNtfyAction:
|
|
"""Tests pour le modèle NtfyAction."""
|
|
|
|
def test_view_action(self):
|
|
"""Test d'une action view."""
|
|
action = NtfyAction(
|
|
action="view",
|
|
label="Open Dashboard",
|
|
url="http://dashboard.local"
|
|
)
|
|
assert action.action == "view"
|
|
assert action.label == "Open Dashboard"
|
|
assert action.url == "http://dashboard.local"
|
|
|
|
def test_http_action(self):
|
|
"""Test d'une action HTTP."""
|
|
action = NtfyAction(
|
|
action="http",
|
|
label="Restart Service",
|
|
url="http://api.local/restart",
|
|
method="POST",
|
|
headers={"Authorization": "Bearer token"},
|
|
body='{"service": "nginx"}',
|
|
clear=True
|
|
)
|
|
assert action.action == "http"
|
|
assert action.method == "POST"
|
|
assert action.clear is True
|