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_organizations.backend.db.models import ( OrganizationModelInstantiation, OrganizationModelTemplate, OrganizationModelTemplateVersion, OrganizationModelUpgrade, ) from govoplan_organizations.backend.dsar_provider import ( ORGANIZATIONS_DSAR_CAPABILITY, OrganizationsDsarProvider, ) from govoplan_organizations.backend.manifest import manifest class _Registry: def __init__( self, provider: OrganizationsDsarProvider, *, organizations_active: bool = True, ) -> None: self.provider = provider self.organizations_active = organizations_active def capability_names(self): return (ORGANIZATIONS_DSAR_CAPABILITY,) def capability_owner(self, name): self._assert_capability(name) return "organizations" def tenant_entitlement_resolver(self): organizations_active = self.organizations_active class _Resolver: @staticmethod def resolve(session, tenant_id): del session, tenant_id return type( "State", (), { "effective_modules": ( ("organizations",) if organizations_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": "organizations"})(),) @staticmethod def _assert_capability(name: str) -> None: if name != ORGANIZATIONS_DSAR_CAPABILITY: raise KeyError(name) class OrganizationsDsarProviderTests(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.template = OrganizationModelTemplate( id="template-1", slug="municipality", name="Municipality", description="Global institutional template", created_by_account_id="account-1", settings={"secret": "global-template-settings-do-not-export"}, ) self.version_one = OrganizationModelTemplateVersion( id="version-1", template_id=self.template.id, version="1.0.0", status="published", definition={"secret": "definition-one-do-not-export"}, definition_sha256="a" * 64, published_at=now, published_by_account_id="account-1", ) self.version_two = OrganizationModelTemplateVersion( id="version-2", template_id=self.template.id, version="2.0.0", status="published", definition={"secret": "definition-two-do-not-export"}, definition_sha256="b" * 64, published_at=now, published_by_account_id="account-1", ) self.version_three = OrganizationModelTemplateVersion( id="version-3", template_id=self.template.id, version="3.0.0", status="published", definition={"secret": "definition-three-do-not-export"}, definition_sha256="c" * 64, ) self.instantiation = OrganizationModelInstantiation( id="instantiation-1", tenant_id="tenant-1", template_id=self.template.id, template_version_id=self.version_one.id, source_definition_sha256=self.version_one.definition_sha256, status="superseded", instantiated_by_account_id="account-1", object_counts={"units": 5}, provenance={"secret": "instantiation-provenance-do-not-export"}, ) unrelated_instantiation = OrganizationModelInstantiation( id="instantiation-unrelated", tenant_id="tenant-1", template_id=self.template.id, template_version_id=self.version_three.id, source_definition_sha256=self.version_three.definition_sha256, status="applied", instantiated_by_account_id="account-other", object_counts={"units": 9}, provenance={"secret": "unrelated-provenance-do-not-export"}, ) tenant_two_instantiation = OrganizationModelInstantiation( id="instantiation-tenant-2", tenant_id="tenant-2", template_id=self.template.id, template_version_id=self.version_one.id, source_definition_sha256=self.version_one.definition_sha256, status="applied", instantiated_by_account_id="account-1", object_counts={"units": 99}, provenance={"secret": "other-tenant-provenance-do-not-export"}, ) self.upgrade = OrganizationModelUpgrade( id="upgrade-1", tenant_id="tenant-1", template_id=self.template.id, source_instantiation_id=self.instantiation.id, source_template_version_id=self.version_one.id, target_template_version_id=self.version_two.id, status="applied", revision=2, base_definition_sha256="d" * 64, local_definition_sha256="e" * 64, target_definition_sha256="f" * 64, preview={"secret": "upgrade-preview-do-not-export"}, decisions={"secret": "upgrade-decisions-do-not-export"}, idempotency_key="idempotency-key-do-not-export", request_digest="1" * 64, requested_by_account_id="account-1", applied_by_account_id="account-other", applied_at=now, provenance={"secret": "upgrade-provenance-do-not-export"}, ) unrelated_upgrade = OrganizationModelUpgrade( id="upgrade-unrelated", tenant_id="tenant-1", template_id=self.template.id, source_instantiation_id=unrelated_instantiation.id, source_template_version_id=self.version_three.id, target_template_version_id=self.version_two.id, status="cancelled", base_definition_sha256="2" * 64, local_definition_sha256="3" * 64, target_definition_sha256="4" * 64, idempotency_key="unrelated-key", request_digest="5" * 64, requested_by_account_id="account-other", cancelled_by_account_id="account-other", ) tenant_two_upgrade = OrganizationModelUpgrade( id="upgrade-tenant-2", tenant_id="tenant-2", template_id=self.template.id, source_instantiation_id=tenant_two_instantiation.id, source_template_version_id=self.version_one.id, target_template_version_id=self.version_two.id, status="previewed", base_definition_sha256="6" * 64, local_definition_sha256="7" * 64, target_definition_sha256="8" * 64, idempotency_key="tenant-two-key", request_digest="9" * 64, requested_by_account_id="account-1", ) self.session.add_all( [ self.template, self.version_one, self.version_two, self.version_three, self.instantiation, unrelated_instantiation, tenant_two_instantiation, self.upgrade, unrelated_upgrade, tenant_two_upgrade, ] ) self.session.commit() self.provider = OrganizationsDsarProvider() self.subject = DsarSubjectRef(account_id="account-1") def tearDown(self) -> None: self.session.close() self.engine.dispose() def test_manifest_publishes_protocol_conforming_provider(self) -> None: provided = {item.name for item in manifest.provides_interfaces} self.assertIn(ORGANIZATIONS_DSAR_CAPABILITY, provided) provider = manifest.capability_factories[ORGANIZATIONS_DSAR_CAPABILITY](None) self.assertIsInstance(provider, DsarProvider) def test_search_is_tenant_scoped_narrow_and_minimized(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self.subject, ) self.assertEqual( { "organizations_model_instantiation", "organizations_model_upgrade", }, {record.resource_type for record in records}, ) serialized = repr([record.to_dict() for record in records]) self.assertIn("instantiation-1", serialized) self.assertIn("upgrade-1", serialized) excluded = ( "global-template-settings-do-not-export", "definition-one-do-not-export", "instantiation-provenance-do-not-export", "upgrade-preview-do-not-export", "upgrade-decisions-do-not-export", "idempotency-key-do-not-export", "upgrade-provenance-do-not-export", "instantiation-unrelated", "upgrade-unrelated", "instantiation-tenant-2", "upgrade-tenant-2", "other-tenant-provenance-do-not-export", ) for value in excluded: self.assertNotIn(value, serialized) def test_conflicting_account_selectors_fail_closed(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"organizations.account": "account-other"}, ), ) self.assertEqual((), records) def test_plan_retains_governance_evidence_and_execution_is_non_mutating( 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({"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-organizations-1", ) self.assertEqual({"blocked"}, {result.status for result in results}) self.assertIsNotNone( self.session.get(OrganizationModelUpgrade, self.upgrade.id) ) def test_execution_rejects_foreign_and_forged_executable_actions(self) -> None: actions = ( DsarErasureActionRef( action_id="idm:delete:upgrade:upgrade-1", provider_id="idm", module_id="idm", kind="delete", resource_type="organizations_model_upgrade", resource_id=self.upgrade.id, title="Foreign action", rationale="Must be rejected", executable=True, ), DsarErasureActionRef( action_id="organizations:delete:upgrade:upgrade-1", provider_id="organizations", module_id="organizations", kind="delete", resource_type="organizations_model_upgrade", resource_id=self.upgrade.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-organizations-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-ORGANIZATIONS-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(["organizations"], request.coverage["covered_modules"]) disabled = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-ORGANIZATIONS-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, organizations_active=False), row=disabled, expected_revision=1, ) self.assertEqual(0, disabled.search_result["record_count"]) self.assertEqual( [ORGANIZATIONS_DSAR_CAPABILITY], disabled.coverage["inactive_provider_capabilities"], ) if __name__ == "__main__": unittest.main()