from __future__ import annotations import json import unittest 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_quick_access.backend.db.models import QuickAccessProfile from govoplan_quick_access.backend.dsar_provider import ( QUICK_ACCESS_DSAR_CAPABILITY, QuickAccessDsarProvider, ) from govoplan_quick_access.backend.manifest import manifest class _Registry: def __init__( self, provider: QuickAccessDsarProvider, *, active: bool = True, ) -> None: self.provider = provider self.active = active def capability_names(self): return (QUICK_ACCESS_DSAR_CAPABILITY,) def capability_owner(self, name): self._assert_capability(name) return "quick_access" 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": ("quick_access",) 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": "quick_access"})(),) @staticmethod def _assert_capability(name: str) -> None: if name != QUICK_ACCESS_DSAR_CAPABILITY: raise KeyError(name) class QuickAccessDsarProviderTests(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 = QuickAccessDsarProvider() 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( ( QuickAccessProfile( id="profile-personal", scope_type="user", tenant_id="tenant-1", scope_id="account-1", scope_key="user:tenant-1:account-1", category_preferences={"work": {"enabled": False}}, tool_preferences={"tasks.mine": {"order": 5}}, revision=3, created_by="account-1", updated_by="account-1", ), QuickAccessProfile( id="profile-tenant", scope_type="tenant", tenant_id="tenant-1", scope_id="tenant-1", scope_key="tenant:tenant-1", category_preferences={ "private-policy-do-not-export": {"enabled": False} }, tool_preferences={"tasks.mine": {"forced": True}}, revision=4, created_by="account-1", updated_by="account-other", ), QuickAccessProfile( id="profile-other-account", scope_type="user", tenant_id="tenant-1", scope_id="account-other", scope_key="user:tenant-1:account-other", category_preferences={"calendar": {"enabled": False}}, tool_preferences={}, revision=2, created_by="account-other", updated_by="account-other", ), QuickAccessProfile( id="profile-other-tenant", scope_type="user", tenant_id="tenant-2", scope_id="account-1", scope_key="user:tenant-2:account-1", category_preferences={"files": {"enabled": False}}, tool_preferences={}, revision=2, created_by="account-1", updated_by="account-1", ), QuickAccessProfile( id="profile-system", scope_type="system", tenant_id=None, scope_id=None, scope_key="system:*", category_preferences={"messages": {"enabled": False}}, tool_preferences={}, revision=2, created_by="account-1", updated_by="account-1", ), ) ) def test_search_exports_personal_preferences_and_minimized_attribution( self, ) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(account_id="account-1"), ) self.assertEqual( [ "quick_access_personal_profile", "quick_access_tenant_attribution", ], [record.resource_type for record in records], ) exported = json.dumps([record.to_dict() for record in records]) self.assertIn("tasks.mine", exported) self.assertIn("created_tenant_quick_access_policy", exported) self.assertNotIn("private-policy-do-not-export", exported) self.assertNotIn("profile-other-account", exported) self.assertNotIn("profile-other-tenant", exported) self.assertNotIn("profile-system", exported) def test_profile_reference_narrows_and_alias_conflicts_fail_closed(self) -> None: narrowed = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"quick_access.profile": "profile-personal"}, ), ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"quick_access.account": "account-other"}, ), ) no_account = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={"quick_access.profile": "profile-personal"} ), ) self.assertEqual(["profile-personal"], [item.resource_id for item in narrowed]) self.assertEqual((), conflict) self.assertEqual((), no_account) def test_erasure_deletes_only_personal_profile_and_is_idempotent(self) -> None: subject = DsarSubjectRef(account_id="account-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.assertEqual(["delete", "retain"], [action.kind for action in actions]) first = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=actions, request_id="dsar-1", ) second = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=actions, request_id="dsar-1", ) self.assertEqual(["executed", "blocked"], [item.status for item in first]) self.assertEqual(["unchanged", "blocked"], [item.status for item in second]) self.assertIsNone(self.session.get(QuickAccessProfile, "profile-personal")) self.assertIsNotNone(self.session.get(QuickAccessProfile, "profile-tenant")) self.assertIsNotNone(self.session.get(QuickAccessProfile, "profile-system")) def test_changed_and_foreign_resources_are_blocked(self) -> None: subject = DsarSubjectRef(account_id="account-1") record = next( item for item in self.provider.search_subject( self.session, tenant_id="tenant-1", subject=subject, ) if item.resource_type == "quick_access_personal_profile" ) action = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=subject, records=(record,), )[0] self.session.get(QuickAccessProfile, "profile-personal").revision += 1 changed = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=(action,), request_id="dsar-1", ) self.assertEqual("blocked", changed[0].status) with self.assertRaisesRegex(ValueError, "foreign provider record"): self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=subject, records=( DsarRecordRef( provider_id="dashboard", module_id="dashboard", resource_type="quick_access_personal_profile", resource_id="profile-personal", category="preference", title="Foreign profile", ), ), ) with self.assertRaisesRegex(ValueError, "foreign provider action"): self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=( DsarErasureActionRef( action_id="dashboard:delete:profile:profile-personal", provider_id="dashboard", module_id="dashboard", kind="delete", resource_type="quick_access_personal_profile", resource_id="profile-personal", title="Delete profile", rationale="Foreign action", executable=True, ), ), request_id="dsar-1", ) def test_core_workflow_reports_active_and_inactive_provider(self) -> None: row = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-QUICK-ACCESS-1", request_kind="access", subject=DsarSubjectRef(account_id="account-1"), purpose="Respond to a verified 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=row, expected_revision=1, ) self.assertEqual( [QUICK_ACCESS_DSAR_CAPABILITY], row.coverage["provider_capabilities"] ) self.assertEqual(2, row.search_result["record_count"]) inactive = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-QUICK-ACCESS-2", request_kind="access", subject=DsarSubjectRef(account_id="account-1"), purpose="Respond to a verified 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, active=False), row=inactive, expected_revision=1, ) self.assertEqual([], inactive.coverage["provider_capabilities"]) self.assertEqual( [QUICK_ACCESS_DSAR_CAPABILITY], inactive.coverage["inactive_provider_capabilities"], ) def test_manifest_registers_and_documents_the_capability(self) -> None: self.assertIn(QUICK_ACCESS_DSAR_CAPABILITY, manifest.capability_factories) self.assertIn( QUICK_ACCESS_DSAR_CAPABILITY, manifest.capability_documentation, ) self.assertIn( QUICK_ACCESS_DSAR_CAPABILITY, {item.name for item in manifest.provides_interfaces}, ) self.assertTrue( any( topic.id == "quick-access.data-subject-requests" and {"admin", "user"}.issubset(topic.documentation_types) for topic in manifest.documentation ) ) if __name__ == "__main__": unittest.main()