from __future__ import annotations import tempfile import unittest from unittest.mock import patch from pathlib import Path from types import SimpleNamespace from datetime import datetime, timedelta, timezone 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.mail import CAPABILITY_MAIL_NOTIFICATION_DELIVERY 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 ( NotificationError, create_notification, deliver_pending, list_notifications, notification_preferences_response, notification_response, notification_summary, update_notification_preferences, update_notification, ) class _NotificationMailProvider: def __init__(self) -> None: self.request = None def submit_notification_mail(self, session, request): self.request = request return { "status": "accepted", "provider": "mail.delivery_outbox", "external_message_id": "mail-command-1", } class _CapabilityRegistry: def __init__(self, provider: object) -> None: self.provider = provider def has_capability(self, name: str) -> bool: return name == CAPABILITY_MAIL_NOTIFICATION_DELIVERY def require_capability(self, name: str) -> object: if not self.has_capability(name): raise LookupError(name) return self.provider 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_status_multiselection_filters_before_limit_and_preserves_visibility(self) -> None: with self.Session() as session: for index, (state, recipient, tenant) in enumerate([ ("pending", "user-1", "tenant-1"), ("failed", "user-1", "tenant-1"), ("sent", "user-1", "tenant-1"), ("sent", "user-1", "tenant-1"), ("failed", "user-2", "tenant-1"), ("failed", "user-1", "tenant-2"), ]): item = create_notification(session, tenant_id=tenant, payload=self._inbox_payload(recipient_id=recipient, subject=f"Message {index}")) item.status = state item.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + timedelta(minutes=index) session.flush() principal = self._principal(tenant_id="tenant-1", user_id="user-1", account_id="account-1", scopes={"notifications:notification:read"}) result = api_list_notifications(status_filter=["pending", "failed"], channel=None, source_module=None, recipient_id=None, view="personal", limit=2, session=session, principal=principal) self.assertEqual([item.status for item in result.notifications], ["failed", "pending"]) self.assertEqual(list_notifications(session, tenant_id="tenant-1", status=[]), []) single = list_notifications(session, tenant_id="tenant-1", status="pending") self.assertEqual(len(single), 1) def test_http_repeated_status_parameters_use_or_filter(self) -> None: from fastapi import FastAPI from fastapi.testclient import TestClient from govoplan_core.auth import get_api_principal from govoplan_core.db.session import get_session from govoplan_notifications.backend.router import router app = FastAPI() app.include_router(router) principal = self._principal(tenant_id="tenant-1", user_id="user-1", account_id="account-1", scopes={"notifications:notification:read"}) app.dependency_overrides[get_api_principal] = lambda: principal app.dependency_overrides[get_session] = lambda: object() with patch("govoplan_notifications.backend.router.get_notification_preferences", return_value=SimpleNamespace(muted_source_modules=[])), \ patch("govoplan_notifications.backend.router.list_notifications", return_value=[]) as listing, TestClient(app) as client: response = client.get("/notifications?status=pending&status=failed") self.assertEqual(response.status_code, 200) self.assertEqual(listing.call_args.kwargs["status"], ["pending", "failed"]) self.assertIn("user-1", listing.call_args.kwargs["recipient_ids"]) self.assertEqual(client.get("/notifications?status=sent").status_code, 200) self.assertEqual(listing.call_args.kwargs["status"], ["sent"]) self.assertEqual(client.get("/notifications").status_code, 200) self.assertIsNone(listing.call_args.kwargs["status"]) 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_cancellation_is_limited_to_locally_controllable_delivery_states(self) -> None: with self.Session() as session: notification = create_notification( session, tenant_id="tenant-1", payload=self._inbox_payload(recipient_id="user-1", subject="Cancellation boundary"), ) notification.status = "sent" session.flush() with self.assertRaisesRegex( NotificationError, "cannot be cancelled from status sent", ): update_notification( session, tenant_id="tenant-1", notification_id=notification.id, payload=NotificationUpdateRequest(status="cancelled"), ) notification.status = "queued" cancelled = update_notification( session, tenant_id="tenant-1", notification_id=notification.id, payload=NotificationUpdateRequest(status="cancelled"), ) self.assertEqual("cancelled", cancelled.status) self.assertIsNotNone(cancelled.cancelled_at) 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", (), {"app_env": "dev", "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_production_mail_delivery_pauses_without_mail_capability(self) -> None: with tempfile.TemporaryDirectory() as tmpdir, 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="mail", recipient="person@example.test", subject="Created", ), ) result = deliver_pending( session, tenant_id="tenant-1", settings=SimpleNamespace( app_env="production", mock_mailbox_dir=tmpdir, ), registry=object(), ) self.assertEqual(result["paused"], 1) self.assertEqual(notification.status, "paused") self.assertEqual( notification.last_error, "Mail-backed notification delivery is unavailable.", ) self.assertEqual(list(Path(tmpdir).rglob("*.eml")), []) def test_mail_delivery_uses_optional_mail_capability(self) -> None: provider = _NotificationMailProvider() registry = _CapabilityRegistry(provider) 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="mail", recipient="person@example.test", subject="Created", body_text="Hello", metadata={ "mail_delivery": { "mail_profile_id": "profile-1", "from_address": "notifications@example.test", } }, ), ) result = deliver_pending( session, tenant_id="tenant-1", settings=SimpleNamespace(app_env="production"), registry=registry, ) self.assertEqual(result["accepted"], 1) self.assertEqual(notification.status, "accepted") self.assertEqual(notification.external_message_id, "mail-command-1") self.assertEqual(provider.request.mail_profile_id, "profile-1") self.assertEqual( provider.request.from_address, "notifications@example.test", ) 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") self.assertEqual( "tenant-1", capability.tenant_id_for_notification( session, notification_id=str(payload["id"]), ), ) self.assertIsNone( capability.tenant_id_for_notification( session, notification_id="missing", ) ) 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"]) def test_personal_source_mutes_apply_across_recipient_identifiers(self) -> None: with self.Session() as session: postbox = create_notification( session, tenant_id="tenant-1", payload=NotificationCreateRequest( source_module="postbox", source_resource_type="postbox", event_kind="postbox.assignment.newly_visible.v1", channel="inbox", recipient_type="account", recipient_id="account-1", subject="Postbox responsibility changed", enqueue_delivery=False, ), ) calendar = create_notification( session, tenant_id="tenant-1", payload=NotificationCreateRequest( source_module="calendar", source_resource_type="calendar_event", event_kind="calendar.reminder.v1", channel="inbox", recipient_type="account", recipient_id="account-1", subject="Calendar reminder", enqueue_delivery=False, ), ) update_notification_preferences( session, tenant_id="tenant-1", user_id="user-1", payload=NotificationPreferencesUpdateRequest( muted_source_modules=["postbox"], ), ) principal = self._principal( tenant_id="tenant-1", user_id="user-1", account_id="account-1", scopes={"notifications:notification:read"}, ) personal = api_list_notifications( status_filter=None, channel=None, source_module=None, recipient_id=None, view="personal", limit=100, session=session, principal=principal, ) personal_summary = api_notification_summary( view="personal", session=session, principal=principal, ) tenant_admin = self._principal( tenant_id="tenant-1", user_id="user-1", account_id="account-1", scopes={ "notifications:notification:read", "notifications:notification:admin", }, ) tenant_view = api_list_notifications( status_filter=None, channel=None, source_module=None, recipient_id=None, view="tenant", limit=100, session=session, principal=tenant_admin, ) self.assertEqual([calendar.id], [item.id for item in personal.notifications]) self.assertEqual(1, personal_summary.total) self.assertEqual( {postbox.id, calendar.id}, {item.id for item in tenant_view.notifications}, ) if __name__ == "__main__": unittest.main()