intermittent commit

2026-07-14 13:22:11 +02:00
parent 3444a4920f
commit 6163c8f733
24 changed files with 2367 additions and 0 deletions
--- a/tests/test_notifications.py
+++ b/tests/test_notifications.py
@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+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.schemas import NotificationCreateRequest, NotificationPreferencesUpdateRequest
+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 test_create_and_deliver_inbox_notification(self) -> None:
+        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)
+
+    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()