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_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_action_url_accepts_only_application_relative_paths(self) -> None: payload = self._inbox_payload(recipient_id="user-1", subject="Safe action") safe = payload.model_copy(update={"action_url": "/calendar?event=event-1#details"}) self.assertEqual( NotificationCreateRequest.model_validate(safe.model_dump()).action_url, "/calendar?event=event-1#details", ) for action_url in ( "javascript:alert(1)", "data:text/html,unsafe", "file:///etc/passwd", "custom:unsafe", "https://attacker.example.test/collect", "//attacker.example.test/collect", "/\\attacker.example.test/collect", "/calendar\nmalformed", ): with self.subTest(action_url=action_url): with self.assertRaisesRegex(ValueError, "application-relative path"): NotificationCreateRequest.model_validate( {**payload.model_dump(), "action_url": action_url} ) bypassed_validation = NotificationCreateRequest.model_construct( **{**payload.model_dump(), "action_url": "javascript:alert(1)"} ) with self.Session() as session: with self.assertRaisesRegex(ValueError, "application-relative path"): create_notification( session, tenant_id="tenant-1", payload=bypassed_validation, ) def test_legacy_unsafe_action_url_is_not_projected_as_a_link(self) -> None: with self.Session() as session: notification = create_notification( session, tenant_id="tenant-1", payload=self._inbox_payload(recipient_id="user-1", subject="Legacy action"), ) notification.action_url = "javascript:alert(1)" session.flush() self.assertIsNone(notification_response(notification)["action_url"]) 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()