Add ntfy push notification system with comprehensive configuration, API endpoints, and automatic event notifications for tasks, schedules, health checks, and bootstrap operations
This commit is contained in:
@@ -0,0 +1,431 @@
|
||||
"""
|
||||
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
|
||||
Reference in New Issue
Block a user