feat: provide access DSAR integration

2026-08-07 14:53:24 +02:00
parent e04671034f
commit 0b9e3751c2
3 changed files with 819 additions and 0 deletions
@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+import unittest
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from govoplan_access.backend.db.base import AccessBase
+from govoplan_access.backend.db.models import Account, ApiKey, AuthSession, User
+from govoplan_access.backend.dsar_provider import AccessDsarProvider
+from govoplan_core.core.dsar import DsarSubjectRef
+
+
+class AccessDsarProviderTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine(
+            "sqlite+pysqlite://",
+            connect_args={"check_same_thread": False},
+            poolclass=StaticPool,
+        )
+        AccessBase.metadata.create_all(bind=self.engine)
+        self.session = sessionmaker(bind=self.engine, expire_on_commit=False)()
+        self.account = Account(
+            id="account-1",
+            email="ada@example.test",
+            normalized_email="ada@example.test",
+            display_name="Ada Example",
+            password_hash="secret-hash",
+        )
+        self.user = User(
+            id="membership-1",
+            tenant_id="tenant-1",
+            account_id=self.account.id,
+            email="ada@example.test",
+            display_name="Ada Example",
+            password_hash="tenant-secret-hash",
+            settings={"locale": "de"},
+            mail_profile_policy={"profile": "one"},
+        )
+        self.key = ApiKey(
+            id="key-1",
+            tenant_id="tenant-1",
+            user_id=self.user.id,
+            name="Automation",
+            prefix="gpn_example",
+            key_hash="do-not-export",
+            scopes=["files:read"],
+        )
+        self.auth_session = AuthSession(
+            id="session-1",
+            tenant_id="tenant-1",
+            user_id=self.user.id,
+            account_id=self.account.id,
+            token_hash="do-not-export",
+            csrf_token_hash="do-not-export",
+            expires_at=datetime.now(timezone.utc) + timedelta(hours=1),
+            user_agent="Browser fingerprint",
+            ip_address="192.0.2.10",
+        )
+        self.session.add_all([self.account, self.user, self.key, self.auth_session])
+        self.session.commit()
+        self.provider = AccessDsarProvider()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        AccessBase.metadata.drop_all(bind=self.engine)
+        self.engine.dispose()
+
+    def test_search_omits_secret_and_client_fingerprint_material(self) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(email="ADA@example.test"),
+        )
+        serialized = repr([record.to_dict() for record in records])
+        self.assertIn("membership-1", serialized)
+        self.assertNotIn("do-not-export", serialized)
+        self.assertNotIn("Browser fingerprint", serialized)
+        self.assertNotIn("192.0.2.10", serialized)
+
+    def test_multiple_subject_selectors_must_identify_the_same_membership(self) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                membership_id=self.user.id,
+                email="different@example.test",
+            ),
+        )
+
+        self.assertEqual((), records)
+
+    def test_plan_and_execution_anonymize_membership_and_revoke_credentials(self) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(account_id=self.account.id),
+        )
+        actions = self.provider.plan_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(account_id=self.account.id),
+            records=records,
+        )
+        executable = tuple(action for action in actions if action.executable)
+        self.assertEqual(3, len(executable))
+        self.assertTrue(any(action.kind == "manual_review" for action in actions))
+
+        results = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(account_id=self.account.id),
+            actions=executable,
+            request_id="dsar-1",
+        )
+        self.assertEqual({"executed"}, {result.status for result in results})
+        self.assertTrue(self.user.email.endswith("@invalid.govoplan"))
+        self.assertFalse(self.user.is_active)
+        self.assertEqual({}, self.user.settings)
+        self.assertIsNotNone(self.key.revoked_at)
+        self.assertIsNotNone(self.auth_session.revoked_at)
+        self.assertIsNone(self.auth_session.user_agent)
+        self.assertIsNone(self.auth_session.ip_address)
+        self.assertEqual("ada@example.test", self.account.email)
+
+        repeated = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(account_id=self.account.id),
+            actions=executable,
+            request_id="dsar-1",
+        )
+        self.assertEqual({"unchanged"}, {result.status for result in repeated})
+
+
+if __name__ == "__main__":
+    unittest.main()