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_reporting.backend.db.models import ( ReportingDefinitionGrant, ReportingDefinitionIdentity, ReportingDefinitionRevision, ReportingDrillContext, ReportingExecution, ReportingImportAssessment, ReportingProviderExecution, ReportingProviderExport, ReportingPublication, ReportingQualityResult, ReportingSavedView, ReportingSchedule, ) from govoplan_reporting.backend.dsar_provider import ( REPORTING_DSAR_CAPABILITY, ReportingDsarProvider, ) from govoplan_reporting.backend.manifest import manifest NOW = datetime(2026, 8, 21, 18, 0, tzinfo=UTC) SECRET = "personal-report-detail-do-not-export" class _Registry: def __init__(self, provider: ReportingDsarProvider, *, active: bool = True) -> None: self.provider = provider self.active = active def capability_names(self): return (REPORTING_DSAR_CAPABILITY,) def capability_owner(self, name): self._assert_capability(name) return "reporting" 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": ("reporting",) 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": "reporting"})(),) @staticmethod def _assert_capability(name: str) -> None: if name != REPORTING_DSAR_CAPABILITY: raise KeyError(name) class ReportingDsarProviderTests(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 = ReportingDsarProvider() self.assertIsInstance(self.provider, DsarProvider) self._seed() self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def _seed(self) -> None: identity = ReportingDefinitionIdentity( id="definition-row-1", tenant_id="tenant-1", definition_kind="report", definition_id="report-1", definition_key="resident-permits", created_by="account-1", ) self.session.add(identity) self.session.flush() self.session.add( ReportingDefinitionRevision( id="revision-row-1", tenant_id="tenant-1", identity_id=identity.id, definition_kind="report", definition_id="report-1", definition_key="resident-permits", revision=1, name=SECRET, description=SECRET, status="active", visibility="restricted", content_hash="a" * 64, change_reason=SECRET, idempotency_key=SECRET, request_sha256="b" * 64, event_id="event-1", recorded_at=NOW, payload={"secret": SECRET}, changed_by="account-1", ) ) self.session.add_all( ( self._execution( row_id="execution-row-1", tenant_id="tenant-1", execution_id="execution-1", actor_id="account-1", ), self._execution( row_id="execution-row-2", tenant_id="tenant-2", execution_id="execution-1", actor_id="account-1", ), ) ) provider_execution = ReportingProviderExecution( id="provider-row-1", tenant_id="tenant-1", execution_id="provider-execution-1", provider_id="cases.reports", report_id="resident-permits", report_revision="1", contract_version="1.0.0", idempotency_key=SECRET, request_sha256="c" * 64, purpose=SECRET, audience_scope={"secret": SECRET}, parameters={"secret": SECRET}, result_schema=[{"secret": SECRET}], result_payload={"secret": SECRET}, source_revisions=[{"secret": SECRET}], effective_scope={"secret": SECRET}, privacy_transforms=["small_cell_suppression"], provenance={"secret": SECRET}, governance_provenance={"secret": SECRET}, retention_class="short", retention_days=30, expires_at=NOW + timedelta(days=30), output_hash="d" * 64, generated_at=NOW, actor_id="account-1", ) self.session.add(provider_execution) self.session.flush() self.session.add_all( ( ReportingProviderExport( id="export-row-1", tenant_id="tenant-1", export_id="export-1", provider_execution_id=provider_execution.id, execution_id=provider_execution.execution_id, format="json", purpose=SECRET, audience_scope={"secret": SECRET}, output_hash="e" * 64, exported_at=NOW, actor_id="account-1", ), ReportingDefinitionGrant( id="grant-row-1", tenant_id="tenant-1", definition_kind="report", definition_id="report-1", subject_kind="account", subject_id="account-1", permissions=["view"], active=True, source_revision=1, ), ReportingSavedView( id="view-row-1", tenant_id="tenant-1", view_id="view-1", report_id="report-1", report_revision=1, owner_kind="account", owner_id="account-1", name=SECRET, state={"secret": SECRET}, shared=False, access={"secret": SECRET}, ), ReportingSavedView( id="view-row-2", tenant_id="tenant-1", view_id="view-shared", report_id="report-1", report_revision=1, owner_kind="account", owner_id="account-1", name=SECRET, state={"secret": SECRET}, shared=True, access={"secret": SECRET}, ), ReportingSchedule( id="schedule-row-1", tenant_id="tenant-1", schedule_id="schedule-1", report_id="report-1", report_revision=1, name=SECRET, trigger_kind="interval", trigger_config={"secret": SECRET}, parameters={"secret": SECRET}, query={"secret": SECRET}, publication_target={"secret": SECRET}, enabled=True, next_run_at=NOW + timedelta(days=1), created_by="account-1", ), ReportingPublication( id="publication-row-1", tenant_id="tenant-1", publication_id="publication-1", execution_id="execution-1", target_capability="files.artifact_store", target_ref=SECRET, format="json", status="succeeded", idempotency_key=SECRET, evidence={"secret": SECRET}, error=SECRET, completed_at=NOW, ), ReportingDrillContext( id="drill-row-1", tenant_id="tenant-1", drill_context_id="drill-1", execution_id="execution-1", token_sha256="f" * 64, context_sha256="0" * 64, actor_id="account-1", dimension_path=[{"secret": SECRET}], source_fingerprints=[{"secret": SECRET}], policy_provenance={"secret": SECRET}, expires_at=NOW + timedelta(minutes=10), ), ReportingQualityResult( id="quality-row-1", tenant_id="tenant-1", result_id="quality-1", quality_plan_id="quality-plan-1", quality_plan_revision=1, dataset_id="dataset-1", dataset_revision=1, status="passed", output_hash="1" * 64, assertions=[{"secret": SECRET}], source_fingerprints=[{"secret": SECRET}], evaluated_at=NOW, actor_id="account-1", ), ReportingImportAssessment( id="assessment-row-1", tenant_id="tenant-1", assessment_id="assessment-1", source_system="legacy-bi", source_id=SECRET, source_fingerprint="2" * 64, mapping_report={"secret": SECRET}, status="blocked", accepted_approximations=[SECRET], assessed_by="account-1", ), ) ) @staticmethod def _execution( *, row_id: str, tenant_id: str, execution_id: str, actor_id: str, ) -> ReportingExecution: return ReportingExecution( id=row_id, tenant_id=tenant_id, execution_id=execution_id, report_id="report-1", report_revision=1, semantic_model_id="semantic-1", semantic_model_revision=1, dataset_id="dataset-1", dataset_revision=1, status="succeeded", idempotency_key=SECRET, request_sha256="3" * 64, parameters={"secret": SECRET}, query={"secret": SECRET}, source_fingerprints=[{"secret": SECRET}], definition_hashes={"secret": SECRET}, output_hash="4" * 64, executor_version="reporting-v1", result_schema=[{"secret": SECRET}], result_rows=[{"secret": SECRET}], total_rows=1, truncated=False, diagnostics=[{"secret": SECRET}], provenance={"secret": SECRET}, started_at=NOW, finished_at=NOW, actor_id=actor_id, ) def test_canonical_selector_returns_minimized_owned_and_attribution_rows( self, ) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(account_id="account-1"), ) self.assertEqual(12, len(records)) self.assertIn( "subject_owned_reporting_view", {record.category for record in records}, ) self.assertIn( "reporting_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("execution-row-2", exported) def test_direct_references_are_exact_tenant_scoped_and_corroborated(self) -> None: direct = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"reporting.execution": "execution-1"}, ), ) mismatch = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-2", external_references={"reporting.execution": "execution-1"}, ), ) wrong_tenant = self.provider.search_subject( self.session, tenant_id="tenant-2", subject=DsarSubjectRef( external_references={"reporting.saved_view": "view-1"} ), ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={ "reporting.definition_revision": "revision-row-1", "reporting.revision": "different-revision", } ), ) self.assertEqual(["execution-row-1"], [item.resource_id for item in direct]) self.assertEqual("derived_report_result", direct[0].category) self.assertEqual((), mismatch) self.assertEqual((), wrong_tenant) self.assertEqual((), conflict) def test_every_exact_artifact_reference_is_supported(self) -> None: references = { "reporting.definition": ("report-1", 2), "reporting.definition_revision": ("revision-row-1", 1), "reporting.provider_execution": ("provider-execution-1", 1), "reporting.provider_export": ("export-1", 1), "reporting.definition_grant": ("grant-row-1", 1), "reporting.saved_view": ("view-1", 1), "reporting.schedule": ("schedule-1", 1), "reporting.publication": ("publication-1", 1), "reporting.drill_context": ("drill-1", 1), "reporting.quality_result": ("quality-1", 1), "reporting.import_assessment": ("assessment-1", 1), } for key, (value, expected) 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(expected, len(records)) def test_planning_and_execution_preserve_governed_boundaries(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, ) by_resource = {action.resource_id: action for action in actions} self.assertEqual("delete", by_resource["view-row-1"].kind) self.assertEqual("manual_review", by_resource["view-row-2"].kind) self.assertEqual("delete", by_resource["drill-row-1"].kind) self.assertEqual("revoke", by_resource["grant-row-1"].kind) self.assertEqual("retain", by_resource["execution-row-1"].kind) 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.assertIn("executed", {result.status for result in first}) self.assertIn("blocked", {result.status for result in first}) self.assertIn("unchanged", {result.status for result in second}) self.assertIsNone(self.session.get(ReportingSavedView, "view-row-1")) self.assertIsNone(self.session.get(ReportingDrillContext, "drill-row-1")) self.assertFalse( self.session.get(ReportingDefinitionGrant, "grant-row-1").active ) self.assertEqual( [{"secret": SECRET}], self.session.get(ReportingExecution, "execution-row-1").result_rows, ) def test_exact_derived_detail_is_minimized_idempotently(self) -> None: subject = DsarSubjectRef( external_references={ "reporting.execution": "execution-1", "reporting.provider_execution": "provider-execution-1", "reporting.provider_export": "export-1", "reporting.publication": "publication-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.assertEqual({"anonymize"}, {action.kind for action in actions}) first = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=actions, request_id="dsar-2", ) second = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=actions, request_id="dsar-2-retry", ) self.assertTrue(all(result.status == "executed" for result in first)) self.assertTrue(all(result.status == "unchanged" for result in second)) execution = self.session.get(ReportingExecution, "execution-row-1") provider_execution = self.session.get( ReportingProviderExecution, "provider-row-1", ) publication = self.session.get( ReportingPublication, "publication-row-1", ) provider_export = self.session.get(ReportingProviderExport, "export-row-1") self.assertEqual([], execution.result_rows) self.assertEqual({}, execution.parameters) self.assertEqual({}, provider_execution.result_payload) self.assertIsNotNone(provider_execution.retention_redacted_at) self.assertEqual({}, provider_export.audience_scope) self.assertEqual( "Redacted by data-subject request.", provider_export.purpose, ) self.assertIsNone(publication.target_ref) self.assertEqual({}, publication.evidence) self.assertIsNone(publication.error) 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-REPORTING-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( [REPORTING_DSAR_CAPABILITY], row.coverage["provider_capabilities"], ) self.assertEqual(12, row.search_result["record_count"]) inactive = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-REPORTING-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( [REPORTING_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(REPORTING_DSAR_CAPABILITY, manifest.capability_factories) self.assertIn(REPORTING_DSAR_CAPABILITY, manifest.capability_documentation) self.assertIn( REPORTING_DSAR_CAPABILITY, {item.name for item in manifest.provides_interfaces}, ) self.assertTrue( any( topic.id == "reporting.data-subject-requests" and {"admin", "user"}.issubset(topic.documentation_types) for topic in manifest.documentation ) ) if __name__ == "__main__": unittest.main()