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_tasks.backend.db.models import TaskAssignment, TaskItem from govoplan_tasks.backend.dsar_provider import ( TASKS_DSAR_CAPABILITY, TasksDsarProvider, ) from govoplan_tasks.backend.manifest import manifest class _Registry: def __init__(self, provider: TasksDsarProvider, *, active: bool = True) -> None: self.provider = provider self.active = active def capability_names(self): return (TASKS_DSAR_CAPABILITY,) def capability_owner(self, name): self._assert_capability(name) return "tasks" 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": ("tasks",) 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": "tasks"})(),) @staticmethod def _assert_capability(name: str) -> None: if name != TASKS_DSAR_CAPABILITY: raise KeyError(name) class TasksDsarProviderTests(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 = TasksDsarProvider() self.assertIsInstance(self.provider, DsarProvider) self._seed() self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def _task( self, task_id: str, *, tenant_id: str = "tenant-1", created_by: str = "account-other", updated_by: str = "account-other", completed_by: str | None = None, ) -> TaskItem: return TaskItem( id=task_id, tenant_id=tenant_id, title=f"Private title for {task_id}", summary=f"Private summary for {task_id}", status="completed" if completed_by else "open", priority="high", required_action="Review the source decision", action_url="/cases/case-1", source_module="cases", source_resource_type="case", source_resource_id="case-1", source_revision="4", sources=[ { "module_id": "cases", "resource_type": "case", "resource_id": "case-1", "revision": "4", "url": "/cases/case-1", "label": "Case reference", } ], provenance={"secret": "provenance-secret-do-not-export"}, metadata_={"secret": "metadata-secret-do-not-export"}, revision=2, idempotency_key=f"idempotency-{task_id}-do-not-export", request_sha256="a" * 64, created_by=created_by, updated_by=updated_by, completed_by=completed_by, ) def _seed(self) -> None: assigned = self._task("task-assigned") assigned.assignments.extend( ( TaskAssignment( id="assignment-account", tenant_id="tenant-1", assignment_kind="account", assignment_id="account-1", assignment_label="Resident account", ), TaskAssignment( id="assignment-group", tenant_id="tenant-1", assignment_kind="group", assignment_id="group-private", assignment_label="Private group label do not export", ), ) ) actor_only = self._task( "task-actor-only", created_by="account-1", updated_by="account-1", completed_by="account-1", ) actor_only.assignments.append( TaskAssignment( id="assignment-other", tenant_id="tenant-1", assignment_kind="account", assignment_id="account-other", assignment_label="Other account", ) ) other = self._task("task-other") other.assignments.append( TaskAssignment( id="assignment-other-task", tenant_id="tenant-1", assignment_kind="account", assignment_id="account-other", ) ) other_tenant = self._task( "task-other-tenant", tenant_id="tenant-2", created_by="account-1", updated_by="account-1", ) other_tenant.assignments.append( TaskAssignment( id="assignment-other-tenant", tenant_id="tenant-2", assignment_kind="account", assignment_id="account-1", ) ) self.session.add_all((assigned, actor_only, other, other_tenant)) @staticmethod def _subject() -> DsarSubjectRef: return DsarSubjectRef(account_id="account-1") def test_search_exports_assigned_task_and_minimized_actor_attribution(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self._subject(), ) by_id = {record.resource_id: record for record in records} self.assertEqual( {"task-assigned", "task-actor-only"}, set(by_id), ) self.assertEqual("assigned_task", by_id["task-assigned"].resource_type) self.assertEqual( "task_actor_attribution", by_id["task-actor-only"].resource_type, ) exported = json.dumps([record.to_dict() for record in records]) self.assertIn("Private summary for task-assigned", exported) self.assertIn('"module_id": "cases"', exported) self.assertIn('"resource_id": "case-1"', exported) self.assertIn("completed_task", exported) self.assertNotIn("Private summary for task-actor-only", exported) self.assertNotIn("Private group label do not export", exported) self.assertNotIn("group-private", exported) self.assertNotIn("provenance-secret-do-not-export", exported) self.assertNotIn("metadata-secret-do-not-export", exported) self.assertNotIn("idempotency-task-assigned-do-not-export", exported) self.assertNotIn("task-other-tenant", exported) def test_exact_task_reference_narrows_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={"tasks.task": "task-assigned"}, ), ) conflict = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( account_id="account-1", external_references={"tasks.account": "account-other"}, ), ) reference_only = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef(external_references={"tasks.task": "task-assigned"}), ) self.assertEqual(["task-assigned"], [item.resource_id for item in narrowed]) self.assertEqual((), conflict) self.assertEqual((), reference_only) def test_erasure_requires_review_or_retention_and_changes_nothing(self) -> None: subject = self._subject() 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( {"manual_review", "retain"}, {action.kind for action in actions}, ) self.assertTrue(all(not action.executable for action in actions)) results = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=actions, request_id="dsar-1", ) self.assertTrue(all(result.status == "blocked" for result in results)) self.assertIsNotNone(self.session.get(TaskItem, "task-assigned")) self.assertIsNotNone(self.session.get(TaskAssignment, "assignment-account")) def test_foreign_records_and_actions_are_rejected(self) -> None: subject = self._subject() with self.assertRaisesRegex(ValueError, "foreign provider record"): self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=subject, records=( DsarRecordRef( provider_id="workflow_engine", module_id="workflow_engine", resource_type="assigned_task", resource_id="task-assigned", category="work", title="Foreign task", ), ), ) with self.assertRaisesRegex(ValueError, "foreign provider action"): self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=subject, actions=( DsarErasureActionRef( action_id="workflow_engine:retain:task:task-assigned", provider_id="workflow_engine", module_id="workflow_engine", kind="retain", resource_type="task_actor_attribution", resource_id="task-assigned", title="Retain task", rationale="Foreign action", executable=False, ), ), 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-TASKS-1", request_kind="access", subject=self._subject(), 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([TASKS_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-TASKS-2", request_kind="access", subject=self._subject(), 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( [TASKS_DSAR_CAPABILITY], inactive.coverage["inactive_provider_capabilities"], ) self.assertIn(TASKS_DSAR_CAPABILITY, manifest.capability_factories) self.assertIn(TASKS_DSAR_CAPABILITY, manifest.capability_documentation) self.assertIn( TASKS_DSAR_CAPABILITY, {item.name for item in manifest.provides_interfaces}, ) self.assertTrue( any( topic.id == "tasks.data-subject-requests" and {"admin", "user"}.issubset(topic.documentation_types) for topic in manifest.documentation ) ) if __name__ == "__main__": unittest.main()