from __future__ import annotations import hashlib 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_forms_runtime.backend.db.models import ( FormAcknowledgement, FormAssistedConfirmation, FormHandoffEffect, FormInstanceEvent, FormInstanceIdentity, FormInstanceRevision, FormIntakeProfile, FormIntakeSession, FormStatusAccessGrant, FormStatusAccessPolicy, FormStatusAccessToken, ) from govoplan_forms_runtime.backend.dsar_provider import ( FORMS_RUNTIME_DSAR_CAPABILITY, FormsRuntimeDsarProvider, ) from govoplan_forms_runtime.backend.manifest import manifest NOW = datetime(2026, 8, 21, 14, 0, tzinfo=UTC) class _Registry: def __init__( self, provider: FormsRuntimeDsarProvider, *, active: bool = True, ) -> None: self.provider = provider self.active = active def capability_names(self): return (FORMS_RUNTIME_DSAR_CAPABILITY,) def capability_owner(self, name): self._assert_capability(name) return "forms_runtime" 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": ("forms_runtime",) 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": "forms_runtime"})(),) @staticmethod def _assert_capability(name: str) -> None: if name != FORMS_RUNTIME_DSAR_CAPABILITY: raise KeyError(name) def _snapshot(instance_id: str, *, name: str) -> dict[str, object]: return { "instance_id": instance_id, "definition_ref": { "kind": "form", "owner_module": "forms", "object_id": "permit-form", "version": "3", }, "service_ref": { "kind": "service", "owner_module": "services", "object_id": "permit-service", "version": "2", }, "values": { "name": name, "email": "subject@example.org", "password_token": "credential-value-do-not-export", "nested": {"address": "Example Street 1"}, }, "validation_results": [{"message": "private-validation-do-not-export"}], "attachment_refs": [{"evidence_id": "private-attachment-do-not-export"}], "signature_refs": [{"evidence_id": "private-signature-do-not-export"}], "handoff_refs": [{"object_id": "case-1"}], "receipt_id": "receipt-1", "change_reason": "private-change-reason-do-not-export", "metadata": {"secret": "private-metadata-do-not-export"}, } class FormsRuntimeDsarProviderTests(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 = FormsRuntimeDsarProvider() self.assertIsInstance(self.provider, DsarProvider) self._seed_subject_instance() self._seed_assisted_instance() self._seed_unrelated_instances() self._seed_operator_configuration() self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def _seed_subject_instance(self) -> None: identity = FormInstanceIdentity( id="identity-row-1", tenant_id="tenant-1", instance_id="instance-1", definition_id="permit-form", definition_revision="3", created_by="applicant-1", ) self.session.add(identity) self.session.flush() first = FormInstanceRevision( id="revision-row-1", tenant_id="tenant-1", instance_id="instance-1", identity_id=identity.id, revision=1, status="draft", recorded_at=NOW, superseded_at=NOW + timedelta(minutes=1), snapshot=_snapshot("instance-1", name="Ada Example draft"), changed_by="applicant-1", ) second = FormInstanceRevision( id="revision-row-2", tenant_id="tenant-1", instance_id="instance-1", identity_id=identity.id, revision=2, previous_revision_id=first.id, status="submitted", recorded_at=NOW + timedelta(minutes=1), snapshot=_snapshot("instance-1", name="Ada Example"), changed_by="applicant-1", ) events = ( FormInstanceEvent( id="event-row-1", tenant_id="tenant-1", instance_id="instance-1", instance_revision=1, event_id="event-public-1", event_type="forms_runtime.instance.started", status="draft", occurred_at=NOW, actor_id="applicant-1", idempotency_key="private-event-key-do-not-export", request_sha256="a" * 64, payload={"secret": "private-event-payload-do-not-export"}, ), FormInstanceEvent( id="event-row-2", tenant_id="tenant-1", instance_id="instance-1", instance_revision=2, event_id="event-public-2", event_type="forms_runtime.instance.submitted", status="submitted", occurred_at=NOW + timedelta(minutes=1), actor_id="applicant-1", idempotency_key="private-submit-key-do-not-export", request_sha256="b" * 64, payload={"secret": "private-submit-payload-do-not-export"}, ), ) handoff = FormHandoffEffect( id="handoff-row-1", tenant_id="tenant-1", instance_id="instance-1", effect_id="handoff-public-1", instance_revision=2, idempotency_key="private-handoff-key-do-not-export", provider_key="private-provider-key-do-not-export", request_sha256="c" * 64, binding_kind="case", binding_reference="permit-case", provider_capability="cases.service_launcher", state="succeeded", attempt_count=1, requested_at=NOW + timedelta(minutes=1), resolved_at=NOW + timedelta(minutes=2), target_ref={ "kind": "case", "owner_module": "cases", "object_id": "case-1", "version": "1", }, evidence=[{"secret": "private-handoff-evidence-do-not-export"}], last_error="private-error-do-not-export", details={"secret": "private-handoff-details-do-not-export"}, ) intake = FormIntakeSession( id="intake-row-1", tenant_id="tenant-1", session_id="intake-public-1", profile_id="profile-row-1", token_sha256="d" * 64, mode="authenticated", status="submitted", instance_id="instance-1", actor_id="applicant-1", idempotency_key="private-intake-key-do-not-export", request_sha256="e" * 64, expires_at=NOW + timedelta(hours=1), started_at=NOW, submitted_at=NOW + timedelta(minutes=1), created_by="applicant-1", details={"secret": "private-intake-details-do-not-export"}, ) grant_id = "status-grant-public-1" grant = FormStatusAccessGrant( id="grant-row-1", tenant_id="tenant-1", grant_id=grant_id, policy_id="policy-row-1", instance_id="instance-1", tracking_id="private-tracking-id-do-not-export", mode="email_link", applicant_actor_id="applicant-1", recipient_email_sha256=hashlib.sha256( f"{grant_id}\0subject@example.org".encode() ).hexdigest(), token_ttl_seconds=900, request_limit_per_hour=3, issued_at=NOW + timedelta(minutes=1), details={"secret": "private-grant-details-do-not-export"}, ) token = FormStatusAccessToken( id="token-row-1", tenant_id="tenant-1", token_id="private-token-id-do-not-export", grant_id=grant.id, token_sha256="f" * 64, issued_at=NOW + timedelta(minutes=1), expires_at=NOW + timedelta(minutes=16), notification_id="private-notification-id-do-not-export", details={"secret": "private-token-details-do-not-export"}, ) confirmation = FormAssistedConfirmation( id="confirmation-row-1", tenant_id="tenant-1", confirmation_id="confirmation-public-1", intake_session_id=intake.id, instance_id="instance-1", instance_revision=2, outcome="confirmed", method="read_back", confirmed_by_ref="applicant-1", operator_actor_id="operator-other", confirmed_at=NOW + timedelta(minutes=1), payload_sha256="1" * 64, idempotency_key="private-confirmation-key-do-not-export", request_sha256="2" * 64, correction_note="private-correction-note-do-not-export", details={"secret": "private-confirmation-details-do-not-export"}, ) acknowledgement = FormAcknowledgement( id="ack-row-1", tenant_id="tenant-1", acknowledgement_id="ack-public-1", instance_id="instance-1", instance_revision=2, statement_id="truthful-submission", statement_version="1", actor_id="applicant-1", accepted_at=NOW + timedelta(minutes=1), payload_sha256="3" * 64, idempotency_key="private-ack-key-do-not-export", request_sha256="4" * 64, details={"secret": "private-ack-details-do-not-export"}, ) self.session.add_all( ( first, second, *events, handoff, intake, grant, token, confirmation, acknowledgement, ) ) def _seed_assisted_instance(self) -> None: identity = FormInstanceIdentity( id="identity-assisted", tenant_id="tenant-1", instance_id="instance-assisted", definition_id="permit-form", definition_revision="3", created_by="operator-1", ) self.session.add(identity) self.session.flush() self.session.add_all( ( FormInstanceRevision( id="revision-assisted", tenant_id="tenant-1", instance_id="instance-assisted", identity_id=identity.id, revision=1, status="draft", recorded_at=NOW, snapshot=_snapshot( "instance-assisted", name="Assisted Applicant Private Name", ), changed_by="operator-1", ), FormInstanceEvent( id="event-assisted", tenant_id="tenant-1", instance_id="instance-assisted", instance_revision=1, event_id="event-assisted-public", event_type="forms_runtime.instance.started", status="draft", occurred_at=NOW, actor_id="operator-1", idempotency_key="assisted-private-key", request_sha256="5" * 64, payload={"secret": "assisted-private-payload"}, ), FormIntakeSession( id="intake-assisted", tenant_id="tenant-1", session_id="intake-assisted-public", profile_id="profile-row-1", token_sha256="6" * 64, mode="assisted", status="started", instance_id="instance-assisted", actor_id="applicant-assisted", idempotency_key="assisted-intake-private-key", request_sha256="7" * 64, expires_at=NOW + timedelta(hours=1), started_at=NOW, created_by="operator-1", details={"secret": "assisted-details-private"}, ), ) ) def _seed_unrelated_instances(self) -> None: for tenant_id, suffix, actor in ( ("tenant-1", "unrelated", "unrelated-actor"), ("tenant-2", "other-tenant", "applicant-1"), ): identity = FormInstanceIdentity( id=f"identity-{suffix}", tenant_id=tenant_id, instance_id=f"instance-{suffix}", definition_id="permit-form", definition_revision="3", created_by=actor, ) self.session.add(identity) self.session.flush() self.session.add( FormInstanceRevision( id=f"revision-{suffix}", tenant_id=tenant_id, instance_id=identity.instance_id, identity_id=identity.id, revision=1, status="submitted", recorded_at=NOW, snapshot=_snapshot( identity.instance_id, name=f"private-{suffix}-name-do-not-export", ), changed_by=actor, ) ) def _seed_operator_configuration(self) -> None: self.session.add_all( ( FormIntakeProfile( id="profile-row-1", tenant_id="tenant-1", profile_id="profile-public-1", public_id="public-profile-id", definition_id="permit-form", definition_revision="3", mode="assisted", enabled=True, revision=1, custodian_ref="unit-1", draft_ttl_seconds=3600, invitation_ttl_seconds=3600, rate_limit_per_minute=10, created_by="operator-1", updated_by="operator-1", details={"secret": "private-profile-details-do-not-export"}, ), FormStatusAccessPolicy( id="policy-row-1", tenant_id="tenant-1", policy_id="policy-public-1", definition_id="permit-form", definition_revision="3", mode="email_link", enabled=True, revision=1, email_field_key="email", token_ttl_seconds=900, request_limit_per_hour=3, created_by="operator-1", updated_by="operator-1", details={"secret": "private-policy-details-do-not-export"}, ), ) ) def test_actor_search_exports_subject_instance_and_minimizes_internals( self, ) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(account_id="applicant-1"), ) types = {item.resource_type for item in records} self.assertIn("forms_runtime_instance_revision", types) self.assertIn("forms_runtime_handoff_effect", types) self.assertIn("forms_runtime_status_access_grant", types) self.assertIn("forms_runtime_status_token_lifecycle", types) self.assertIn("forms_runtime_assisted_confirmation", types) self.assertIn("forms_runtime_acknowledgement", types) exported = json.dumps([item.to_dict() for item in records], sort_keys=True) self.assertIn("Ada Example", exported) self.assertIn("Example Street 1", exported) self.assertNotIn("credential-value-do-not-export", exported) self.assertNotIn("private-attachment-do-not-export", exported) self.assertNotIn("private-signature-do-not-export", exported) self.assertNotIn("private-validation-do-not-export", exported) self.assertNotIn("private-event-payload-do-not-export", exported) self.assertNotIn("private-token-id-do-not-export", exported) self.assertNotIn("private-tracking-id-do-not-export", exported) self.assertNotIn("private-notification-id-do-not-export", exported) self.assertNotIn("private-provider-key-do-not-export", exported) self.assertNotIn("private-error-do-not-export", exported) self.assertNotIn("private-correction-note-do-not-export", exported) self.assertNotIn("private-unrelated-name-do-not-export", exported) self.assertNotIn("private-other-tenant-name-do-not-export", exported) def test_email_selector_matches_grant_specific_digest(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(email=" Subject@Example.org "), ) self.assertTrue( any( item.resource_type == "forms_runtime_status_access_grant" and item.data["email_selector_matched"] is True for item in records ) ) self.assertTrue( any( item.resource_type == "forms_runtime_instance_revision" for item in records ) ) def test_assisted_operator_is_attributed_but_not_treated_as_applicant(self) -> None: operator_records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(account_id="operator-1"), ) exported = json.dumps( [item.to_dict() for item in operator_records], sort_keys=True ) self.assertIn("forms_runtime_operator_attribution", exported) self.assertIn("created_form_instance", exported) self.assertIn("intake_profile_configuration", exported) self.assertNotIn("Assisted Applicant Private Name", exported) self.assertNotIn("assisted-private-payload", exported) self.assertNotIn("private-profile-details-do-not-export", exported) self.assertNotIn("private-policy-details-do-not-export", exported) applicant_records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(account_id="applicant-assisted"), ) self.assertIn( "Assisted Applicant Private Name", json.dumps([item.to_dict() for item in applicant_records]), ) def test_direct_and_canonical_conflicts_fail_closed(self) -> None: direct = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={"forms_runtime.instance": "instance-1"} ), ) self.assertTrue( any( item.resource_type == "forms_runtime_instance_revision" for item in direct ) ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="applicant-1", external_references={"forms_runtime.instance": "instance-unrelated"}, ), ) self.assertEqual((), conflict) alias_conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="applicant-1", external_references={"forms_runtime.account": "other-account"}, ), ) self.assertEqual((), alias_conflict) def test_planning_retains_evidence_and_reviews_active_status_access(self) -> None: subject = DsarSubjectRef(account_id="applicant-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.assertGreater(sum(item.kind == "retain" for item in actions), 5) self.assertEqual(1, sum(item.kind == "manual_review" for item in actions)) self.assertTrue(all(not item.executable for item in actions)) results = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=actions, request_id="dsar-forms-1", ) self.assertTrue(all(item.status == "blocked" for item in results)) def test_foreign_records_and_actions_are_rejected(self) -> None: subject = DsarSubjectRef(account_id="applicant-1") foreign_record = DsarRecordRef( provider_id="foreign", module_id="foreign", resource_type="foreign", resource_id="foreign-1", category="foreign", title="Foreign record", ) with self.assertRaisesRegex(ValueError, "foreign provider record"): self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=subject, records=(foreign_record,), ) foreign_action = DsarErasureActionRef( action_id="foreign:delete:1", provider_id="foreign", module_id="foreign", kind="delete", resource_type="foreign", resource_id="foreign-1", title="Delete foreign", rationale="No", executable=True, ) with self.assertRaisesRegex(ValueError, "foreign provider action"): self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=(foreign_action,), request_id="dsar-forms-1", ) def test_workflow_discovers_only_the_active_tenant_capability(self) -> None: active = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-FORMS-1", request_kind="access", subject=DsarSubjectRef(account_id="applicant-1"), purpose="Subject access request", legal_basis=None, due_at=None, requested_by_account_id="privacy-operator", ) search_data_subject_request( self.session, registry=_Registry(self.provider), row=active, expected_revision=1, ) self.assertEqual("searched", active.status) self.assertEqual( [FORMS_RUNTIME_DSAR_CAPABILITY], active.coverage["provider_capabilities"], ) self.assertEqual(["forms_runtime"], active.coverage["covered_modules"]) inactive = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-FORMS-2", request_kind="access", subject=DsarSubjectRef(account_id="applicant-1"), purpose="Inactive module coverage", legal_basis=None, due_at=None, requested_by_account_id="privacy-operator", ) 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( [FORMS_RUNTIME_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(FORMS_RUNTIME_DSAR_CAPABILITY, manifest.capability_factories) self.assertIn( FORMS_RUNTIME_DSAR_CAPABILITY, manifest.capability_documentation, ) self.assertIn( FORMS_RUNTIME_DSAR_CAPABILITY, {item.name for item in manifest.provides_interfaces}, ) self.assertTrue( any( topic.id == "forms_runtime.data-subject-requests" and {"admin", "user"}.issubset(topic.documentation_types) for topic in manifest.documentation ) ) if __name__ == "__main__": unittest.main()