from __future__ import annotations import unittest from datetime import datetime, timezone from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from govoplan_core.core.dsar import ( DsarErasureActionRef, 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.backend.db.models import CanonicalIdentity from govoplan_idm.backend.db.models import ( IdmFunctionAssignmentChange, IdmFunctionAssignmentChangeEvent, IdmIdentityRelationship, IdmOrganizationFunctionAssignment, IdmTypedGroup, ) from govoplan_idm.backend.dsar_provider import IDM_DSAR_CAPABILITY, IdmDsarProvider from govoplan_idm.backend.manifest import manifest from govoplan_organizations.backend.db.models import ( OrganizationFunction, OrganizationUnit, ) class _Registry: def __init__(self, provider: IdmDsarProvider, *, idm_active: bool = True) -> None: self.provider = provider self.idm_active = idm_active def capability_names(self): return (IDM_DSAR_CAPABILITY,) def capability_owner(self, name): self._assert_capability(name) return "idm" def tenant_entitlement_resolver(self): idm_active = self.idm_active class _Resolver: @staticmethod def resolve(session, tenant_id): del session, tenant_id return type( "State", (), {"effective_modules": ("idm",) if idm_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": "idm"})(),) @staticmethod def _assert_capability(name: str) -> None: if name != IDM_DSAR_CAPABILITY: raise KeyError(name) class IdmDsarProviderTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite:///:memory:", future=True) Base.metadata.create_all(bind=self.engine) self.session = sessionmaker(bind=self.engine, future=True)() now = datetime.now(timezone.utc) self.identity = CanonicalIdentity( id="identity-1", display_name="Subject", settings={"secret": "identity-settings-do-not-export"}, ) other_identity = CanonicalIdentity( id="identity-other", display_name="Unrelated Person", settings={"secret": "other-identity-settings-do-not-export"}, ) unit = OrganizationUnit( id="unit-1", tenant_id="tenant-1", slug="residents", name="Residents Office", ) function = OrganizationFunction( id="function-1", tenant_id="tenant-1", organization_unit_id=unit.id, slug="case-worker", name="Case worker", ) acting_function = OrganizationFunction( id="function-2", tenant_id="tenant-1", organization_unit_id=unit.id, slug="acting-case-worker", name="Acting case worker", ) self.assignment = IdmOrganizationFunctionAssignment( id="assignment-1", tenant_id="tenant-1", identity_id=self.identity.id, account_id="account-1", function_id=function.id, organization_unit_id=unit.id, source="direct", valid_from=now, settings={"secret": "assignment-settings-do-not-export"}, ) acting_assignment = IdmOrganizationFunctionAssignment( id="assignment-acting", tenant_id="tenant-1", identity_id=other_identity.id, account_id="account-other", function_id=acting_function.id, organization_unit_id=unit.id, source="acting_for", delegated_from_assignment_id=self.assignment.id, acting_for_account_id="account-1", valid_from=now, settings={"secret": "acting-settings-do-not-export"}, ) self.unrelated_assignment = IdmOrganizationFunctionAssignment( id="assignment-unrelated", tenant_id="tenant-1", identity_id=other_identity.id, account_id="account-other", function_id=function.id, organization_unit_id=unit.id, source="direct", settings={"secret": "unrelated-assignment-do-not-export"}, ) tenant_two_assignment = IdmOrganizationFunctionAssignment( id="assignment-tenant-2", tenant_id="tenant-2", identity_id=self.identity.id, account_id="account-1", function_id=function.id, organization_unit_id=unit.id, source="directory", settings={"secret": "other-tenant-assignment-do-not-export"}, ) self.group = IdmTypedGroup( id="group-1", tenant_id="tenant-1", key="residents", name="Residents", group_type="business_group", source_provider="ldap", source_resource_id="private-group-ref-do-not-export", properties={"secret": "group-properties-do-not-export"}, provenance={"secret": "group-provenance-do-not-export"}, ) self.relationship = IdmIdentityRelationship( id="relationship-1", tenant_id="tenant-1", relationship_kind="member_of", subject_identity_id=self.identity.id, target_group_id=self.group.id, role="member", valid_from=now, source_provider="ldap", source_resource_id="private-relationship-ref-do-not-export", source_revision="private-source-revision-do-not-export", properties={"secret": "relationship-properties-do-not-export"}, provenance={"secret": "relationship-provenance-do-not-export"}, ) related_relationship = IdmIdentityRelationship( id="relationship-related", tenant_id="tenant-1", relationship_kind="representative_for", subject_identity_id=other_identity.id, related_identity_id=self.identity.id, role="representative", properties={"secret": "related-properties-do-not-export"}, provenance={"secret": "related-provenance-do-not-export"}, ) unrelated_relationship = IdmIdentityRelationship( id="relationship-unrelated", tenant_id="tenant-1", relationship_kind="member_of", subject_identity_id=other_identity.id, target_group_id=self.group.id, role="member", ) tenant_two_relationship = IdmIdentityRelationship( id="relationship-tenant-2", tenant_id="tenant-2", relationship_kind="member_of", subject_identity_id=self.identity.id, target_group_id=self.group.id, role="member", properties={"secret": "other-tenant-relationship-do-not-export"}, ) self.change = IdmFunctionAssignmentChange( id="change-1", tenant_id="tenant-1", kind="request", state="approved", profile="self_request", function_id=function.id, organization_unit_id=unit.id, candidate_identity_id=self.identity.id, candidate_account_id="account-1", initiator_account_id="account-other", initiator_identity_id=other_identity.id, justification="private-justification-do-not-export", evidence=["private-evidence-do-not-export"], requested_valid_from=now, required_steps=["approval-secret-do-not-export"], completed_steps=["approval-secret-do-not-export"], policy_decision={"secret": "policy-decision-do-not-export"}, workflow_definition_id="workflow-secret-do-not-export", workflow_instance_id="workflow-instance-do-not-export", idempotency_key="idempotency-key-do-not-export", outcome_reason="outcome-reason-do-not-export", metadata_={"secret": "change-metadata-do-not-export"}, ) initiated_change = IdmFunctionAssignmentChange( id="change-initiated", tenant_id="tenant-1", kind="grant", state="pending", profile="authority_grant", function_id=acting_function.id, organization_unit_id=unit.id, candidate_identity_id=other_identity.id, candidate_account_id="account-other", initiator_account_id="account-1", initiator_identity_id=self.identity.id, justification="third-party-justification-do-not-export", evidence=["third-party-evidence-do-not-export"], idempotency_key="initiated-change-key-do-not-export", metadata_={"secret": "initiated-metadata-do-not-export"}, ) unrelated_change = IdmFunctionAssignmentChange( id="change-unrelated", tenant_id="tenant-1", kind="grant", state="pending", profile="authority_grant", function_id=function.id, organization_unit_id=unit.id, candidate_identity_id=other_identity.id, candidate_account_id="account-other", initiator_account_id="account-other", initiator_identity_id=other_identity.id, justification="unrelated-change-do-not-export", idempotency_key="unrelated-change-key", ) tenant_two_change = IdmFunctionAssignmentChange( id="change-tenant-2", tenant_id="tenant-2", kind="request", state="pending", profile="self_request", function_id=function.id, organization_unit_id=unit.id, candidate_identity_id=self.identity.id, candidate_account_id="account-1", initiator_account_id="account-1", initiator_identity_id=self.identity.id, justification="other-tenant-change-do-not-export", idempotency_key="tenant-two-key", ) self.event = IdmFunctionAssignmentChangeEvent( id="event-1", tenant_id="tenant-1", change_id=self.change.id, sequence=1, action="approved", from_state="pending", to_state="approved", actor_account_id="account-other", actor_identity_id=other_identity.id, actor_assignment_id=self.unrelated_assignment.id, comment="private-event-comment-do-not-export", evidence=["private-event-evidence-do-not-export"], policy_decision={"secret": "event-policy-do-not-export"}, workflow_step_id="workflow-step-do-not-export", details={"secret": "event-details-do-not-export"}, created_at=now, ) actor_event = IdmFunctionAssignmentChangeEvent( id="event-actor", tenant_id="tenant-1", change_id=unrelated_change.id, sequence=1, action="reviewed", from_state="pending", to_state="pending", actor_account_id="account-1", actor_identity_id=self.identity.id, actor_assignment_id=self.assignment.id, comment="actor-comment-do-not-export", evidence=["actor-evidence-do-not-export"], policy_decision={"secret": "actor-policy-do-not-export"}, workflow_step_id="actor-workflow-step-do-not-export", details={"secret": "actor-details-do-not-export"}, created_at=now, ) tenant_two_event = IdmFunctionAssignmentChangeEvent( id="event-tenant-2", tenant_id="tenant-2", change_id=tenant_two_change.id, sequence=1, action="requested", to_state="pending", actor_account_id="account-1", actor_identity_id=self.identity.id, details={"secret": "other-tenant-event-do-not-export"}, created_at=now, ) self.session.add_all( [ self.identity, other_identity, unit, function, acting_function, self.assignment, acting_assignment, self.unrelated_assignment, tenant_two_assignment, self.group, self.relationship, related_relationship, unrelated_relationship, tenant_two_relationship, self.change, initiated_change, unrelated_change, tenant_two_change, self.event, actor_event, tenant_two_event, ] ) self.session.commit() self.provider = IdmDsarProvider() self.subject = DsarSubjectRef( account_id="account-1", identity_id=self.identity.id, ) def tearDown(self) -> None: self.session.close() self.engine.dispose() def test_manifest_publishes_protocol_conforming_provider(self) -> None: self.assertIn( IDM_DSAR_CAPABILITY, {item.name for item in manifest.provides_interfaces}, ) provider = manifest.capability_factories[IDM_DSAR_CAPABILITY](None) self.assertIsInstance(provider, DsarProvider) def test_search_is_tenant_scoped_third_party_safe_and_minimized(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self.subject, ) self.assertTrue( { "idm_function_assignment", "idm_identity_relationship", "idm_typed_group_context", "idm_function_assignment_change", "idm_function_assignment_change_event", }.issubset({record.resource_type for record in records}) ) serialized = repr([record.to_dict() for record in records]) self.assertIn("assignment-acting", serialized) self.assertIn("relationship-related", serialized) self.assertIn("change-initiated", serialized) self.assertIn("event-actor", serialized) excluded = ( "identity-other", "account-other", "assignment-settings-do-not-export", "private-group-ref-do-not-export", "group-properties-do-not-export", "group-provenance-do-not-export", "private-relationship-ref-do-not-export", "private-source-revision-do-not-export", "relationship-properties-do-not-export", "relationship-provenance-do-not-export", "private-justification-do-not-export", "private-evidence-do-not-export", "approval-secret-do-not-export", "policy-decision-do-not-export", "workflow-secret-do-not-export", "workflow-instance-do-not-export", "idempotency-key-do-not-export", "outcome-reason-do-not-export", "change-metadata-do-not-export", "private-event-comment-do-not-export", "private-event-evidence-do-not-export", "event-policy-do-not-export", "workflow-step-do-not-export", "event-details-do-not-export", "unrelated-change-do-not-export", "other-tenant-assignment-do-not-export", "other-tenant-relationship-do-not-export", "other-tenant-change-do-not-export", "other-tenant-event-do-not-export", ) for value in excluded: self.assertNotIn(value, serialized) def test_conflicting_and_uncorroborated_direct_selectors_fail_closed(self) -> None: conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( identity_id=self.identity.id, external_references={"idm.identity": "identity-other"}, ), ) direct_conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( identity_id=self.identity.id, external_references={ "idm.assignment": self.unrelated_assignment.id, }, ), ) self.assertEqual((), conflict) self.assertEqual((), direct_conflict) def test_plan_retains_evidence_and_routes_facts_to_manual_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}, ) self.assertFalse(any(action.executable for action in actions)) results = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=self.subject, actions=actions, request_id="dsar-idm-1", ) self.assertEqual({"blocked"}, {result.status for result in results}) self.assertIsNotNone( self.session.get(IdmOrganizationFunctionAssignment, self.assignment.id) ) def test_execution_rejects_foreign_and_forged_executable_actions(self) -> None: actions = ( DsarErasureActionRef( action_id="identity:delete:assignment:assignment-1", provider_id="identity", module_id="identity", kind="delete", resource_type="idm_function_assignment", resource_id=self.assignment.id, title="Foreign action", rationale="Must be rejected", executable=True, ), DsarErasureActionRef( action_id="idm:delete:assignment:assignment-1", provider_id="idm", module_id="idm", kind="delete", resource_type="idm_function_assignment", resource_id=self.assignment.id, title="Forged action", rationale="Must be rejected", executable=True, ), ) for action in actions: with self.assertRaises(ValueError): self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=self.subject, actions=(action,), request_id="dsar-idm-2", ) def test_core_workflow_discovers_active_and_inactive_provider(self) -> None: request = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-IDM-1", request_kind="access", subject=self.subject, purpose="Respond to an authorized privacy 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=request, expected_revision=1, ) self.assertEqual(["idm"], request.coverage["covered_modules"]) disabled = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-IDM-DISABLED", request_kind="access", subject=self.subject, purpose="Verify disabled-module coverage.", legal_basis="Article 15 GDPR", due_at=None, requested_by_account_id="privacy-officer", ) search_data_subject_request( self.session, registry=_Registry(self.provider, idm_active=False), row=disabled, expected_revision=1, ) self.assertEqual(0, disabled.search_result["record_count"]) self.assertEqual( [IDM_DSAR_CAPABILITY], disabled.coverage["inactive_provider_capabilities"], ) if __name__ == "__main__": unittest.main()