feat(admin): declare governed DSAR coverage

2026-08-21 13:17:25 +02:00
parent 550c14b92e
commit c22da17c64
3 changed files with 322 additions and 16 deletions
@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_admin.backend.dsar_provider import (
+    ADMIN_DSAR_CAPABILITY,
+    AdminDsarProvider,
+)
+from govoplan_admin.backend.manifest import manifest
+from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef
+from govoplan_core.db.base import Base
+from govoplan_core.privacy.dsar_workflow import (
+    create_data_subject_request,
+    search_data_subject_request,
+)
+
+
+class _Registry:
+    def __init__(self, provider: AdminDsarProvider) -> None:
+        self.provider = provider
+
+    def capability_names(self):
+        return (ADMIN_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        if name != ADMIN_DSAR_CAPABILITY:
+            raise KeyError(name)
+        return "admin"
+
+    def tenant_entitlement_resolver(self):
+        class _Resolver:
+            @staticmethod
+            def resolve(session, tenant_id):
+                del session, tenant_id
+                return type("State", (), {"effective_modules": ("admin",)})()
+
+        return _Resolver()
+
+    def require_tenant_capability(self, name, session, **kwargs):
+        del session, kwargs
+        if name != ADMIN_DSAR_CAPABILITY:
+            raise KeyError(name)
+        return self.provider
+
+    def manifests(self):
+        return (type("Manifest", (), {"id": "admin"})(),)
+
+
+class AdminDsarProviderTests(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 = AdminDsarProvider()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_provider_explicitly_reports_no_subject_records(self) -> None:
+        self.assertIsInstance(self.provider, DsarProvider)
+        subject = DsarSubjectRef(account_id="account-1")
+        self.assertEqual(
+            (),
+            self.provider.search_subject(
+                self.session, tenant_id="tenant-1", subject=subject
+            ),
+        )
+        self.assertEqual(
+            (),
+            self.provider.plan_erasure(
+                self.session,
+                tenant_id="tenant-1",
+                subject=subject,
+                records=(),
+            ),
+        )
+
+    def test_manifest_and_workflow_record_reviewed_coverage(self) -> None:
+        self.assertIn(ADMIN_DSAR_CAPABILITY, manifest.capability_factories)
+        self.assertIn(
+            "admin.data-subject-request-coverage",
+            {topic.id for topic in manifest.documentation},
+        )
+        row = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-ADMIN-1",
+            request_kind="access",
+            subject=DsarSubjectRef(account_id="account-1"),
+            purpose="Administration coverage review",
+            legal_basis=None,
+            due_at=None,
+            requested_by_account_id="operator-1",
+        )
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider),
+            row=row,
+            expected_revision=row.resource_revision,
+        )
+        self.assertEqual("searched", row.status)
+        self.assertEqual(0, row.search_result["record_count"])
+        self.assertEqual(["admin"], row.coverage["covered_modules"])
+        self.assertEqual([], row.coverage["modules_without_provider"])
+
+
+if __name__ == "__main__":
+    unittest.main()