from __future__ import annotations import json import unittest from sqlalchemy import Column, String, Table, create_engine from sqlalchemy.orm import Session from govoplan_audit.backend.db.models import ( AuditEvidenceBundle, AuditLog, AuditOutboxDelivery, AuditOutboxEvent, ) from govoplan_audit.backend.dsar_provider import ( AUDIT_DSAR_CAPABILITY, AuditDsarProvider, ) from govoplan_audit.backend.manifest import manifest from govoplan_core.core.dsar import ( DsarErasureActionRef, DsarProvider, DsarRecordRef, DsarSubjectRef, ) from govoplan_core.db.base import Base from govoplan_core.privacy.dsar_workflow import ( DataSubjectRequest, create_data_subject_request, search_data_subject_request, ) class _Registry: def __init__(self, provider: AuditDsarProvider, *, active: bool = True) -> None: self.provider = provider self.active = active def capability_names(self): return (AUDIT_DSAR_CAPABILITY,) def capability_owner(self, name): self._assert_capability(name) return "audit" 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": ("audit",) 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": "audit"})(),) @staticmethod def _assert_capability(name: str) -> None: if name != AUDIT_DSAR_CAPABILITY: raise KeyError(name) class AuditDsarProviderTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") if "access_users" not in Base.metadata.tables: Table( "access_users", Base.metadata, Column("id", String(36), primary_key=True), ) if "access_api_keys" not in Base.metadata.tables: Table( "access_api_keys", Base.metadata, Column("id", String(36), primary_key=True), ) Base.metadata.create_all( self.engine, tables=[ Base.metadata.tables["access_users"], Base.metadata.tables["access_api_keys"], AuditLog.__table__, AuditOutboxEvent.__table__, AuditOutboxDelivery.__table__, AuditEvidenceBundle.__table__, DataSubjectRequest.__table__, ], ) self.session = Session(self.engine) self.provider = AuditDsarProvider() self.assertIsInstance(self.provider, DsarProvider) self._seed() self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() @staticmethod def _event_payload( *, event_id: str, tenant_id: str, actor_id: str, ) -> dict[str, object]: return { "type": "case.decision.recorded", "module_id": "cases", "payload": { "authorization_token": "event-secret-do-not-export", "private_case_data": "third-party-data-do-not-export", }, "occurred_at": "2026-08-21T10:00:00+00:00", "event_id": event_id, "correlation_id": f"trace-{event_id}", "causation_id": None, "actor": {"type": "user", "id": actor_id, "label": "Private name"}, "tenant": {"id": tenant_id, "label": "Private tenant label"}, "subject": {"type": "case", "id": "case-1", "label": "Private subject"}, "resource": { "type": "decision", "id": "decision-1", "label": "Private resource", }, "classification": "confidential", "institutional_context": { "authorization": "institutional-secret-do-not-export" }, } def _seed(self) -> None: event = AuditOutboxEvent( id="outbox-1", event_id="event-1", event_type="case.decision.recorded", module_id="cases", correlation_id="trace-event-1", causation_id=None, classification="confidential", payload=self._event_payload( event_id="event-1", tenant_id="tenant-1", actor_id="user-1", ), status="dispatched", attempts=1, ) self.session.add_all( ( AuditLog( id="audit-log-1", scope="tenant", tenant_id="tenant-1", user_id="user-1", action="case.decision.recorded", object_type="case", object_id="case-1", details={ "correlation_id": "trace-audit-1", "policy_decision_ref": "policy:decision-1", "source_ref": "cases:case-1:v4", "message_body": "audit-secret-do-not-export", "authorization_token": "audit-token-do-not-export", }, ), AuditLog( id="audit-log-other", scope="tenant", tenant_id="tenant-1", user_id="user-other", action="other.action", object_type="case", object_id="case-other", details={"private": "other actor"}, ), AuditLog( id="audit-log-other-tenant", scope="tenant", tenant_id="tenant-2", user_id="user-1", action="other.tenant.action", object_type="case", object_id="case-other-tenant", details={"private": "other tenant"}, ), event, AuditOutboxDelivery( id="delivery-1", outbox_event_id="outbox-1", consumer_id="reporting.audit-consumer", delivery_key="event-1:reporting.audit-consumer", policy_decision_ref="policy:delivery-1", status="delivered", attempts=1, replay_count=1, last_replayed_by="user-1", last_replay_reason="private-replay-reason-do-not-export", last_error="private-delivery-error-do-not-export", ), AuditEvidenceBundle( id="bundle-1", scope="tenant", tenant_id="tenant-1", requested_by="user-1", status="ready", request_payload={"secret": "request-secret-do-not-export"}, bundle_payload={"secret": "bundle-secret-do-not-export"}, bundle_sha256="a" * 64, record_count=1, reference_count=2, ), ) ) @staticmethod def _subject() -> DsarSubjectRef: return DsarSubjectRef( account_id="account-1", membership_id="membership-1", external_references={"audit.user": "user-1"}, ) def test_search_is_tenant_actor_scoped_and_minimized(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self._subject(), ) self.assertEqual( [ "audit_actor_record", "audit_event_actor_record", "audit_replay_attribution", "audit_evidence_bundle_attribution", ], [record.resource_type for record in records], ) exported = json.dumps([record.to_dict() for record in records]) self.assertIn("trace-audit-1", exported) self.assertIn("policy:decision-1", exported) self.assertIn("decision-1", exported) self.assertIn("a" * 64, exported) for excluded in ( "audit-secret-do-not-export", "audit-token-do-not-export", "event-secret-do-not-export", "third-party-data-do-not-export", "institutional-secret-do-not-export", "private-replay-reason-do-not-export", "private-delivery-error-do-not-export", "request-secret-do-not-export", "bundle-secret-do-not-export", "audit-log-other", "audit-log-other-tenant", ): self.assertNotIn(excluded, exported) def test_exact_references_narrow_and_alias_conflicts_fail_closed(self) -> None: log = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={ "audit.user": "user-1", "audit.log": "audit-log-1", } ), ) event = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={ "audit.user": "user-1", "audit.event": "event-1", } ), ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"audit.account": "account-other"}, ), ) reference_only = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(external_references={"audit.log": "audit-log-1"}), ) self.assertEqual(["audit-log-1"], [item.resource_id for item in log]) self.assertEqual(["outbox-1"], [item.resource_id for item in event]) self.assertEqual((), conflict) self.assertEqual((), reference_only) def test_erasure_is_retain_only_and_execution_is_blocked(self) -> None: subject = self._subject() 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)) self.assertIsNotNone(self.session.get(AuditLog, "audit-log-1")) self.assertIsNotNone(self.session.get(AuditOutboxEvent, "outbox-1")) def test_foreign_records_and_actions_are_rejected(self) -> None: subject = self._subject() 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="audit_actor_record", resource_id="audit-log-1", category="evidence", title="Foreign evidence", ), ), ) 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:audit:audit-log-1", provider_id="cases", module_id="cases", kind="retain", resource_type="audit_actor_record", resource_id="audit-log-1", title="Retain evidence", rationale="Evidence", executable=False, ), ), request_id="dsar-1", ) def test_core_workflow_and_manifest_register_provider(self) -> None: row = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-AUDIT-1", request_kind="access", subject=self._subject(), 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([AUDIT_DSAR_CAPABILITY], row.coverage["provider_capabilities"]) self.assertEqual(4, row.search_result["record_count"]) inactive = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-AUDIT-2", request_kind="access", subject=self._subject(), 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( [AUDIT_DSAR_CAPABILITY], inactive.coverage["inactive_provider_capabilities"], ) self.assertIn(AUDIT_DSAR_CAPABILITY, manifest.capability_factories) self.assertIn(AUDIT_DSAR_CAPABILITY, manifest.capability_documentation) self.assertIn( AUDIT_DSAR_CAPABILITY, {item.name for item in manifest.provides_interfaces}, ) self.assertTrue( any( topic.id == "audit.data-subject-requests" and {"admin", "user"}.issubset(topic.documentation_types) for topic in manifest.documentation ) ) if __name__ == "__main__": unittest.main()