Amirhossein Khalili
ff99130938
Split backend tests into unit and integration suites, add isolated fixtures and coverage enforcement, and update CI and test documentation. Cover current authentication, content, notification, payment, media, and admin flows; fix behaviors revealed by the new tests. Validate both suites and the per-module gate against local PostgreSQL.
77 lines
3.9 KiB
Python
77 lines
3.9 KiB
Python
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from apps.communications.push_notifications import PushNotificationService
|
|
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
def test_push_delivery_passes_payload_to_gateway(mocker):
|
|
gateway = mocker.patch("apps.communications.push_notifications.webpush")
|
|
service = PushNotificationService()
|
|
assert service.send_notification({"endpoint": "https://push.test"}, {"title": "News"})
|
|
assert gateway.call_args.kwargs["ttl"] == 86400
|
|
assert '"title": "News"' in gateway.call_args.kwargs["data"]
|
|
|
|
|
|
def test_push_batch_counts_success_and_failures(mocker):
|
|
service = PushNotificationService()
|
|
mocker.patch.object(service, "send_notification", side_effect=[True, False])
|
|
devices = [SimpleNamespace(endpoint="a", p256dh_key="key", auth_key="auth"), SimpleNamespace(endpoint="b", p256dh_key="key", auth_key="auth")]
|
|
assert service.send_to_multiple(devices, {"title": "News"}) == {"sent": 1, "failed": 1}
|
|
|
|
|
|
def test_push_delivery_handles_gateway_failure(mocker):
|
|
gateway = mocker.patch("apps.communications.push_notifications.webpush", side_effect=RuntimeError("offline"))
|
|
service = PushNotificationService()
|
|
assert not service.send_notification({"endpoint": "https://push.test"}, {"title": "News"})
|
|
gateway.assert_called_once()
|
|
|
|
|
|
def test_push_announcement_and_reminder_build_expected_payloads(mocker):
|
|
service = PushNotificationService()
|
|
sender = mocker.patch.object(service, "send_to_multiple", return_value={"sent": 1, "failed": 0})
|
|
device = SimpleNamespace(endpoint="push", p256dh_key="key", auth_key="auth")
|
|
announcement = SimpleNamespace(id=7, title="Notice", content="x" * 110)
|
|
assert service.send_announcement_notification(announcement, [device]) == {"sent": 1, "failed": 0}
|
|
assert sender.call_args.args[0] == [device]
|
|
assert sender.call_args.args[1]["body"] == "x" * 100 + "..."
|
|
event = SimpleNamespace(id=9, title="Workshop")
|
|
service.send_event_reminder_notification(event, [device])
|
|
assert sender.call_args.args[1]["data"]["url"] == "/events/9/"
|
|
|
|
|
|
@pytest.mark.parametrize("audience,expected", [("all", {"is_active": True}),
|
|
("members", {"user__is_member": True, "is_active": True}),
|
|
("committee", {"user__is_committee_member": True, "is_active": True})])
|
|
def test_push_announcement_selects_audience(mocker, audience, expected):
|
|
service = PushNotificationService()
|
|
devices = mocker.patch("apps.communications.push_notifications.PushNotificationDevice.objects.filter")
|
|
sender = mocker.patch.object(service, "send_to_multiple", return_value={"sent": 0, "failed": 0})
|
|
announcement = SimpleNamespace(id=1, title="Alert", content="Short", audience=audience)
|
|
service.send_announcement_notification(announcement)
|
|
devices.assert_called_once_with(**expected)
|
|
sender.assert_called_once()
|
|
|
|
|
|
def test_push_reminder_selects_registered_devices(mocker):
|
|
service = PushNotificationService()
|
|
registrations = mocker.patch("apps.communications.push_notifications.Registration.objects.filter")
|
|
registrations.return_value.values_list.return_value = [1, 2]
|
|
devices = mocker.patch("apps.communications.push_notifications.PushNotificationDevice.objects.filter")
|
|
mocker.patch.object(service, "send_to_multiple", return_value={"sent": 0, "failed": 0})
|
|
event = SimpleNamespace(id=3, title="Event")
|
|
service.send_event_reminder_notification(event)
|
|
registrations.assert_called_once_with(event=event, status="confirmed")
|
|
devices.assert_called_once_with(user_id__in=[1, 2], is_active=True)
|
|
|
|
|
|
def test_invalid_push_subscription_is_removed(mocker):
|
|
service = PushNotificationService()
|
|
deleted = mocker.patch("apps.communications.push_notifications.PushNotificationDevice.objects.filter")
|
|
service._remove_invalid_subscription({"endpoint": "https://push.test"})
|
|
deleted.assert_called_once_with(endpoint="https://push.test")
|
|
deleted.return_value.delete.assert_called_once()
|