Bruno Charest
ecefbc8611
Some checks failed
Tests / Backend Tests (Python) (3.10) (push) Has been cancelled
Tests / Backend Tests (Python) (3.11) (push) Has been cancelled
Tests / Backend Tests (Python) (3.12) (push) Has been cancelled
Tests / Frontend Tests (JS) (push) Has been cancelled
Tests / Integration Tests (push) Has been cancelled
Tests / All Tests Passed (push) Has been cancelled
385 lines
13 KiB
Python
385 lines
13 KiB
Python
"""
|
|
Tests pour le service de notification ntfy.
|
|
|
|
Couvre:
|
|
- Configuration NtfyConfig
|
|
- Construction des headers
|
|
- Envoi de notifications (mocked HTTP)
|
|
- Templates de notification
|
|
- Gestion des erreurs réseau
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
import httpx
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
class TestNtfyConfig:
|
|
"""Tests pour la configuration NtfyConfig."""
|
|
|
|
def test_default_config(self):
|
|
"""Configuration par défaut."""
|
|
from app.schemas.notification import NtfyConfig
|
|
|
|
config = NtfyConfig()
|
|
|
|
assert config.base_url == "http://localhost:8150"
|
|
assert config.default_topic == "homelab-events"
|
|
assert config.enabled is True
|
|
assert config.timeout == 5
|
|
|
|
def test_custom_config(self):
|
|
"""Configuration personnalisée."""
|
|
from app.schemas.notification import NtfyConfig
|
|
|
|
config = NtfyConfig(
|
|
base_url="http://ntfy.example.com:8080",
|
|
default_topic="my-topic",
|
|
enabled=False,
|
|
timeout=10
|
|
)
|
|
|
|
assert config.base_url == "http://ntfy.example.com:8080"
|
|
assert config.default_topic == "my-topic"
|
|
assert config.enabled is False
|
|
assert config.timeout == 10
|
|
|
|
def test_has_auth_with_username_password(self):
|
|
"""has_auth avec username/password."""
|
|
from app.schemas.notification import NtfyConfig
|
|
|
|
config = NtfyConfig(username="user", password="pass")
|
|
assert config.has_auth is True
|
|
|
|
def test_has_auth_with_token(self):
|
|
"""has_auth avec token."""
|
|
from app.schemas.notification import NtfyConfig
|
|
|
|
config = NtfyConfig(token="my-token")
|
|
assert config.has_auth is True
|
|
|
|
def test_has_auth_without_credentials(self):
|
|
"""has_auth sans credentials."""
|
|
from app.schemas.notification import NtfyConfig
|
|
|
|
config = NtfyConfig()
|
|
assert config.has_auth is False
|
|
|
|
def test_has_auth_partial_credentials(self):
|
|
"""has_auth avec credentials partiels (username seul)."""
|
|
from app.schemas.notification import NtfyConfig
|
|
|
|
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):
|
|
"""Chargement depuis variables d'environnement."""
|
|
from app.schemas.notification import NtfyConfig
|
|
|
|
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 TestNotificationRequest:
|
|
"""Tests pour NotificationRequest."""
|
|
|
|
def test_minimal_request(self):
|
|
"""Requête minimale."""
|
|
from app.schemas.notification import NotificationRequest
|
|
|
|
req = NotificationRequest(message="Hello")
|
|
|
|
assert req.message == "Hello"
|
|
assert req.topic is None
|
|
assert req.priority is None
|
|
|
|
def test_full_request(self):
|
|
"""Requête complète."""
|
|
from app.schemas.notification import NotificationRequest
|
|
|
|
req = NotificationRequest(
|
|
topic="my-topic",
|
|
message="Hello World",
|
|
title="Greeting",
|
|
priority=5,
|
|
tags=["wave", "robot"],
|
|
click="http://example.com"
|
|
)
|
|
|
|
assert req.topic == "my-topic"
|
|
assert req.message == "Hello World"
|
|
assert req.title == "Greeting"
|
|
assert req.priority == 5
|
|
assert req.tags == ["wave", "robot"]
|
|
|
|
def test_priority_validation_valid(self):
|
|
"""Validation priorité valide (1-5)."""
|
|
from app.schemas.notification import NotificationRequest
|
|
|
|
for p in [1, 2, 3, 4, 5]:
|
|
req = NotificationRequest(message="Test", priority=p)
|
|
assert req.priority == p
|
|
|
|
def test_priority_validation_invalid(self):
|
|
"""Validation priorité invalide."""
|
|
from app.schemas.notification import NotificationRequest
|
|
|
|
with pytest.raises(ValueError):
|
|
NotificationRequest(message="Test", priority=0)
|
|
|
|
with pytest.raises(ValueError):
|
|
NotificationRequest(message="Test", priority=6)
|
|
|
|
|
|
class TestNotificationService:
|
|
"""Tests pour NotificationService."""
|
|
|
|
@pytest.fixture
|
|
def service(self):
|
|
"""Service avec config de test."""
|
|
from app.services.notification_service import NotificationService
|
|
from app.schemas.notification import NtfyConfig
|
|
|
|
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):
|
|
"""Service désactivé."""
|
|
from app.services.notification_service import NotificationService
|
|
from app.schemas.notification import NtfyConfig
|
|
|
|
config = NtfyConfig(enabled=False)
|
|
return NotificationService(config)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_when_disabled(self, disabled_service):
|
|
"""send retourne True quand désactivé."""
|
|
result = await disabled_service.send(message="Test", topic="test")
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_success(self, service):
|
|
"""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()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_failure_http_error(self, service):
|
|
"""Envoi échoué (erreur HTTP 500)."""
|
|
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):
|
|
"""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):
|
|
"""Gestion 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):
|
|
"""Utilise le topic par défaut 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):
|
|
"""Headers avec auth Basic."""
|
|
from app.services.notification_service import NotificationService
|
|
from app.schemas.notification import NtfyConfig
|
|
|
|
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):
|
|
"""Headers avec auth Bearer."""
|
|
from app.services.notification_service import NotificationService
|
|
from app.schemas.notification import NtfyConfig
|
|
|
|
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):
|
|
"""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):
|
|
"""Reconfiguration du service."""
|
|
from app.schemas.notification import NtfyConfig
|
|
|
|
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
|
|
|
|
|
|
class TestNotificationTemplates:
|
|
"""Tests pour les templates de notification."""
|
|
|
|
def test_backup_success(self):
|
|
"""Template backup réussi."""
|
|
from app.schemas.notification import NotificationTemplates
|
|
|
|
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
|
|
|
|
def test_backup_failed(self):
|
|
"""Template backup échoué."""
|
|
from app.schemas.notification import NotificationTemplates
|
|
|
|
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
|
|
|
|
def test_bootstrap_started(self):
|
|
"""Template bootstrap démarré."""
|
|
from app.schemas.notification import NotificationTemplates
|
|
|
|
req = NotificationTemplates.bootstrap_started("new-host.home")
|
|
|
|
assert req.topic == "homelab-bootstrap"
|
|
assert "new-host.home" in req.message
|
|
|
|
def test_health_status_down(self):
|
|
"""Template changement d'état (down)."""
|
|
from app.schemas.notification import NotificationTemplates
|
|
|
|
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
|
|
|
|
def test_health_status_up(self):
|
|
"""Template changement d'état (up)."""
|
|
from app.schemas.notification import NotificationTemplates
|
|
|
|
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
|