from __future__ import annotations import unittest from datetime import datetime, timedelta, timezone from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from govoplan_access.backend.db.models import Account, Group, User from govoplan_core.core.change_sequence import ChangeSequenceEntry from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef from govoplan_core.db.base import Base from govoplan_core.privacy.dsar_workflow import ( DataSubjectRequest, create_data_subject_request, plan_data_subject_erasure, search_data_subject_request, ) from govoplan_mail.backend.db.models import ( MailBounceObservation, MailDeliveryAttempt, MailDeliveryCommand, MailDeliveryReconciliation, MailMailboxMessageIndex, MailPop3Import, MailServerEndpoint, MailServerProfile, ) from govoplan_mail.backend.dsar_provider import MAIL_DSAR_CAPABILITY, MailDsarProvider from govoplan_mail.backend.manifest import manifest class _Registry: def __init__(self, provider, active=True): self.provider = provider self.active = active def capability_names(self): return (MAIL_DSAR_CAPABILITY,) def capability_owner(self, name): assert name == MAIL_DSAR_CAPABILITY return "mail" 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": ("mail",) if active else ()} )() return Resolver() def require_tenant_capability(self, name, session, **kwargs): del session, kwargs assert name == MAIL_DSAR_CAPABILITY return self.provider class MailDsarProviderTests(unittest.TestCase): def setUp(self): self.engine = create_engine("sqlite:///:memory:", future=True) Base.metadata.create_all( self.engine, tables=[ Account.__table__, User.__table__, Group.__table__, ChangeSequenceEntry.__table__, DataSubjectRequest.__table__, MailServerProfile.__table__, MailServerEndpoint.__table__, MailMailboxMessageIndex.__table__, MailPop3Import.__table__, MailDeliveryCommand.__table__, MailDeliveryAttempt.__table__, MailDeliveryReconciliation.__table__, MailBounceObservation.__table__, ], ) self.session = sessionmaker(bind=self.engine, future=True)() now = datetime.now(timezone.utc) account = Account( id="account-1", email="subject@example.test", normalized_email="subject@example.test", display_name="Subject", ) user = User( id="membership-1", tenant_id="tenant-1", account_id=account.id, email="subject@example.test", display_name="Subject", ) profile = MailServerProfile( id="profile-subject", tenant_id="tenant-1", scope_type="user", scope_id=user.id, name="Personal mail", slug="personal", smtp_config={"host": "smtp-secret-do-not-export"}, smtp_username="smtp-user-do-not-export", smtp_password_encrypted="smtp-cipher-do-not-export", imap_config={"host": "imap-secret-do-not-export"}, imap_username="imap-user-do-not-export", imap_password_encrypted="imap-cipher-do-not-export", created_by_user_id=user.id, ) message = MailMailboxMessageIndex( id="message-subject", tenant_id="tenant-1", profile_id=profile.id, folder="INBOX-secret-do-not-export", uid="uid-secret-do-not-export", uid_int=1, sort_position=1, subject="Subject notice", from_header="Office ", to_header="Subject Person ", cc_header="Unrelated Person ", date="2026-08-20", message_id="message-locator-do-not-export", flags=["\\Seen"], size_bytes=42, body_preview="Message preview for the subject", attachment_count=1, indexed_at=now, ) unrelated = MailMailboxMessageIndex( id="message-other", tenant_id="tenant-1", profile_id=profile.id, folder="INBOX", uid="2", uid_int=2, sort_position=2, subject="Unrelated message do not export", from_header="other@example.test", to_header="someone@example.test", indexed_at=now, ) tenant_two_profile = MailServerProfile( id="profile-tenant-2", tenant_id="tenant-2", name="Tenant two", slug="tenant-two", smtp_config={}, ) tenant_two = MailMailboxMessageIndex( id="message-tenant-2", tenant_id="tenant-2", profile_id=tenant_two_profile.id, folder="INBOX", uid="1", uid_int=1, sort_position=1, subject="Tenant two message do not export", to_header="subject@example.test", indexed_at=now, ) pop3_server = MailServerEndpoint( id="pop3-server-subject", profile_id=profile.id, tenant_id="tenant-1", protocol="pop3", name="Legacy POP3", config={"host": "pop3-secret-do-not-export"}, scope_type="tenant", scope_id="tenant-1", ) pop3_import = MailPop3Import( id="pop3-import-subject", tenant_id="tenant-1", profile_id=profile.id, pop3_server_id=pop3_server.id, pop3_credential_id="credential-secret-do-not-export", transport_revision="pop3-revision-secret-do-not-export", provider_uidl="provider-uidl-secret-do-not-export", fingerprint="e" * 64, raw_sha256="f" * 64, raw_message_encrypted="pop3-message-cipher-do-not-export", message_id="pop3-message-id", subject="Imported subject notice", from_header="Legacy office ", to_header="Subject Person ", date="2026-08-19", body_preview="Imported message preview for the subject", size_bytes=84, status="pending_review", imported_at=now, deletion_requested=False, deletion_status="not_requested", ) command = MailDeliveryCommand( id="command-subject", tenant_id="tenant-1", command_type="send", source_module="notifications", source_resource_type="notification", idempotency_key="idempotency-do-not-export", canonical_request_hash="a" * 64, profile_id=profile.id, expected_smtp_transport_revision="revision-secret", envelope_recipients_encrypted="recipient-cipher-do-not-export", message_encrypted="message-cipher-do-not-export", message_sha256="b" * 64, rfc_message_id="rfc-message-id", message_size_bytes=100, recipient_count=1, status="succeeded", attempt_count=1, accepted_count=1, created_by_user_id=user.id, completed_at=now, expires_at=now + timedelta(days=30), ) attempt = MailDeliveryAttempt( id="attempt-subject", command_id=command.id, attempt_number=1, worker_id="worker-do-not-export", status="succeeded", started_at=now, completed_at=now, accepted_count=1, diagnostic_summary="diagnostic-do-not-export", ) reconciliation = MailDeliveryReconciliation( id="reconciliation-subject", command_id=command.id, decision="confirmed_sent", evidence_reference="private-reference-do-not-export", note_encrypted="private-note-do-not-export", created_by_user_id=user.id, ) bounce = MailBounceObservation( id="bounce-subject", tenant_id="tenant-1", profile_id=profile.id, folder="bounce-folder-do-not-export", uid="bounce-uid-do-not-export", fingerprint="c" * 64, raw_sha256="d" * 64, original_message_id="original-id-do-not-export", command_id=command.id, recipient="subject@example.test", action="failed", status_code="5.1.1", diagnostic="bounce-diagnostic-do-not-export", permanent=True, observed_at=now, matched=True, evidence={"secret": "bounce-evidence-do-not-export"}, ) self.session.add_all( [ account, user, profile, message, unrelated, tenant_two_profile, tenant_two, pop3_server, pop3_import, command, attempt, reconciliation, bounce, ] ) self.session.commit() self.provider = MailDsarProvider() self.subject = DsarSubjectRef( membership_id=user.id, email="subject@example.test" ) def tearDown(self): self.session.close() self.engine.dispose() def test_manifest_and_minimized_tenant_scoped_search(self): self.assertIn( MAIL_DSAR_CAPABILITY, {item.name for item in manifest.provides_interfaces} ) self.assertIsInstance( manifest.capability_factories[MAIL_DSAR_CAPABILITY](None), DsarProvider ) records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self.subject ) self.assertTrue( { "mail_server_profile", "mailbox_message_index", "mail_delivery_command", "mail_delivery_attempt", "mail_delivery_reconciliation", "mail_bounce_observation", "mail_pop3_import", }.issubset({r.resource_type for r in records}) ) serialized = repr([record.to_dict() for record in records]) for hidden in ( "other@example.test", "Unrelated Person", "message-other", "Unrelated message do not export", "message-tenant-2", "Tenant two message do not export", "smtp-secret-do-not-export", "smtp-user-do-not-export", "smtp-cipher-do-not-export", "imap-secret-do-not-export", "imap-user-do-not-export", "imap-cipher-do-not-export", "INBOX-secret-do-not-export", "uid-secret-do-not-export", "message-locator-do-not-export", "idempotency-do-not-export", "recipient-cipher-do-not-export", "message-cipher-do-not-export", "worker-do-not-export", "diagnostic-do-not-export", "private-reference-do-not-export", "private-note-do-not-export", "bounce-folder-do-not-export", "bounce-uid-do-not-export", "original-id-do-not-export", "bounce-diagnostic-do-not-export", "bounce-evidence-do-not-export", "pop3-secret-do-not-export", "credential-secret-do-not-export", "pop3-revision-secret-do-not-export", "provider-uidl-secret-do-not-export", "pop3-message-cipher-do-not-export", ): self.assertNotIn(hidden, serialized) def test_conflicting_email_fails_closed(self): records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( email="subject@example.test", external_references={"mail.email": "other@example.test"}, ), ) self.assertEqual((), records) def test_plan_preserves_evidence_and_executes_nothing(self): records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self.subject ) actions = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=self.subject, records=records ) self.assertTrue({"retain", "manual_review"}.issubset({a.kind for a in actions})) self.assertFalse(any(action.executable for action in actions)) def test_core_workflow_discovers_active_and_skips_disabled_provider(self): request = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-MAIL-1", request_kind="access_and_erasure", subject=self.subject, purpose="Authorized request", legal_basis="GDPR", due_at=None, requested_by_account_id="officer", ) self.session.commit() search_data_subject_request( self.session, registry=_Registry(self.provider), row=request, expected_revision=1, ) self.assertEqual(["mail"], request.coverage["covered_modules"]) plan_data_subject_erasure( self.session, registry=_Registry(self.provider), row=request, expected_revision=2, ) self.assertFalse( any(action["executable"] for action in request.erasure_plan["actions"]) ) disabled = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-MAIL-OFF", request_kind="access", subject=self.subject, purpose="Coverage", legal_basis=None, due_at=None, requested_by_account_id="officer", ) search_data_subject_request( self.session, registry=_Registry(self.provider, active=False), row=disabled, expected_revision=1, ) self.assertEqual( [MAIL_DSAR_CAPABILITY], disabled.coverage["inactive_provider_capabilities"] ) if __name__ == "__main__": unittest.main()