352 lines
15 KiB
Python
352 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from types import SimpleNamespace
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.modules import ModuleContext
|
|
from govoplan_core.core.notifications import NotificationDispatchRequest
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_notifications.backend.capabilities import dispatch_capability
|
|
from govoplan_notifications.backend.db.models import NotificationDeliveryAttempt, NotificationMessage, NotificationPreference
|
|
from govoplan_notifications.backend.router import api_get_notification, api_list_notifications, api_notification_summary, api_update_notification
|
|
from govoplan_notifications.backend.schemas import NotificationCreateRequest, NotificationPreferencesUpdateRequest, NotificationUpdateRequest
|
|
from govoplan_notifications.backend.service import create_notification, deliver_pending, notification_preferences_response, notification_summary, update_notification_preferences
|
|
|
|
|
|
class NotificationServiceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(self.engine, tables=[NotificationMessage.__table__, NotificationDeliveryAttempt.__table__, NotificationPreference.__table__])
|
|
self.Session = sessionmaker(bind=self.engine)
|
|
|
|
def tearDown(self) -> None:
|
|
Base.metadata.drop_all(
|
|
self.engine,
|
|
tables=[NotificationDeliveryAttempt.__table__, NotificationMessage.__table__, NotificationPreference.__table__],
|
|
)
|
|
self.engine.dispose()
|
|
|
|
@staticmethod
|
|
def _principal(*, tenant_id: str, user_id: str, account_id: str, scopes: set[str]) -> ApiPrincipal:
|
|
user = SimpleNamespace(id=user_id)
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id=account_id,
|
|
membership_id=user_id,
|
|
tenant_id=tenant_id,
|
|
scopes=frozenset(scopes),
|
|
),
|
|
account=SimpleNamespace(id=account_id),
|
|
user=user,
|
|
)
|
|
|
|
@staticmethod
|
|
def _inbox_payload(*, recipient_id: str, subject: str) -> NotificationCreateRequest:
|
|
return NotificationCreateRequest(
|
|
source_module="test",
|
|
source_resource_type="thing",
|
|
event_kind="created",
|
|
channel="inbox",
|
|
recipient_type="user",
|
|
recipient_id=recipient_id,
|
|
subject=subject,
|
|
enqueue_delivery=False,
|
|
)
|
|
|
|
def test_personal_inbox_cannot_read_or_update_another_recipient(self) -> None:
|
|
with self.Session() as session:
|
|
own_membership = create_notification(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
payload=self._inbox_payload(recipient_id="user-1", subject="Membership addressed"),
|
|
)
|
|
own_account = create_notification(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
payload=self._inbox_payload(recipient_id="account-1", subject="Account addressed"),
|
|
)
|
|
another_user = create_notification(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
payload=self._inbox_payload(recipient_id="user-2", subject="Private for another user"),
|
|
)
|
|
other_tenant = create_notification(
|
|
session,
|
|
tenant_id="tenant-2",
|
|
payload=self._inbox_payload(recipient_id="user-1", subject="Other tenant"),
|
|
)
|
|
principal = self._principal(
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
account_id="account-1",
|
|
scopes={"notifications:notification:read", "notifications:notification:write"},
|
|
)
|
|
|
|
result = api_list_notifications(
|
|
status_filter=None,
|
|
channel=None,
|
|
source_module=None,
|
|
recipient_id=None,
|
|
view="personal",
|
|
limit=100,
|
|
session=session,
|
|
principal=principal,
|
|
)
|
|
self.assertEqual({own_membership.id, own_account.id}, {item.id for item in result.notifications})
|
|
|
|
summary = api_notification_summary(view="personal", session=session, principal=principal)
|
|
self.assertEqual(summary.total, 2)
|
|
|
|
with self.assertRaises(HTTPException) as hidden_read:
|
|
api_get_notification(another_user.id, view="personal", session=session, principal=principal)
|
|
self.assertEqual(hidden_read.exception.status_code, 404)
|
|
|
|
with self.assertRaises(HTTPException) as recipient_impersonation:
|
|
api_list_notifications(
|
|
status_filter=None,
|
|
channel=None,
|
|
source_module=None,
|
|
recipient_id="user-2",
|
|
view="personal",
|
|
limit=100,
|
|
session=session,
|
|
principal=principal,
|
|
)
|
|
self.assertEqual(recipient_impersonation.exception.status_code, 403)
|
|
|
|
with self.assertRaises(HTTPException) as hidden_update:
|
|
api_update_notification(
|
|
another_user.id,
|
|
NotificationUpdateRequest(status="read"),
|
|
view="personal",
|
|
session=session,
|
|
principal=principal,
|
|
)
|
|
self.assertEqual(hidden_update.exception.status_code, 404)
|
|
self.assertIsNone(another_user.read_at)
|
|
|
|
with self.assertRaises(HTTPException) as cross_tenant:
|
|
api_get_notification(other_tenant.id, view="personal", session=session, principal=principal)
|
|
self.assertEqual(cross_tenant.exception.status_code, 404)
|
|
|
|
def test_tenant_notification_view_is_an_explicit_admin_operation(self) -> None:
|
|
with self.Session() as session:
|
|
another_user = create_notification(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
payload=self._inbox_payload(recipient_id="user-2", subject="Administrative outbox entry"),
|
|
)
|
|
reader = self._principal(
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
account_id="account-1",
|
|
scopes={"notifications:notification:read"},
|
|
)
|
|
admin = self._principal(
|
|
tenant_id="tenant-1",
|
|
user_id="user-admin",
|
|
account_id="account-admin",
|
|
scopes={"notifications:notification:read", "notifications:notification:admin"},
|
|
)
|
|
|
|
with self.assertRaises(HTTPException) as forbidden:
|
|
api_get_notification(another_user.id, view="tenant", session=session, principal=reader)
|
|
self.assertEqual(forbidden.exception.status_code, 403)
|
|
|
|
visible = api_get_notification(another_user.id, view="tenant", session=session, principal=admin)
|
|
self.assertEqual(visible.id, another_user.id)
|
|
|
|
def test_create_and_deliver_inbox_notification(self) -> None:
|
|
with patch("govoplan_notifications.backend.service._enqueue_celery_delivery") as enqueue:
|
|
with self.Session() as session:
|
|
notification = create_notification(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
payload=NotificationCreateRequest(
|
|
source_module="test",
|
|
source_resource_type="thing",
|
|
source_resource_id="thing-1",
|
|
event_kind="created",
|
|
channel="inbox",
|
|
recipient_id="user-1",
|
|
subject="Created",
|
|
),
|
|
)
|
|
result = deliver_pending(session, tenant_id="tenant-1")
|
|
self.assertEqual(result["sent"], 1)
|
|
self.assertEqual(notification.status, "sent")
|
|
self.assertEqual(notification.attempt_count, 1)
|
|
session.commit()
|
|
enqueue.assert_not_called()
|
|
|
|
def test_delivery_task_is_enqueued_only_after_notification_commit(self) -> None:
|
|
with patch("govoplan_notifications.backend.service._enqueue_celery_delivery") as enqueue:
|
|
with self.Session() as session:
|
|
notification = create_notification(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
payload=NotificationCreateRequest(
|
|
source_module="test",
|
|
source_resource_type="thing",
|
|
event_kind="created",
|
|
channel="inbox",
|
|
recipient_id="user-1",
|
|
subject="Committed",
|
|
),
|
|
)
|
|
notification_id = notification.id
|
|
enqueue.assert_not_called()
|
|
|
|
session.commit()
|
|
|
|
enqueue.assert_called_once_with(notification_id)
|
|
with self.Session() as verification_session:
|
|
self.assertIsNotNone(verification_session.get(NotificationMessage, notification_id))
|
|
|
|
def test_rolled_back_notification_never_enqueues_a_delivery_task(self) -> None:
|
|
with patch("govoplan_notifications.backend.service._enqueue_celery_delivery") as enqueue:
|
|
with self.Session() as session:
|
|
notification = create_notification(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
payload=NotificationCreateRequest(
|
|
source_module="test",
|
|
source_resource_type="thing",
|
|
event_kind="created",
|
|
channel="inbox",
|
|
recipient_id="user-1",
|
|
subject="Rolled back",
|
|
),
|
|
)
|
|
notification_id = notification.id
|
|
session.rollback()
|
|
session.commit()
|
|
|
|
enqueue.assert_not_called()
|
|
with self.Session() as verification_session:
|
|
self.assertIsNone(verification_session.get(NotificationMessage, notification_id))
|
|
|
|
def test_mail_delivery_uses_local_file_transport(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir, self.Session() as session:
|
|
settings = type("Settings", (), {"mock_mailbox_dir": tmpdir})()
|
|
notification = create_notification(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
payload=NotificationCreateRequest(
|
|
source_module="test",
|
|
source_resource_type="thing",
|
|
source_resource_id="thing-1",
|
|
event_kind="created",
|
|
channel="mail",
|
|
recipient="person@example.test",
|
|
subject="Created",
|
|
body_text="Hello",
|
|
),
|
|
)
|
|
result = deliver_pending(session, tenant_id="tenant-1", settings=settings)
|
|
self.assertEqual(result["sent"], 1)
|
|
self.assertEqual(notification.status, "sent")
|
|
self.assertTrue(notification.external_message_id.endswith(".eml"))
|
|
|
|
def test_dispatch_capability_enqueues_notification(self) -> None:
|
|
with self.Session() as session:
|
|
capability = dispatch_capability(ModuleContext(registry=object(), settings=object()))
|
|
payload = capability.enqueue_notification(
|
|
session,
|
|
NotificationDispatchRequest(
|
|
tenant_id="tenant-1",
|
|
source_module="scheduling",
|
|
source_resource_type="request",
|
|
source_resource_id="req-1",
|
|
event_kind="invitation",
|
|
channel="inbox",
|
|
recipient_id="user-1",
|
|
subject="Invite",
|
|
),
|
|
enqueue_delivery=False,
|
|
)
|
|
self.assertEqual(payload["source_module"], "scheduling")
|
|
self.assertEqual(payload["status"], "pending")
|
|
|
|
def test_summary_counts_unread_active_notifications(self) -> None:
|
|
with self.Session() as session:
|
|
unread = create_notification(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
payload=NotificationCreateRequest(
|
|
source_module="test",
|
|
source_resource_type="thing",
|
|
event_kind="created",
|
|
channel="inbox",
|
|
subject="Unread",
|
|
enqueue_delivery=False,
|
|
),
|
|
)
|
|
read = create_notification(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
payload=NotificationCreateRequest(
|
|
source_module="test",
|
|
source_resource_type="thing",
|
|
event_kind="read",
|
|
channel="inbox",
|
|
subject="Read",
|
|
enqueue_delivery=False,
|
|
),
|
|
)
|
|
cancelled = create_notification(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
payload=NotificationCreateRequest(
|
|
source_module="test",
|
|
source_resource_type="thing",
|
|
event_kind="cancelled",
|
|
channel="inbox",
|
|
subject="Cancelled",
|
|
enqueue_delivery=False,
|
|
),
|
|
)
|
|
read.read_at = read.created_at
|
|
cancelled.status = "cancelled"
|
|
session.flush()
|
|
|
|
summary = notification_summary(session, tenant_id="tenant-1")
|
|
|
|
self.assertEqual(summary["total"], 3)
|
|
self.assertEqual(summary["unread"], 1)
|
|
self.assertEqual(summary["pending"], 2)
|
|
self.assertTrue(summary["show_unread_badge"])
|
|
self.assertEqual(unread.status, "pending")
|
|
|
|
def test_preferences_update_controls_summary_badge_flag(self) -> None:
|
|
with self.Session() as session:
|
|
preference = update_notification_preferences(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
payload=NotificationPreferencesUpdateRequest(
|
|
show_unread_badge=False,
|
|
email_enabled=True,
|
|
muted_source_modules=["Calendar", "calendar", " campaign "],
|
|
),
|
|
)
|
|
response = notification_preferences_response(preference)
|
|
summary = notification_summary(session, tenant_id="tenant-1", user_id="user-1")
|
|
|
|
self.assertFalse(response["show_unread_badge"])
|
|
self.assertTrue(response["email_enabled"])
|
|
self.assertEqual(["calendar", "campaign"], response["muted_source_modules"])
|
|
self.assertFalse(summary["show_unread_badge"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|