from __future__ import annotations import json import unittest from sqlalchemy import create_engine from sqlalchemy.orm import Session from govoplan_core.core.dsar import ( DsarErasureActionRef, DsarProvider, DsarRecordRef, DsarSubjectRef, ) from govoplan_core.db.base import Base from govoplan_core.privacy.dsar_workflow import ( create_data_subject_request, search_data_subject_request, ) from govoplan_notifications.backend.db.models import ( NotificationDeliveryAttempt, NotificationMessage, NotificationPreference, ) from govoplan_notifications.backend.dsar_provider import ( NOTIFICATIONS_DSAR_CAPABILITY, NotificationsDsarProvider, ) from govoplan_notifications.backend.manifest import manifest class _Registry: def __init__( self, provider: NotificationsDsarProvider, *, active: bool = True, ) -> None: self.provider = provider self.active = active def capability_names(self): return (NOTIFICATIONS_DSAR_CAPABILITY,) def capability_owner(self, name): self._assert_capability(name) return "notifications" def tenant_entitlement_resolver(self): active = self.active class _Resolver: @staticmethod def resolve(session, tenant_id): del session, tenant_id return type( "State", (), {"effective_modules": ("notifications",) if active else ()}, )() return _Resolver() def require_tenant_capability(self, name, session, **kwargs): del session, kwargs self._assert_capability(name) return self.provider def manifests(self): return (type("Manifest", (), {"id": "notifications"})(),) @staticmethod def _assert_capability(name: str) -> None: if name != NOTIFICATIONS_DSAR_CAPABILITY: raise KeyError(name) class NotificationsDsarProviderTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all(self.engine) self.session = Session(self.engine) self.provider = NotificationsDsarProvider() self.assertIsInstance(self.provider, DsarProvider) self._seed() self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def _message( self, notification_id: str, *, tenant_id: str = "tenant-1", channel: str = "inbox", recipient: str | None = None, recipient_id: str | None = "membership-1", subject: str, ) -> NotificationMessage: return NotificationMessage( id=notification_id, tenant_id=tenant_id, source_module="tasks", source_resource_type="task", source_resource_id=f"task-{notification_id}", event_kind="task_due", channel=channel, recipient=recipient, recipient_type="email" if recipient else "membership", recipient_id=recipient_id, recipient_label="Resident Example", subject=subject, body_text=f"Private body for {notification_id}", body_html=f"

Private HTML for {notification_id}

", action_url="/tasks", priority=2, status="sent", attempt_count=0, payload={ "case_reference": f"CASE-{notification_id}", "access_token": "provider-secret-do-not-export", }, metadata_={"classification": "personal"}, ) def _seed(self) -> None: inbox = self._message( "notification-inbox", subject="Personal inbox notice", ) mail = self._message( "notification-mail", channel="mail", recipient="Resident@Example.test", recipient_id=None, subject="Personal mail notice", ) mail.attempt_count = 1 mail.attempts.append( NotificationDeliveryAttempt( id="attempt-mail", tenant_id="tenant-1", attempt_no=1, channel="mail", provider="mail.delivery_outbox", status="accepted", external_message_id="mail-command-1", details={ "provider_state": "accepted", "authorization": "Bearer provider-secret-do-not-export", }, ) ) self.session.add_all( ( inbox, mail, self._message( "notification-other-recipient", recipient_id="membership-other", subject="Other recipient private notice", ), self._message( "notification-other-tenant", tenant_id="tenant-2", subject="Other tenant private notice", ), NotificationPreference( id="preference-personal", tenant_id="tenant-1", user_id="membership-1", show_unread_badge=False, email_enabled=True, email_digest_enabled=True, muted_source_modules=["campaign", "calendar"], metadata_={"digest_hour": 8}, ), NotificationPreference( id="preference-other", tenant_id="tenant-1", user_id="membership-other", show_unread_badge=True, email_enabled=False, email_digest_enabled=False, muted_source_modules=[], metadata_={}, ), ) ) @staticmethod def _subject() -> DsarSubjectRef: return DsarSubjectRef( account_id="account-1", membership_id="membership-1", identity_id="identity-1", email="resident@example.test", ) def test_search_is_tenant_recipient_scoped_and_exports_bounded_content( self, ) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self._subject(), ) self.assertEqual( [ "preference-personal", "notification-inbox", "notification-mail", ], [record.resource_id for record in records], ) exported = json.dumps([record.to_dict() for record in records]) self.assertIn("Private body for notification-inbox", exported) self.assertIn("CASE-notification-mail", exported) self.assertIn("mail-command-1", exported) self.assertIn("digest_hour", exported) self.assertIn("[redacted]", exported) self.assertNotIn("provider-secret-do-not-export", exported) self.assertNotIn("notification-other-recipient", exported) self.assertNotIn("notification-other-tenant", exported) def test_resource_references_narrow_and_alias_conflicts_fail_closed(self) -> None: notification = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( membership_id="membership-1", external_references={ "notifications.notification": "notification-inbox" }, ), ) preference = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( membership_id="membership-1", external_references={"notifications.preference": "preference-personal"}, ), ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"notifications.account": "account-other"}, ), ) reference_only = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={"notifications.notification": "notification-inbox"} ), ) self.assertEqual( ["notification-inbox"], [item.resource_id for item in notification] ) self.assertEqual( ["preference-personal"], [item.resource_id for item in preference] ) self.assertEqual((), conflict) self.assertEqual((), reference_only) def test_erasure_deletes_preferences_and_inbox_but_reviews_mail(self) -> None: subject = self._subject() records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=subject, ) actions = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=subject, records=records, ) self.assertEqual( ["delete", "delete", "manual_review"], [action.kind for action in actions], ) first = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=actions, request_id="dsar-1", ) second = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=actions, request_id="dsar-1", ) self.assertEqual( ["executed", "executed", "blocked"], [result.status for result in first], ) self.assertEqual( ["unchanged", "unchanged", "blocked"], [result.status for result in second], ) self.assertIsNone( self.session.get(NotificationPreference, "preference-personal") ) self.assertIsNone(self.session.get(NotificationMessage, "notification-inbox")) self.assertIsNotNone(self.session.get(NotificationMessage, "notification-mail")) self.assertIsNotNone( self.session.get(NotificationMessage, "notification-other-recipient") ) def test_changed_and_foreign_resources_are_blocked(self) -> None: subject = self._subject() record = next( item for item in self.provider.search_subject( self.session, tenant_id="tenant-1", subject=subject, ) if item.resource_id == "notification-inbox" ) action = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=subject, records=(record,), )[0] notification = self.session.get(NotificationMessage, "notification-inbox") notification.subject = "Changed after planning" self.session.flush() changed = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=(action,), request_id="dsar-1", ) self.assertEqual("blocked", changed[0].status) with self.assertRaisesRegex(ValueError, "foreign provider record"): self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=subject, records=( DsarRecordRef( provider_id="mail", module_id="mail", resource_type="notification_message", resource_id="notification-inbox", category="message", title="Foreign message", ), ), ) with self.assertRaisesRegex(ValueError, "foreign provider action"): self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=( DsarErasureActionRef( action_id="mail:delete:notification:notification-inbox", provider_id="mail", module_id="mail", kind="delete", resource_type="notification_message", resource_id="notification-inbox", title="Delete notification", rationale="Foreign action", executable=True, ), ), request_id="dsar-1", ) def test_oversized_content_fails_closed(self) -> None: notification = self.session.get( NotificationMessage, "notification-inbox", ) notification.body_text = "x" * 100_001 self.session.flush() with self.assertRaisesRegex(ValueError, "body text exceeds"): self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( membership_id="membership-1", external_references={ "notifications.notification": "notification-inbox" }, ), ) def test_core_workflow_and_manifest_register_provider(self) -> None: row = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-NOTIFICATIONS-1", request_kind="access", subject=self._subject(), purpose="Respond to a verified request.", legal_basis="Article 15 GDPR", due_at=None, requested_by_account_id="privacy-officer", ) self.session.commit() search_data_subject_request( self.session, registry=_Registry(self.provider), row=row, expected_revision=1, ) self.assertEqual( [NOTIFICATIONS_DSAR_CAPABILITY], row.coverage["provider_capabilities"], ) self.assertEqual(3, row.search_result["record_count"]) inactive = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-NOTIFICATIONS-2", request_kind="access", subject=self._subject(), purpose="Respond to a verified request.", legal_basis="Article 15 GDPR", due_at=None, requested_by_account_id="privacy-officer", ) self.session.commit() search_data_subject_request( self.session, registry=_Registry(self.provider, active=False), row=inactive, expected_revision=1, ) self.assertEqual( [NOTIFICATIONS_DSAR_CAPABILITY], inactive.coverage["inactive_provider_capabilities"], ) self.assertIn(NOTIFICATIONS_DSAR_CAPABILITY, manifest.capability_factories) self.assertIn( NOTIFICATIONS_DSAR_CAPABILITY, manifest.capability_documentation, ) self.assertIn( NOTIFICATIONS_DSAR_CAPABILITY, {item.name for item in manifest.provides_interfaces}, ) self.assertTrue( any( topic.id == "notifications.data-subject-requests" and {"admin", "user"}.issubset(topic.documentation_types) for topic in manifest.documentation ) ) if __name__ == "__main__": unittest.main()