GovOPlaN/govoplan-notifications
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(notifications): enforce personal source mutes

Browse Source
This commit is contained in:
zemion
2026-08-20 04:58:34 +02:00
parent f9f7ab75d3
commit 5c28ac5094
4 changed files with 122 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_notifications/backend/manifest.py
+1 -1
View File
@@ -100,7 +100,7 @@ manifest = ModuleManifest(
id="notifications.center-and-preferences",
title="Use the notification center",
summary="The title-bar badge and notification center collect durable notices that require attention outside an immediate request.",
body="Open the notification center to read, acknowledge, or follow notifications from enabled modules. Preferences control eligible delivery channels and categories. Disabling an optional external channel does not remove the in-product notification unless the originating module's retention policy does so.",
body="Open the notification center to read, acknowledge, or follow notifications from enabled modules. Personal source-muting preferences remove matching entries from the personal list and badge counts even when a producer addressed the actor through an account, membership, or identity identifier; tenant-administrator evidence views remain complete. Preferences also control eligible delivery channels and categories. Disabling an optional external channel does not remove an unmuted in-product notification unless the originating module's retention policy does so.",
documentation_types=("user",),
audience=("user",),
related_modules=("mail", "calendar", "scheduling", "workflow_engine"),
src/govoplan_notifications/backend/router.py
+10 -1
View File
@@ -95,6 +95,14 @@ def api_list_notifications(
recipient_ids = _recipient_ids_for_view(principal, view)
if recipient_id is not None and recipient_ids is not None and recipient_id not in recipient_ids:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot read another recipient's notifications")
muted_source_modules: tuple[str, ...] = ()
if view == "personal":
preference = get_notification_preferences(
session,
tenant_id=principal.tenant_id,
user_id=principal.user.id,
)
muted_source_modules = tuple(preference.muted_source_modules or ())
notifications = list_notifications(
session,
tenant_id=principal.tenant_id,
@@ -103,6 +111,7 @@ def api_list_notifications(
source_module=source_module,
recipient_id=recipient_id,
recipient_ids=recipient_ids,
muted_source_modules=muted_source_modules,
limit=limit,
)
return NotificationListResponse(notifications=[_response(notification) for notification in notifications])
@@ -119,7 +128,7 @@ def api_notification_summary(
notification_summary(
session,
tenant_id=principal.tenant_id,
user_id=principal.user.id,
user_id=principal.user.id if view == "personal" else None,
recipient_ids=_recipient_ids_for_view(principal, view),
)
)
src/govoplan_notifications/backend/service.py
+24 -4
View File
@@ -1,6 +1,6 @@
from __future__ import annotations
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from datetime import datetime, timezone
from email.message import EmailMessage
from pathlib import Path
@@ -180,6 +180,7 @@ def list_notifications(
source_module: str | None = None,
recipient_id: str | None = None,
recipient_ids: tuple[str, ...] | None = None,
muted_source_modules: Sequence[str] = (),
limit: int = 100,
) -> list[NotificationMessage]:
query = session.query(NotificationMessage).filter(
@@ -196,6 +197,9 @@ def list_notifications(
query = query.filter(NotificationMessage.recipient_id.in_(recipient_ids))
if recipient_id:
query = query.filter(NotificationMessage.recipient_id == recipient_id)
muted_sources = _clean_source_modules(list(muted_source_modules))
if muted_sources:
query = query.filter(NotificationMessage.source_module.notin_(muted_sources))
return query.order_by(NotificationMessage.created_at.desc(), NotificationMessage.id.asc()).limit(limit).all()
@@ -206,12 +210,28 @@ def notification_summary(
user_id: str | None = None,
recipient_ids: tuple[str, ...] | None = None,
) -> dict[str, int | bool]:
preference = (
get_notification_preferences(
session,
tenant_id=tenant_id,
user_id=user_id,
)
if user_id
else None
)
filters = [
NotificationMessage.tenant_id == tenant_id,
NotificationMessage.deleted_at.is_(None),
]
if recipient_ids is not None:
filters.append(NotificationMessage.recipient_id.in_(recipient_ids))
muted_sources = _clean_source_modules(
list(preference.muted_source_modules or [])
if preference is not None
else []
)
if muted_sources:
filters.append(NotificationMessage.source_module.notin_(muted_sources))
active = NotificationMessage.status.notin_(["cancelled", "skipped"])
total, unread, pending, failed = (
session.query(
@@ -267,9 +287,9 @@ def notification_summary(
.filter(*filters)
.one()
)
show_unread_badge = True
if user_id:
show_unread_badge = get_notification_preferences(session, tenant_id=tenant_id, user_id=user_id).show_unread_badge
show_unread_badge = (
preference.show_unread_badge if preference is not None else True
)
return {
"total": int(total or 0),
"unread": int(unread or 0),
tests/test_notifications.py
+87
View File
@@ -551,6 +551,93 @@ class NotificationServiceTests(unittest.TestCase):
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()