159 lines
6.7 KiB
Python
159 lines
6.7 KiB
Python
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()
|