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_views.backend.db.models import ( ViewAssignment, ViewDefinition, ViewPreference, ViewRevision, ) from govoplan_views.backend.dsar_provider import ( VIEWS_DSAR_CAPABILITY, ViewsDsarProvider, _presentation_projection, ) from govoplan_views.backend.manifest import manifest class _Registry: def __init__(self, provider: ViewsDsarProvider, *, active: bool = True) -> None: self.provider = provider self.active = active def capability_names(self): return (VIEWS_DSAR_CAPABILITY,) def capability_owner(self, name): self._assert_capability(name) return "views" 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": ("views",) 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": "views"})(),) @staticmethod def _assert_capability(name: str) -> None: if name != VIEWS_DSAR_CAPABILITY: raise KeyError(name) class ViewsDsarProviderTests(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 = ViewsDsarProvider() self.assertIsInstance(self.provider, DsarProvider) self._seed() self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def _definition( self, definition_id: str, *, tenant_id: str, scope_type: str, scope_id: str, actor_id: str, name: str, ) -> ViewDefinition: definition = ViewDefinition( id=definition_id, tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id, scope_key=f"{scope_type}:{tenant_id}:{scope_id}", definition_key=definition_id, name=name, description=f"Description for {name}", status="published", current_revision=1, published_revision_id=f"revision-{definition_id}", created_by=actor_id, updated_by=actor_id, ) definition.revisions.append( ViewRevision( id=f"revision-{definition_id}", tenant_id=tenant_id, revision=1, surface_contract_version="1.0.0", visible_surface_ids=["files.route.files"], presentation={ "navigation": { "order": ["separator:work", "files.nav.files"], "hidden": ["mail.nav.mail"], "separators": [{"id": "separator:work", "label": "Personal layout", "private": "separator-private-do-not-export"}], "private": "navigation-private-do-not-export", }, "navigation_mode": "flat", "quick_access_focused_tool_ids": ["files.recent"], "private_payload_do_not_export": f"private-{definition_id}", }, content_hash=("a" * 64), created_by=actor_id, ) ) return definition def _seed(self) -> None: personal = self._definition( "definition-personal", tenant_id="tenant-1", scope_type="user", scope_id="account-1", actor_id="account-1", name="My personal work", ) tenant = self._definition( "definition-tenant", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", actor_id="account-1", name="private-tenant-definition-do-not-export", ) other_account = self._definition( "definition-other-account", tenant_id="tenant-1", scope_type="user", scope_id="account-other", actor_id="account-other", name="Other account", ) other_tenant = self._definition( "definition-other-tenant", tenant_id="tenant-2", scope_type="user", scope_id="account-1", actor_id="account-1", name="Other tenant", ) self.session.add_all((personal, tenant, other_account, other_tenant)) self.session.flush() self.session.add_all( ( ViewAssignment( id="assignment-personal", tenant_id="tenant-1", scope_type="user", scope_id="account-1", target_key="user:tenant-1:account-1", definition_id=personal.id, revision_id=None, mode="available", priority=1, is_active=True, metadata_={"private": "personal-metadata-do-not-export"}, created_by="account-1", updated_by="account-1", ), ViewAssignment( id="assignment-tenant", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", target_key="tenant:tenant-1", definition_id=tenant.id, revision_id=None, mode="default", priority=2, is_active=True, metadata_={"private": "tenant-metadata-do-not-export"}, created_by="account-1", updated_by="account-other", ), ViewPreference( id="preference-personal", tenant_id="tenant-1", account_id="account-1", selection_kind="selected", view_id=personal.id, ), ViewPreference( id="preference-other-account", tenant_id="tenant-1", account_id="account-other", selection_kind="auto", view_id=None, ), ViewPreference( id="preference-other-tenant", tenant_id="tenant-2", account_id="account-1", selection_kind="auto", view_id=None, ), ) ) def test_search_exports_personal_records_and_minimized_attribution(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(account_id="account-1"), ) self.assertEqual( [ "personal_view_assignment", "personal_view_preference", "personal_view_definition", "view_assignment_attribution", "view_definition_attribution", "view_revision_attribution", ], [record.resource_type for record in records], ) exported = json.dumps([record.to_dict() for record in records]) self.assertIn("My personal work", exported) self.assertIn("files.recent", exported) self.assertIn("Personal layout", exported) self.assertIn("files.nav.files", exported) self.assertNotIn("navigation-private-do-not-export", exported) self.assertNotIn("separator-private-do-not-export", exported) self.assertNotIn("personal-metadata-do-not-export", exported) self.assertNotIn("tenant-metadata-do-not-export", exported) self.assertNotIn("private-tenant-definition-do-not-export", exported) self.assertNotIn("definition-other-account", exported) self.assertNotIn("definition-other-tenant", exported) def test_navigation_projection_is_bounded_and_preserves_inheritance(self) -> None: self.assertEqual({"navigation": None}, _presentation_projection({"navigation": None})) self.assertNotIn("separators", _presentation_projection({"navigation": {"order": []}})["navigation"]) self.assertEqual([], _presentation_projection({"navigation": {"separators": []}})["navigation"]["separators"]) for navigation in ( {"order": ["files.nav.files"] * 257}, {"separators": [{"id": "separator:work"}] * 257}, {"separators": ["invalid"]}, ): with self.subTest(navigation=navigation), self.assertRaises(ValueError): _presentation_projection({"navigation": navigation}) def test_resource_references_narrow_and_conflicts_fail_closed(self) -> None: definition = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"views.definition": "definition-personal"}, ), ) assignment = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"views.assignment": "assignment-personal"}, ), ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"views.account": "account-other"}, ), ) no_account = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( external_references={"views.preference": "preference-personal"} ), ) self.assertEqual( { "personal_view_assignment", "personal_view_preference", "personal_view_definition", }, { item.resource_type for item in definition if item.resource_type.startswith("personal_") }, ) self.assertEqual( ["personal_view_assignment"], [item.resource_type for item in assignment], ) self.assertEqual((), conflict) self.assertEqual((), no_account) def test_erasure_deletes_personal_records_and_retains_institutional_ones( 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", "delete", "delete", "retain", "retain", "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", "executed", "executed", "blocked", "blocked", "blocked"], [result.status for result in first], ) self.assertEqual( ["unchanged", "unchanged", "unchanged", "blocked", "blocked", "blocked"], [result.status for result in second], ) self.assertIsNone(self.session.get(ViewAssignment, "assignment-personal")) self.assertIsNone(self.session.get(ViewPreference, "preference-personal")) self.assertIsNone(self.session.get(ViewDefinition, "definition-personal")) self.assertIsNone( self.session.get(ViewRevision, "revision-definition-personal") ) self.assertIsNotNone(self.session.get(ViewDefinition, "definition-tenant")) self.assertIsNotNone(self.session.get(ViewAssignment, "assignment-tenant")) def test_foreign_dependency_and_changed_definition_block_deletion(self) -> None: subject = DsarSubjectRef( account_id="account-1", external_references={"views.definition": "definition-personal"}, ) definition_record = next( item for item in self.provider.search_subject( self.session, tenant_id="tenant-1", subject=subject, ) if item.resource_type == "personal_view_definition" ) definition = self.session.get(ViewDefinition, "definition-personal") definition.current_revision += 1 stale_action = DsarErasureActionRef( action_id="views:delete:personal_view_definition:definition-personal:r1", provider_id="views", module_id="views", kind="delete", resource_type="personal_view_definition", resource_id="definition-personal", title="Delete personal View", rationale="Personal preference", executable=True, metadata={ "account_id": "account-1", "current_revision": 1, "updated_at": definition_record.observed_at.isoformat(), }, ) changed = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=(stale_action,), request_id="dsar-1", ) self.assertEqual("blocked", changed[0].status) definition.current_revision = 1 self.session.add( ViewAssignment( id="assignment-foreign-dependent", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", target_key="tenant:tenant-1:foreign-dependent", definition_id=definition.id, revision_id=None, mode="available", priority=0, is_active=True, metadata_={}, created_by="account-other", updated_by="account-other", ) ) self.session.flush() manual = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=subject, records=(definition_record,), ) self.assertEqual("manual_review", manual[0].kind) self.assertFalse(manual[0].executable) def test_foreign_records_and_actions_are_rejected(self) -> None: subject = DsarSubjectRef(account_id="account-1") 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="personal_view_preference", resource_id="preference-personal", category="preference", title="Foreign preference", ), ), ) 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:view:preference-personal", provider_id="dashboard", module_id="dashboard", kind="delete", resource_type="personal_view_preference", resource_id="preference-personal", title="Delete preference", rationale="Foreign action", executable=True, ), ), request_id="dsar-1", ) def test_core_workflow_and_manifest_register_provider(self) -> None: row = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-VIEWS-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([VIEWS_DSAR_CAPABILITY], row.coverage["provider_capabilities"]) self.assertEqual(6, row.search_result["record_count"]) inactive = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-VIEWS-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( [VIEWS_DSAR_CAPABILITY], inactive.coverage["inactive_provider_capabilities"], ) self.assertIn(VIEWS_DSAR_CAPABILITY, manifest.capability_factories) self.assertIn(VIEWS_DSAR_CAPABILITY, manifest.capability_documentation) self.assertIn( VIEWS_DSAR_CAPABILITY, {item.name for item in manifest.provides_interfaces}, ) self.assertTrue( any( topic.id == "views.data-subject-requests" and {"admin", "user"}.issubset(topic.documentation_types) for topic in manifest.documentation ) ) if __name__ == "__main__": unittest.main()