from __future__ import annotations import json import unittest from datetime import UTC, datetime from sqlalchemy import create_engine from sqlalchemy.orm import Session from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef from govoplan_core.db.base import Base from govoplan_dist_lists.backend.db.models import ( DistributionList, DistributionListEntry, DistributionListRevision, DistributionListSnapshot, ) from govoplan_dist_lists.backend.dsar_provider import ( DIST_LISTS_DSAR_CAPABILITY, DistributionListsDsarProvider, ) from govoplan_dist_lists.backend.manifest import manifest NOW = datetime(2026, 8, 22, 16, 0, tzinfo=UTC) def _recipient( *, key: str, email: str, account_id: str | None = None, identity_id: str | None = None, ) -> dict[str, object]: return { "recipient_key": key, "display_name": f"Recipient {email}", "status": "usable", "channels": [ { "channel": "email", "target": email, "target_key": f"email:{email.casefold()}", "status": "usable", "contact_point_id": f"contact-point-{key}", "decision_provenance": {"secret": "do-not-export-provenance"}, } ], "account_id": account_id, "identity_id": identity_id, "contact_id": None, "source_entry_ids": ["entry-1"], "attributes": {"secret": "do-not-export-attributes"}, "provenance": {"secret": "do-not-export-recipient-provenance"}, } class DistributionListsDsarProviderTests(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 = DistributionListsDsarProvider() self._seed() self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def _seed(self) -> None: distribution_list = DistributionList( id="list-1", tenant_id="tenant-1", scope_type="user", scope_id="account-1", name="Sensitive list name do not export", description="Sensitive description do not export", status="active", current_revision_id="revision-1", current_revision=1, resource_revision=1, created_by_account_id="account-1", updated_by_account_id="account-1", metadata_={"secret": "list-metadata-do-not-export"}, created_at=NOW, updated_at=NOW, ) revision = DistributionListRevision( id="revision-1", tenant_id="tenant-1", distribution_list_id="list-1", revision=1, definition_kind="static", definition_hash="definition-hash-do-not-export", parameter_schema=[{"secret": "parameter-do-not-export"}], constraints={"secret": "constraint-do-not-export"}, source_fingerprints=[{"secret": "fingerprint-do-not-export"}], created_by_account_id="account-1", created_at=NOW, updated_at=NOW, ) entry = DistributionListEntry( id="entry-1", tenant_id="tenant-1", revision_id="revision-1", entry_key="subject-email", kind="raw_email", mode="include", source_provider="local", source_resource_type="email", source_resource_id="Subject@Example.test", source_revision="source-revision-do-not-export", source_fingerprint="source-fingerprint-do-not-export", source_label="source-label-do-not-export", source_metadata={"secret": "source-metadata-do-not-export"}, label="entry-label-do-not-export", purpose="service_notice", requested_channels=["email"], order_index=0, configuration={"secret": "configuration-do-not-export"}, created_at=NOW, updated_at=NOW, ) snapshot = DistributionListSnapshot( id="snapshot-1", tenant_id="tenant-1", distribution_list_id="list-1", revision_id="revision-1", revision_number=1, idempotency_key="idempotency-key-do-not-export", request_={"secret": "request-do-not-export"}, expansion_hash="expansion-hash-do-not-export", effective_at=NOW, recipient_count=2, excluded_count=0, recipients=[ _recipient( key="identity-1", email="subject@example.test", account_id="account-1", identity_id="identity-1", ), _recipient( key="identity-other", email="other-person-do-not-export@example.test", account_id="account-other-do-not-export", identity_id="identity-other-do-not-export", ), ], excluded=[], diagnostics=[{"secret": "diagnostics-do-not-export"}], provider_evidence=[{"secret": "provider-evidence-do-not-export"}], stale=False, truncated=False, created_by_account_id="account-1", provenance={"secret": "snapshot-provenance-do-not-export"}, created_at=NOW, updated_at=NOW, ) other_tenant = DistributionList( id="list-other-tenant", tenant_id="tenant-2", scope_type="user", scope_id="account-1", name="Other tenant do not export", status="active", current_revision_id="revision-other", current_revision=1, resource_revision=1, ) self.session.add_all( (distribution_list, revision, entry, snapshot, other_tenant) ) def test_search_exports_only_exact_subject_projection(self) -> None: self.assertIsInstance(self.provider, DsarProvider) records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", identity_id="identity-1", email=" SUBJECT@example.test ", ), ) self.assertEqual( { "distribution_list_subject_scope", "distribution_list_definition_entry", "distribution_list_snapshot_recipient", "distribution_list_actor_attribution", "distribution_list_revision_actor_attribution", "distribution_list_snapshot_actor_attribution", }, {record.resource_type for record in records}, ) exported = json.dumps([record.to_dict() for record in records]) self.assertIn("subject@example.test", exported.casefold()) for excluded in ( "other-person-do-not-export@example.test", "account-other-do-not-export", "identity-other-do-not-export", "list-metadata-do-not-export", "definition-hash-do-not-export", "parameter-do-not-export", "constraint-do-not-export", "fingerprint-do-not-export", "source-revision-do-not-export", "source-fingerprint-do-not-export", "source-label-do-not-export", "source-metadata-do-not-export", "configuration-do-not-export", "idempotency-key-do-not-export", "request-do-not-export", "expansion-hash-do-not-export", "diagnostics-do-not-export", "provider-evidence-do-not-export", "snapshot-provenance-do-not-export", "do-not-export-provenance", "do-not-export-attributes", ): self.assertNotIn(excluded, exported) def test_resource_references_only_narrow_verified_subjects(self) -> None: self.assertEqual( (), self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={"dist_lists.snapshot": "snapshot-1"} ), ), ) narrowed = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( email="subject@example.test", external_references={"dist_lists.snapshot": "snapshot-1"}, ), ) self.assertEqual( ["distribution_list_snapshot_recipient"], [record.resource_type for record in narrowed], ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( email="subject@example.test", external_references={"dist_lists.email": "other@example.test"}, ), ) self.assertEqual((), conflict) def test_definition_change_requires_review_and_snapshot_is_retained(self) -> None: subject = DsarSubjectRef(email="subject@example.test") 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, ) by_type = {action.resource_type: action.kind for action in actions} self.assertEqual("manual_review", by_type["distribution_list_definition_entry"]) self.assertEqual("retain", by_type["distribution_list_snapshot_recipient"]) def test_manifest_registers_provider_and_documentation(self) -> None: self.assertIn(DIST_LISTS_DSAR_CAPABILITY, manifest.capability_factories) self.assertIn( "dist_lists.data-subject-requests", {topic.id for topic in manifest.documentation}, ) if __name__ == "__main__": unittest.main()