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_services.backend.db.models import ServiceDefinitionRevision from govoplan_services.backend.dsar_provider import ( SERVICES_DSAR_CAPABILITY, ServicesDsarProvider, ) from govoplan_services.backend.manifest import manifest NOW = datetime(2026, 8, 21, 16, 0, tzinfo=UTC) class _Registry: def __init__(self, provider: ServicesDsarProvider, *, active: bool = True) -> None: self.provider = provider self.active = active def capability_names(self): return (SERVICES_DSAR_CAPABILITY,) def capability_owner(self, name): self._assert_capability(name) return "services" 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": ("services",) 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": "services"})(),) @staticmethod def _assert_capability(name: str) -> None: if name != SERVICES_DSAR_CAPABILITY: raise KeyError(name) class ServicesDsarProviderTests(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 = ServicesDsarProvider() self.assertIsInstance(self.provider, DsarProvider) self._seed() self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def _seed(self) -> None: self.session.add_all( ( ServiceDefinitionRevision( id="revision-1", tenant_id="tenant-1", service_id="service-1", service_key="permit", revision="1", publication_state="published", valid_from=NOW, recorded_at=NOW, superseded_at=NOW + timedelta(minutes=1), search_text="private-search-text-do-not-export", payload={"secret": "private-payload-do-not-export"}, created_by="account-1", ), ServiceDefinitionRevision( id="revision-2", tenant_id="tenant-1", service_id="service-1", service_key="permit", revision="2", previous_revision_id="revision-1", publication_state="suspended", valid_from=NOW, recorded_at=NOW + timedelta(minutes=1), search_text="private-search-text-do-not-export", payload={"secret": "private-payload-do-not-export"}, created_by="account-1", ), ServiceDefinitionRevision( id="revision-other", tenant_id="tenant-1", service_id="service-other", service_key="other", revision="1", publication_state="published", valid_from=NOW, recorded_at=NOW, search_text="unrelated-private-search", payload={"secret": "unrelated-private-payload"}, created_by="account-other", ), ServiceDefinitionRevision( id="revision-tenant-2", tenant_id="tenant-2", service_id="service-tenant-2", service_key="tenant-2", revision="1", publication_state="published", valid_from=NOW, recorded_at=NOW, search_text="tenant-2-private-search", payload={"secret": "tenant-2-private-payload"}, created_by="account-1", ), ) ) def test_actor_search_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(2, len(records)) self.assertEqual( {"service_definition_attribution"}, {record.resource_type for record in records}, ) exported = json.dumps([record.to_dict() for record in records]) self.assertNotIn("private-payload-do-not-export", exported) self.assertNotIn("private-search-text-do-not-export", exported) self.assertNotIn("service-other", exported) self.assertNotIn("service-tenant-2", exported) self.assertTrue(all(record.immutable_evidence for record in records)) def test_definition_and_revision_references_narrow_and_corroborate(self) -> None: by_definition = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"services.definition": "service-1"}, ), ) by_revision = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"services.revision": "revision-2"}, ), ) actor_conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-other", external_references={"services.definition": "service-1"}, ), ) reference_without_actor = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={"services.definition": "service-1"} ), ) alias_conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={ "services.definition": "service-1", "services.service": "service-other", }, ), ) self.assertEqual(2, len(by_definition)) self.assertEqual(["revision-2"], [record.resource_id for record in by_revision]) self.assertEqual((), actor_conflict) self.assertEqual((), reference_without_actor) self.assertEqual((), alias_conflict) def test_planning_is_retain_only_and_execution_is_blocked(self) -> None: subject = DsarSubjectRef(account_id="account-1") 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.assertTrue(actions) self.assertTrue(all(action.kind == "retain" for action in actions)) self.assertTrue(all(not action.executable for action in actions)) results = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=actions, request_id="dsar-1", ) 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:retain:case:case-1", provider_id="cases", module_id="cases", kind="retain", resource_type="case", resource_id="case-1", title="Retain case", rationale="Evidence", executable=False, ), ), 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-SERVICES-1", 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), row=row, expected_revision=1, ) self.assertEqual( [SERVICES_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-SERVICES-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( [SERVICES_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(SERVICES_DSAR_CAPABILITY, manifest.capability_factories) self.assertIn(SERVICES_DSAR_CAPABILITY, manifest.capability_documentation) self.assertIn( SERVICES_DSAR_CAPABILITY, {item.name for item in manifest.provides_interfaces}, ) self.assertTrue( any( topic.id == "services.data-subject-requests" and {"admin", "user"}.issubset(topic.documentation_types) for topic in manifest.documentation ) ) if __name__ == "__main__": unittest.main()