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_approvals.backend.db.models import ( ApprovalDecisionRecord, ApprovalLifecycleEvent, ApprovalRequestRevision, ApprovalTemplateRevision, ) from govoplan_approvals.backend.dsar_provider import ( APPROVALS_DSAR_CAPABILITY, ApprovalsDsarProvider, ) from govoplan_approvals.backend.manifest import manifest from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef from govoplan_core.db.base import Base from govoplan_core.privacy.dsar_workflow import ( create_data_subject_request, search_data_subject_request, ) NOW = datetime(2026, 8, 21, 13, 0, tzinfo=UTC) class _Registry: def __init__(self, provider: ApprovalsDsarProvider) -> None: self.provider = provider def capability_names(self): return (APPROVALS_DSAR_CAPABILITY,) def capability_owner(self, name): if name != APPROVALS_DSAR_CAPABILITY: raise KeyError(name) return "approvals" def tenant_entitlement_resolver(self): class _Resolver: @staticmethod def resolve(session, tenant_id): del session, tenant_id return type("State", (), {"effective_modules": ("approvals",)})() return _Resolver() def require_tenant_capability(self, name, session, **kwargs): del session, kwargs if name != APPROVALS_DSAR_CAPABILITY: raise KeyError(name) return self.provider def manifests(self): return (type("Manifest", (), {"id": "approvals"})(),) class ApprovalsDsarProviderTests(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 = ApprovalsDsarProvider() 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( ( ApprovalRequestRevision( id="request-revision-1", tenant_id="tenant-1", request_id="request-1", revision=1, state="pending", current_step_key="legal", subject_module="cases", subject_type="case", subject_id="case-1", subject_version="4", subject_digest="subject-digest-do-not-export", recorded_at=NOW, payload={"secret": "request-payload-do-not-export"}, actor_id="account-1", ), ApprovalRequestRevision( id="request-revision-other", tenant_id="tenant-1", request_id="request-other", revision=1, state="pending", current_step_key="legal", subject_module="cases", subject_type="case", subject_id="case-other", subject_digest="other-digest", recorded_at=NOW, payload={"private": "other-request"}, actor_id="account-other", ), ApprovalRequestRevision( id="request-revision-other-tenant", tenant_id="tenant-2", request_id="request-other-tenant", revision=1, state="pending", subject_module="cases", subject_type="case", subject_id="case-other-tenant", subject_digest="other-tenant-digest", recorded_at=NOW, payload={"private": "other-tenant-request"}, actor_id="account-1", ), ) ) self.session.add( ApprovalDecisionRecord( id="decision-1", tenant_id="tenant-1", request_id="request-1", request_revision=1, step_key="legal", outcome="approved", reason="I verified the resident evidence.", actor_id="account-1", effective_actor_id="account-1", delegation_id="delegation-1", authority_provenance={ "secret": "authority-provenance-do-not-export" }, signature_ref={"secret": "signature-object-do-not-export"}, recorded_at=NOW, idempotency_key="decision-idempotency-do-not-export", receipt_sha256="receipt-hash-do-not-export", ) ) self.session.add( ApprovalLifecycleEvent( id="event-1", tenant_id="tenant-1", request_id="request-1", sequence=1, event_type="request.created", recorded_at=NOW, actor_id="account-1", payload={"secret": "event-payload-do-not-export"}, ) ) self.session.add( ApprovalTemplateRevision( id="template-revision-1", tenant_id="tenant-1", template_id="template-1", key="resident-permit", revision=1, state="published", content_sha256="template-hash-do-not-export", recorded_at=NOW, payload={"secret": "template-payload-do-not-export"}, actor_id="account-1", ) ) @staticmethod def _subject() -> DsarSubjectRef: return DsarSubjectRef(account_id="account-1") def test_search_exports_decision_and_minimized_attribution(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self._subject() ) self.assertEqual( { "approval_decision_participation", "approval_request_actor_attribution", "approval_lifecycle_actor_attribution", "approval_template_actor_attribution", }, {record.resource_type for record in records}, ) exported = json.dumps([record.to_dict() for record in records]) self.assertIn("I verified the resident evidence.", exported) self.assertIn("case-1", exported) for excluded in ( "request-payload-do-not-export", "subject-digest-do-not-export", "authority-provenance-do-not-export", "signature-object-do-not-export", "decision-idempotency-do-not-export", "receipt-hash-do-not-export", "event-payload-do-not-export", "template-payload-do-not-export", "other-request", "other-tenant-request", ): self.assertNotIn(excluded, exported) def test_request_narrowing_and_conflicting_actor_fail_closed(self) -> None: narrowed = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"approvals.request": "request-1"}, ), ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"approvals.actor": "account-other"}, ), ) request_only = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={"approvals.request": "request-1"} ), ) self.assertEqual(3, len(narrowed)) self.assertNotIn( "approval_template_actor_attribution", {record.resource_type for record in narrowed}, ) self.assertEqual((), conflict) self.assertEqual((), request_only) def test_erasure_preserves_chain_and_requires_reason_review(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self._subject() ) actions = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=self._subject(), records=records, ) self.assertEqual( {"manual_review", "retain"}, {action.kind for action in actions} ) results = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=self._subject(), actions=actions, request_id="dsar-approvals-1", ) self.assertTrue(all(result.status == "blocked" for result in results)) self.assertEqual(1, self.session.query(ApprovalDecisionRecord).count()) def test_manifest_and_core_workflow_discover_provider(self) -> None: self.assertIn(APPROVALS_DSAR_CAPABILITY, manifest.capability_factories) row = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-APPROVALS-1", request_kind="access", subject=self._subject(), purpose="Approval participation access request", legal_basis=None, due_at=None, requested_by_account_id="operator-1", ) search_data_subject_request( self.session, registry=_Registry(self.provider), row=row, expected_revision=row.resource_revision, ) self.assertEqual("searched", row.status) self.assertEqual(4, row.search_result["record_count"]) if __name__ == "__main__": unittest.main()