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_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, ) from govoplan_identity_trust.backend.db.models import ( AssuranceEvidence, DevicePublicKey, KeyAccessDecisionRecord, TrustKeyEpoch, ) from govoplan_identity_trust.backend.dsar_provider import ( IDENTITY_TRUST_DSAR_CAPABILITY, IdentityTrustDsarProvider, ) from govoplan_identity_trust.backend.manifest import manifest NOW = datetime(2026, 8, 21, 12, 0, tzinfo=UTC) class _Registry: def __init__(self, provider: IdentityTrustDsarProvider) -> None: self.provider = provider def capability_names(self): return (IDENTITY_TRUST_DSAR_CAPABILITY,) def capability_owner(self, name): if name != IDENTITY_TRUST_DSAR_CAPABILITY: raise KeyError(name) return "identity_trust" def tenant_entitlement_resolver(self): class _Resolver: @staticmethod def resolve(session, tenant_id): del session, tenant_id return type( "State", (), {"effective_modules": ("identity_trust",)} )() return _Resolver() def require_tenant_capability(self, name, session, **kwargs): del session, kwargs if name != IDENTITY_TRUST_DSAR_CAPABILITY: raise KeyError(name) return self.provider def manifests(self): return (type("Manifest", (), {"id": "identity_trust"})(),) class IdentityTrustDsarProviderTests(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 = IdentityTrustDsarProvider() 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( ( DevicePublicKey( id="device-key-row-1", tenant_id="tenant-1", identity_id="identity-1", account_id="account-1", device_id="device-1", key_id="key-1", algorithm="EdDSA", public_jwk={ "kty": "OKP", "crv": "Ed25519", "x": "public-coordinate", "d": "private-jwk-do-not-export", "unknown": "arbitrary-jwk-do-not-export", }, purpose="postbox", assurance_level="substantial", attestation_ref="evidence:attestation-1", status="active", epoch=1, registration_digest="registration-digest-do-not-export", idempotency_key="device-idempotency-do-not-export", registered_at=NOW, created_by="account-1", updated_by="account-1", ), DevicePublicKey( id="device-key-other-tenant", tenant_id="tenant-2", identity_id="identity-1", account_id="account-1", device_id="device-other-tenant", key_id="key-other-tenant", algorithm="EdDSA", public_jwk={"kty": "OKP", "x": "other-tenant-public"}, purpose="postbox", assurance_level="substantial", status="active", epoch=1, registration_digest="other-digest", idempotency_key="other-key", registered_at=NOW, ), ) ) self.session.add( AssuranceEvidence( id="assurance-1", tenant_id="tenant-1", account_id="account-1", device_key_id="key-1", evidence_ref="assurance:evidence-1", assurance_level="substantial", provider_id="bund-id", verified_at=NOW, expires_at=NOW, provenance={"secret": "assurance-provenance-do-not-export"}, recorded_by="security-officer", ) ) self.session.add( TrustKeyEpoch( id="epoch-1", tenant_id="tenant-1", subject_kind="account", subject_id="account-1", epoch=2, previous_epoch=1, state="active", history_policy="forward_only", reason="Device rotation", access_decision_ref="access:decision-1", idempotency_key="epoch-idempotency-do-not-export", request_digest="epoch-request-digest-do-not-export", effective_at=NOW, created_by="security-officer", ) ) self.session.add( KeyAccessDecisionRecord( id="access-decision-1", tenant_id="tenant-1", decision_ref="trust-decision-1", request_digest="access-request-digest-do-not-export", account_id="account-1", device_key_id="key-1", subject_kind="postbox", subject_id="postbox-1", key_epoch=2, access_decision_ref="access:authorization-1", purpose="read-message", allowed=True, reason="Current device and epoch", resource_ref="postbox:message-1", provenance={"secret": "access-provenance-do-not-export"}, ) ) @staticmethod def _subject() -> DsarSubjectRef: return DsarSubjectRef(account_id="account-1", identity_id="identity-1") def test_search_exports_bounded_trust_records(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self._subject() ) self.assertEqual( { "device_public_key", "assurance_evidence", "key_epoch", "key_access_decision", }, {record.resource_type for record in records}, ) exported = json.dumps([record.to_dict() for record in records]) self.assertIn("public-coordinate", exported) self.assertIn("assurance:evidence-1", exported) self.assertIn("postbox:message-1", exported) for excluded in ( "private-jwk-do-not-export", "arbitrary-jwk-do-not-export", "registration-digest-do-not-export", "device-idempotency-do-not-export", "assurance-provenance-do-not-export", "epoch-request-digest-do-not-export", "access-request-digest-do-not-export", "access-provenance-do-not-export", "other-tenant-public", ): self.assertNotIn(excluded, exported) def test_key_narrowing_and_conflicts_fail_closed(self) -> None: narrowed = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"identity_trust.key": "key-1"}, ), ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"identity_trust.account": "account-other"}, ), ) reference_only = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={"identity_trust.key": "key-1"} ), ) self.assertEqual( {"device_public_key", "assurance_evidence", "key_access_decision"}, {record.resource_type for record in narrowed}, ) self.assertEqual((), conflict) self.assertEqual((), reference_only) def test_erasure_is_review_or_retain_only(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-trust-1", ) self.assertTrue(all(result.status == "blocked" for result in results)) self.assertEqual( "active", self.session.get(DevicePublicKey, "device-key-row-1").status, ) def test_manifest_and_core_workflow_discover_provider(self) -> None: self.assertIn( IDENTITY_TRUST_DSAR_CAPABILITY, manifest.capability_factories ) row = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-TRUST-1", request_kind="access", subject=self._subject(), purpose="Trust 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()