|
|
|
@ -1,11 +1,11 @@
|
|
|
|
|
import sys
|
|
|
|
|
import os
|
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')))
|
|
|
|
|
from pathlib import Path
|
|
|
|
|
sys.path.append(str(Path(__file__).absolute().parent.parent))
|
|
|
|
|
|
|
|
|
|
import pytest
|
|
|
|
|
from unittest.mock import patch, MagicMock, AsyncMock
|
|
|
|
|
from bson import ObjectId
|
|
|
|
|
from api.app.push_service import PushService
|
|
|
|
|
from app.push_service import PushService
|
|
|
|
|
|
|
|
|
|
# Fixture pour obtenir une instance du service à tester
|
|
|
|
|
@pytest.fixture
|
|
|
|
@ -13,8 +13,8 @@ def push_service():
|
|
|
|
|
return PushService()
|
|
|
|
|
|
|
|
|
|
# Teste l'envoi de notification quand l'utilisateur existe et a des abonnements
|
|
|
|
|
@patch('api.app.push_service.users_collection')
|
|
|
|
|
@patch('api.app.push_service.PushService.send_notification_to_subscription', new_callable=AsyncMock)
|
|
|
|
|
@patch('app.push_service.users_collection')
|
|
|
|
|
@patch('app.push_service.PushService.send_notification_to_subscription', new_callable=AsyncMock)
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_send_notification_success(mock_send, mock_users, push_service):
|
|
|
|
|
user_id = ObjectId()
|
|
|
|
@ -29,7 +29,7 @@ async def test_send_notification_success(mock_send, mock_users, push_service):
|
|
|
|
|
mock_send.assert_awaited()
|
|
|
|
|
|
|
|
|
|
# Teste le cas où l'utilisateur n'existe pas
|
|
|
|
|
@patch('api.app.push_service.users_collection')
|
|
|
|
|
@patch('app.push_service.users_collection')
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_send_notification_no_user(mock_users, push_service):
|
|
|
|
|
user_id = ObjectId()
|
|
|
|
@ -41,7 +41,7 @@ async def test_send_notification_no_user(mock_users, push_service):
|
|
|
|
|
assert result is False
|
|
|
|
|
|
|
|
|
|
# Teste le cas où l'utilisateur n'a pas de souscriptions push
|
|
|
|
|
@patch('api.app.push_service.users_collection')
|
|
|
|
|
@patch('app.push_service.users_collection')
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_send_notification_no_subs(mock_users, push_service):
|
|
|
|
|
user_id = ObjectId()
|
|
|
|
@ -54,8 +54,8 @@ async def test_send_notification_no_subs(mock_users, push_service):
|
|
|
|
|
assert result is False
|
|
|
|
|
|
|
|
|
|
# Teste l'envoi d'une notification à une souscription valide
|
|
|
|
|
@patch('api.app.push_service.webpush')
|
|
|
|
|
@patch('api.app.push_service.get_audience', return_value='https://test')
|
|
|
|
|
@patch('app.push_service.webpush')
|
|
|
|
|
@patch('app.push_service.get_audience', return_value='https://test')
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_send_notification_to_subscription_success(mock_aud, mock_webpush, push_service):
|
|
|
|
|
subscription = '{"endpoint": "https://test"}'
|
|
|
|
@ -68,7 +68,7 @@ async def test_send_notification_to_subscription_success(mock_aud, mock_webpush,
|
|
|
|
|
mock_webpush.assert_called_once()
|
|
|
|
|
|
|
|
|
|
# Teste le cas où webpush lève une exception (échec d'envoi)
|
|
|
|
|
@patch('api.app.push_service.webpush', side_effect=Exception('fail'))
|
|
|
|
|
@patch('app.push_service.webpush', side_effect=Exception('fail'))
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_send_notification_to_subscription_exception(mock_webpush, push_service):
|
|
|
|
|
subscription = '{"endpoint": "https://test"}'
|
|
|
|
@ -80,7 +80,7 @@ async def test_send_notification_to_subscription_exception(mock_webpush, push_se
|
|
|
|
|
assert result is False
|
|
|
|
|
|
|
|
|
|
# Teste l'envoi à plusieurs souscriptions (succès et échec)
|
|
|
|
|
@patch('api.app.push_service.PushService.send_notification', new_callable=AsyncMock)
|
|
|
|
|
@patch('app.push_service.PushService.send_notification', new_callable=AsyncMock)
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_send_notification_to_all(mock_send, push_service):
|
|
|
|
|
mock_send.side_effect = [True, False] # Un succès, un échec
|
|
|
|
@ -103,7 +103,7 @@ def test_create_notification_payload(push_service):
|
|
|
|
|
assert "icon" in payload["notification"]
|
|
|
|
|
|
|
|
|
|
# Teste la suppression d'une souscription (succès)
|
|
|
|
|
@patch('api.app.push_service.users_collection')
|
|
|
|
|
@patch('app.push_service.users_collection')
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_delete_subscription_success(mock_users, push_service):
|
|
|
|
|
mock_users.update_one.return_value = MagicMock() # Simule la suppression
|
|
|
|
@ -115,7 +115,7 @@ async def test_delete_subscription_success(mock_users, push_service):
|
|
|
|
|
assert result is True
|
|
|
|
|
|
|
|
|
|
# Teste la suppression d'une souscription quand une exception est levée (échec)
|
|
|
|
|
@patch('api.app.push_service.users_collection.update_one', side_effect=Exception('fail'))
|
|
|
|
|
@patch('app.push_service.users_collection.update_one', side_effect=Exception('fail'))
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
|
|
async def test_delete_subscription_exception(mock_update_one, push_service):
|
|
|
|
|
user_id = ObjectId()
|
|
|
|
|