feat: contribute tenant erasure for access data

2026-08-24 15:57:20 +02:00
parent e55434f406
commit 0f8a05f8b9
9 changed files with 434 additions and 6 deletions
@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session
+
+from govoplan_access.backend.db.base import AccessBase
+from govoplan_access.backend.db.models import (
+    Account,
+    ApiKey,
+    AuthSession,
+    Group,
+    Role,
+    User,
+)
+from govoplan_access.backend.tenant_erasure_provider import (
+    AccessTenantErasureProvider,
+)
+
+
+def test_access_erasure_is_tenant_bounded_and_retains_global_account() -> None:
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    AccessBase.metadata.create_all(engine)
+    now = datetime.now(UTC)
+    with Session(engine) as session:
+        account = Account(
+            email="shared@example.test",
+            normalized_email="shared@example.test",
+            is_active=True,
+            auth_provider="local",
+        )
+        session.add(account)
+        session.flush()
+        first = User(
+            tenant_id="tenant-1",
+            account_id=account.id,
+            email="shared@example.test",
+            is_active=True,
+            is_tenant_admin=True,
+            auth_provider="local",
+        )
+        second = User(
+            tenant_id="tenant-2",
+            account_id=account.id,
+            email="shared@example.test",
+            is_active=True,
+            is_tenant_admin=False,
+            auth_provider="local",
+        )
+        session.add_all(
+            [
+                first,
+                second,
+                Group(tenant_id="tenant-1", slug="group", name="Group"),
+                Group(tenant_id="tenant-2", slug="group", name="Group"),
+                Role(tenant_id="tenant-1", slug="role", name="Role"),
+                Role(tenant_id="tenant-2", slug="role", name="Role"),
+            ]
+        )
+        session.flush()
+        session.add_all(
+            [
+                ApiKey(
+                    tenant_id="tenant-1",
+                    user_id=first.id,
+                    name="key",
+                    prefix="prefix",
+                    key_hash="hash",
+                    scopes=[],
+                ),
+                AuthSession(
+                    tenant_id="tenant-1",
+                    user_id=first.id,
+                    account_id=account.id,
+                    token_hash="token-hash",
+                    expires_at=now + timedelta(hours=1),
+                ),
+            ]
+        )
+        session.commit()
+
+        provider = AccessTenantErasureProvider()
+        preview = provider.preview_tenant_erasure(session, "tenant-1")
+
+        assert preview.allowed
+        assert [step.step_id for step in preview.steps] == [
+            "revoke-tenant-credentials",
+            "erase-tenant-access",
+        ]
+        assert "revoke-tenant-credentials" in preview.steps[1].depends_on
+
+        revoked = provider.execute_tenant_erasure_step(
+            session,
+            "tenant-1",
+            "revoke-tenant-credentials",
+            "operation:access:credentials",
+        )
+        erased = provider.execute_tenant_erasure_step(
+            session,
+            "tenant-1",
+            "erase-tenant-access",
+            "operation:access:tenant",
+        )
+        session.commit()
+
+        assert revoked.state == "completed"
+        assert erased.state == "completed"
+        assert provider.preview_tenant_erasure(session, "tenant-1").steps == ()
+        assert session.scalar(select(Account).where(Account.id == account.id)) is not None
+        assert session.scalar(select(User).where(User.tenant_id == "tenant-2")) is not None
+        assert session.scalar(select(Group).where(Group.tenant_id == "tenant-2")) is not None
+        assert session.scalar(select(Role).where(Role.tenant_id == "tenant-2")) is not None
+
+
+def test_access_erasure_replay_is_idempotent() -> None:
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    AccessBase.metadata.create_all(engine)
+    provider = AccessTenantErasureProvider()
+    with Session(engine) as session:
+        first = provider.execute_tenant_erasure_step(
+            session,
+            "tenant-1",
+            "erase-tenant-access",
+            "operation:access:tenant",
+        )
+        second = provider.reconcile_tenant_erasure_step(
+            session,
+            "tenant-1",
+            "erase-tenant-access",
+            "operation:access:tenant",
+        )
+
+        assert first.metrics == {"deleted": 0}
+        assert second.metrics == {"deleted": 0}