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 ( 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_datasources.backend.db.models import ( DatasourceGovernanceReferenceRecord, DatasourceMaterializationRecord, DatasourcePayloadRecord, DatasourcePayloadRowRecord, DatasourcePublicationRecord, DatasourceRecord, DatasourceStageRecord, ) from govoplan_datasources.backend.dsar_provider import ( DATASOURCES_DSAR_CAPABILITY, DatasourcesDsarProvider, ) from govoplan_datasources.backend.manifest import manifest NOW = datetime(2026, 8, 21, 19, 0, tzinfo=UTC) SECRET = "private-datasource-detail-do-not-export" class _Registry: def __init__( self, provider: DatasourcesDsarProvider, *, active: bool = True, ) -> None: self.provider = provider self.active = active def capability_names(self): return (DATASOURCES_DSAR_CAPABILITY,) def capability_owner(self, name): self._assert_capability(name) return "datasources" 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": ("datasources",) 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": "datasources"})(),) @staticmethod def _assert_capability(name: str) -> None: if name != DATASOURCES_DSAR_CAPABILITY: raise KeyError(name) class DatasourcesDsarProviderTests(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 = DatasourcesDsarProvider() self.assertIsInstance(self.provider, DsarProvider) self._seed() self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def _seed(self) -> None: datasource = self._datasource("datasource-1", "tenant-1") other = self._datasource("datasource-other", "tenant-2") self.session.add_all((datasource, other)) self.session.flush() referenced_payload = self._payload( "payload-referenced", "tenant-1", created_by="account-1", ) orphan_payload = self._payload( "payload-orphan", "tenant-1", created_by="account-1", ) self.session.add_all((referenced_payload, orphan_payload)) self.session.flush() self.session.add( DatasourcePayloadRowRecord( payload_id=referenced_payload.id, row_index=0, row_={"secret": SECRET}, checksum="a" * 64, ) ) materialization = DatasourceMaterializationRecord( id="materialization-1", tenant_id="tenant-1", datasource_id=datasource.id, revision=1, state="published", schema_version=1, schema_=[{"secret": SECRET}], payload_id=referenced_payload.id, payload_checksum="b" * 64, rows=[{"secret": SECRET}], fingerprint="c" * 64, row_count=1, byte_count=100, source_timestamp=NOW, provenance_={"secret": SECRET}, metadata_={"secret": SECRET}, governance_snapshot_={"secret": SECRET}, created_by="account-1", ) self.session.add(materialization) self.session.flush() datasource.current_materialization_id = materialization.id self.session.add_all( ( DatasourceGovernanceReferenceRecord( id="governance-1", tenant_id="tenant-1", datasource_id=datasource.id, relation="authoritative_source", reference=SECRET, ), self._stage( "stage-1", target_datasource_id=datasource.id, promoted=False, ), self._stage( "stage-promoted", target_datasource_id=datasource.id, promoted=True, ), DatasourcePublicationRecord( id="publication-1", tenant_id="tenant-1", producer_module="dataflow", producer_run_ref=SECRET, idempotency_key=SECRET, request_hash="d" * 64, datasource_id=datasource.id, materialization_id=materialization.id, status="published", details_={"secret": SECRET}, created_by="account-1", ), ) ) @staticmethod def _datasource(row_id: str, tenant_id: str) -> DatasourceRecord: return DatasourceRecord( id=row_id, tenant_id=tenant_id, source_name=f"source-{row_id}", name=SECRET, description=SECRET, kind="table", mode="cached", shape="tabular", status="active", provider="connector.provider", provider_ref=SECRET, schema_version=1, schema_=[{"secret": SECRET}], fingerprint="e" * 64, row_count=1, byte_count=100, provenance_={"secret": SECRET}, metadata_={"secret": SECRET}, owner_ref=SECRET, steward_ref=SECRET, authoritative_source_ref=SECRET, authority_mode="external_mirror", legal_basis_refs=[SECRET], purposes=[SECRET], semantic_definition=SECRET, official_keys=[SECRET], classification="personal", privacy_profile_ref=SECRET, retention_policy_ref=SECRET, hold_refs=[SECRET], publication_state="published", transfer_agreement_ref=SECRET, freshness_policy={"secret": SECRET}, quality_policy={"secret": SECRET}, known_limits=[SECRET], correction_procedure_ref=SECRET, affected_refs=[SECRET], dependency_refs=[SECRET], created_by="account-1", updated_by="account-1", ) @staticmethod def _payload( row_id: str, tenant_id: str, *, created_by: str, ) -> DatasourcePayloadRecord: return DatasourcePayloadRecord( id=row_id, tenant_id=tenant_id, backend="database_rows", state="published", locator=SECRET, media_type="application/x-ndjson", checksum="f" * 64, row_count=1, byte_count=100, checkpoint_={"secret": SECRET}, metadata_={"secret": SECRET}, created_by=created_by, ) @staticmethod def _stage( row_id: str, *, target_datasource_id: str, promoted: bool, ) -> DatasourceStageRecord: return DatasourceStageRecord( id=row_id, tenant_id="tenant-1", target_datasource_id=target_datasource_id, name=SECRET, source_name=SECRET, description=SECRET, kind="table", mode="static", shape="tabular", state="promoted" if promoted else "ready", provider="upload", provider_ref=SECRET, schema_=[{"secret": SECRET}], rows=[{"secret": SECRET}], fingerprint="0" * 64, row_count=1, byte_count=100, validation_={"secret": SECRET}, provenance_={"secret": SECRET}, metadata_={"secret": SECRET}, governance_={"secret": SECRET}, promoted_at=NOW if promoted else None, promoted_materialization_id="materialization-1" if promoted else None, created_by="account-1", ) def test_canonical_selector_exports_only_minimized_attribution(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(account_id="account-1"), ) self.assertEqual(7, len(records)) self.assertEqual( {"datasource_operator_attribution"}, {record.category for record in records}, ) exported = json.dumps([record.to_dict() for record in records]) self.assertNotIn(SECRET, exported) self.assertNotIn("account-1", exported) self.assertNotIn("datasource-other", exported) def test_exact_datasource_returns_minimized_lifecycle_package(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={ "datasources.datasource": "datasource:datasource-1" }, ), ) self.assertEqual(7, len(records)) self.assertEqual( { "datasource", "datasource_governance_reference", "datasource_materialization", "datasource_payload", "datasource_stage", "datasource_publication", }, {record.resource_type for record in records}, ) exported = json.dumps([record.to_dict() for record in records]) self.assertNotIn(SECRET, exported) self.assertNotIn("payload-orphan", exported) actions = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={"datasources.datasource": "datasource-1"} ), records=records, ) self.assertEqual({"manual_review"}, {action.kind for action in actions}) def test_exact_references_conflicts_and_tenants_fail_closed(self) -> None: references = { "datasources.governance_reference": "governance-1", "datasources.materialization": "materialization-1", "datasources.payload": "payload-referenced", "datasources.stage": "stage-1", "datasources.publication": "publication-1", } for key, value in references.items(): with self.subTest(key=key): records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(external_references={key: value}), ) self.assertEqual(1, len(records)) mismatch = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-2", external_references={"datasources.stage": "stage-1"}, ), ) wrong_tenant = self.provider.search_subject( self.session, tenant_id="tenant-2", subject=DsarSubjectRef( external_references={"datasources.stage": "stage-1"} ), ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={ "datasources.datasource": "datasource-1", "datasources.catalogue": "different", } ), ) self.assertEqual((), mismatch) self.assertEqual((), wrong_tenant) self.assertEqual((), conflict) def test_transient_deletion_is_safe_and_idempotent(self) -> None: subject = DsarSubjectRef( external_references={ "datasources.stage": "stage-1", "datasources.payload": "payload-orphan", } ) 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"}, {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-retry", ) self.assertTrue(all(result.status == "executed" for result in first)) self.assertTrue(all(result.status == "unchanged" for result in second)) self.assertIsNone(self.session.get(DatasourceStageRecord, "stage-1")) self.assertIsNone(self.session.get(DatasourcePayloadRecord, "payload-orphan")) self.assertIsNotNone( self.session.get(DatasourcePayloadRecord, "payload-referenced") ) def test_published_and_attribution_state_requires_review_or_retention(self) -> None: direct_subject = DsarSubjectRef( external_references={ "datasources.materialization": "materialization-1", "datasources.payload": "payload-referenced", "datasources.stage": "stage-promoted", } ) direct = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=direct_subject, ) direct_actions = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=direct_subject, records=direct, ) self.assertEqual( {"manual_review"}, {action.kind for action in direct_actions}, ) canonical_subject = DsarSubjectRef(account_id="account-1") canonical = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=canonical_subject, ) canonical_actions = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=canonical_subject, records=canonical, ) self.assertEqual({"retain"}, {action.kind for action in canonical_actions}) results = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=canonical_subject, actions=canonical_actions, request_id="dsar-2", ) self.assertTrue(all(result.status == "blocked" for result in results)) def test_foreign_records_and_actions_are_rejected(self) -> None: subject = DsarSubjectRef(account_id="account-1") with self.assertRaisesRegex(ValueError, "foreign provider record"): self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=subject, records=( DsarRecordRef( provider_id="cases", module_id="cases", resource_type="case", resource_id="case-1", category="case", title="Case", ), ), ) with self.assertRaisesRegex(ValueError, "foreign provider action"): self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=( DsarErasureActionRef( action_id="cases:delete:case:case-1", provider_id="cases", module_id="cases", kind="delete", resource_type="case", resource_id="case-1", title="Delete case", rationale="Foreign", executable=True, ), ), request_id="dsar-3", ) def test_core_workflow_reports_active_and_inactive_provider(self) -> None: row = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-DATASOURCES-1", request_kind="access_and_erasure", subject=DsarSubjectRef(account_id="account-1"), purpose="Respond to a verified request.", legal_basis="Article 15 and 17 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( [DATASOURCES_DSAR_CAPABILITY], row.coverage["provider_capabilities"], ) self.assertEqual(7, row.search_result["record_count"]) inactive = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-DATASOURCES-2", request_kind="access", subject=DsarSubjectRef(account_id="account-1"), 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([], inactive.coverage["provider_capabilities"]) self.assertEqual( [DATASOURCES_DSAR_CAPABILITY], inactive.coverage["inactive_provider_capabilities"], ) self.assertEqual(0, inactive.search_result["record_count"]) def test_manifest_registers_and_documents_capability(self) -> None: self.assertIn(DATASOURCES_DSAR_CAPABILITY, manifest.capability_factories) self.assertIn( DATASOURCES_DSAR_CAPABILITY, manifest.capability_documentation, ) self.assertIn( DATASOURCES_DSAR_CAPABILITY, {item.name for item in manifest.provides_interfaces}, ) self.assertTrue( any( topic.id == "datasources.data-subject-requests" and {"admin", "user"}.issubset(topic.documentation_types) for topic in manifest.documentation ) ) if __name__ == "__main__": unittest.main()