from __future__ import annotations import json import unittest from datetime import UTC, datetime, timedelta 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_search.backend.db.models import ( SearchIndexAclToken, SearchIndexChangeQueue, SearchIndexDocument, ) from govoplan_search.backend.dsar_provider import ( SEARCH_DSAR_CAPABILITY, SearchDsarProvider, ) from govoplan_search.backend.manifest import manifest NOW = datetime(2026, 8, 21, 17, 0, tzinfo=UTC) class _Registry: def __init__(self, provider: SearchDsarProvider, *, active: bool = True) -> None: self.provider = provider self.active = active def capability_names(self): return (SEARCH_DSAR_CAPABILITY,) def capability_owner(self, name): self._assert_capability(name) return "search" 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": ("search",) 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": "search"})(),) @staticmethod def _assert_capability(name: str) -> None: if name != SEARCH_DSAR_CAPABILITY: raise KeyError(name) class SearchDsarProviderTests(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 = SearchDsarProvider() self.assertIsInstance(self.provider, DsarProvider) self._seed() self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def _document( self, *, row_id: str, tenant_id: str, resource_id: str, ) -> SearchIndexDocument: return SearchIndexDocument( id=row_id, tenant_id=tenant_id, module_id="cases", provider_id="cases.search", resource_type="case", resource_id=resource_id, title="private-title-do-not-export", summary="private-summary-do-not-export", body="private-body-do-not-export", keywords=["private-keyword-do-not-export"], search_text="private-search-text-do-not-export", url="/private-url-do-not-export", visibility="restricted", external_reference={"secret": "private-reference-do-not-export"}, metadata_={"secret": "private-metadata-do-not-export"}, content_hash="a" * 64, source_revision="revision-1", change_cursor="private-cursor-do-not-export", source_updated_at=NOW, language="simple", index_version=1, requires_authorization_recheck=True, rebuild_id="private-rebuild-id-do-not-export", active=True, indexed_at=NOW, ) def _seed(self) -> None: document = self._document( row_id="document-1", tenant_id="tenant-1", resource_id="case-1", ) document.acl_tokens.extend( ( SearchIndexAclToken( id="acl-1", token="account:account-1", ), SearchIndexAclToken( id="acl-2", token="identity:identity-1", ), ) ) unrelated = self._document( row_id="document-2", tenant_id="tenant-1", resource_id="case-2", ) unrelated.acl_tokens.append( SearchIndexAclToken( id="acl-3", token="account:account-1", ) ) other_tenant = self._document( row_id="document-3", tenant_id="tenant-2", resource_id="case-1", ) other_tenant.acl_tokens.append( SearchIndexAclToken( id="acl-4", token="account:account-1", ) ) self.session.add_all((document, unrelated, other_tenant)) self.session.add_all( ( SearchIndexChangeQueue( id="change-1", change_id="change-public-1", tenant_id="tenant-1", provider_id="cases.search", module_id="cases", resource_type="case", resource_id="case-1", kind="upsert", source_revision="revision-1", source_cursor="private-source-cursor-do-not-export", document_={"secret": "private-queue-document-do-not-export"}, occurred_at=NOW, status="queued", attempts=1, available_at=NOW, error="private-error-do-not-export", ), SearchIndexChangeQueue( id="change-2", change_id="change-public-2", tenant_id="tenant-1", provider_id="cases.search", module_id="cases", resource_type="case", resource_id="case-2", kind="upsert", source_revision="revision-1", source_cursor="unrelated-private-cursor", document_={"secret": "unrelated-private-document"}, occurred_at=NOW + timedelta(minutes=1), status="queued", attempts=0, available_at=NOW, ), ) ) def test_source_reference_exports_only_minimized_derived_rows(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(external_references={"cases.case": "case-1"}), ) self.assertEqual( {"search_index_document", "search_index_change"}, {record.resource_type for record in records}, ) exported = json.dumps([record.to_dict() for record in records]) self.assertIn("case-1", exported) self.assertNotIn("case-2", exported) for secret in ( "private-title-do-not-export", "private-summary-do-not-export", "private-body-do-not-export", "private-keyword-do-not-export", "private-search-text-do-not-export", "private-url-do-not-export", "private-reference-do-not-export", "private-metadata-do-not-export", "private-cursor-do-not-export", "private-rebuild-id-do-not-export", "private-source-cursor-do-not-export", "private-queue-document-do-not-export", "private-error-do-not-export", ): self.assertNotIn(secret, exported) def test_canonical_selectors_export_acl_projection_not_source_content(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", identity_id="identity-1", ), ) self.assertEqual( {"search_acl_projection"}, {record.resource_type for record in records}, ) self.assertEqual(3, len(records)) exported = json.dumps([record.to_dict() for record in records]) self.assertNotIn("account-1", exported) self.assertNotIn("identity-1", exported) self.assertNotIn("private-title-do-not-export", exported) self.assertNotIn("document-3", exported) def test_direct_references_and_conflicts_fail_closed(self) -> None: direct = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={"search.document": "document-1"} ), ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={ "search.document": "document-1", "cases.case": "case-2", } ), ) alias_conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={ "search.document": "document-1", "search.index_document": "document-2", } ), ) wrong_tenant = self.provider.search_subject( self.session, tenant_id="tenant-2", subject=DsarSubjectRef( external_references={"search.document": "document-1"} ), ) wrong_resource_type = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(external_references={"cases.ticket": "case-1"}), ) change_and_account = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"search.change": "change-1"}, ), ) self.assertEqual(["document-1"], [record.resource_id for record in direct]) self.assertEqual((), conflict) self.assertEqual((), alias_conflict) self.assertEqual((), wrong_tenant) self.assertEqual((), wrong_resource_type) self.assertEqual( {"change-1", "acl-1"}, {record.resource_id for record in change_and_account}, ) def test_derived_deletion_is_idempotent_but_acl_review_is_blocked(self) -> None: direct_subject = DsarSubjectRef(external_references={"cases.case": "case-1"}) direct_records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=direct_subject, ) actions = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=direct_subject, records=direct_records, ) self.assertTrue(actions) self.assertTrue(all(action.kind == "delete" for action in actions)) self.assertTrue(all(action.executable for action in actions)) first = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=direct_subject, actions=actions, request_id="dsar-1", ) second = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=direct_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(SearchIndexDocument, "document-1")) self.assertIsNone(self.session.get(SearchIndexChangeQueue, "change-1")) acl_subject = DsarSubjectRef(account_id="account-1") acl_records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=acl_subject, ) acl_actions = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=acl_subject, records=acl_records, ) self.assertTrue(all(action.kind == "manual_review" for action in acl_actions)) self.assertTrue(all(not action.executable for action in acl_actions)) results = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=acl_subject, actions=acl_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(external_references={"cases.case": "case-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-1", ) def test_core_workflow_reports_active_and_inactive_provider(self) -> None: row = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-SEARCH-1", request_kind="access_and_erasure", subject=DsarSubjectRef(external_references={"cases.case": "case-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( [SEARCH_DSAR_CAPABILITY], row.coverage["provider_capabilities"] ) self.assertEqual(2, row.search_result["record_count"]) inactive = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-SEARCH-2", request_kind="access", subject=DsarSubjectRef(external_references={"cases.case": "case-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( [SEARCH_DSAR_CAPABILITY], inactive.coverage["inactive_provider_capabilities"], ) self.assertEqual(0, inactive.search_result["record_count"]) def test_manifest_registers_and_documents_the_capability(self) -> None: self.assertIn(SEARCH_DSAR_CAPABILITY, manifest.capability_factories) self.assertIn(SEARCH_DSAR_CAPABILITY, manifest.capability_documentation) self.assertIn( SEARCH_DSAR_CAPABILITY, {item.name for item in manifest.provides_interfaces}, ) self.assertTrue( any( topic.id == "search.data-subject-requests" and {"admin", "user"}.issubset(topic.documentation_types) for topic in manifest.documentation ) ) if __name__ == "__main__": unittest.main()