691 lines
29 KiB
Python
691 lines
29 KiB
Python
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()
|