feat: add durable auth principal cache revisions

2026-07-29 19:23:52 +02:00
parent 4bde0495f7
commit 48fb953b93
3 changed files with 228 additions and 0 deletions
--- a/tests/test_principal_cache.py
+++ b/tests/test_principal_cache.py
@@ -0,0 +1,95 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_core.core.change_sequence import (
+    ChangeSequenceEntry,
+    ChangeSequenceRetentionFloor,
+    prune_sequence_entries,
+)
+from govoplan_core.core.principal_cache import (
+    auth_principal_revision,
+    invalidate_auth_principals,
+)
+from govoplan_core.db.base import Base
+
+
+class AuthPrincipalRevisionTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(
+            bind=self.engine,
+            tables=[
+                ChangeSequenceEntry.__table__,
+                ChangeSequenceRetentionFloor.__table__,
+            ],
+        )
+        self.SessionLocal = sessionmaker(bind=self.engine)
+
+    def tearDown(self) -> None:
+        self.engine.dispose()
+
+    def test_revisions_are_tenant_scoped_and_include_system_changes(self) -> None:
+        with self.SessionLocal() as session:
+            initial = auth_principal_revision(session, tenant_id="tenant-1")
+            self.assertEqual(initial.system, 0)
+            self.assertEqual(initial.tenant, 0)
+
+            system_entry = invalidate_auth_principals(
+                session,
+                tenant_id=None,
+                source_module="access",
+                resource_type="system_role",
+                resource_id="role-1",
+            )
+            tenant_entry = invalidate_auth_principals(
+                session,
+                tenant_id="tenant-1",
+                source_module="idm",
+                resource_type="function_assignment",
+                resource_id="assignment-1",
+            )
+            invalidate_auth_principals(
+                session,
+                tenant_id="tenant-2",
+                source_module="access",
+                resource_type="role",
+                resource_id="role-2",
+            )
+            session.commit()
+
+            revision = auth_principal_revision(session, tenant_id="tenant-1")
+            self.assertEqual(revision.system, system_entry.id)
+            self.assertEqual(revision.tenant, tenant_entry.id)
+
+    def test_retention_floor_keeps_revision_monotonic_after_pruning(self) -> None:
+        with self.SessionLocal() as session:
+            entry = invalidate_auth_principals(
+                session,
+                tenant_id="tenant-1",
+                source_module="access",
+                resource_type="role",
+                resource_id="role-1",
+            )
+            session.flush()
+            expected_revision = entry.id
+            prune_sequence_entries(
+                session,
+                before_or_equal_id=expected_revision,
+                tenant_id="tenant-1",
+                module_id="core",
+                collections=("core.auth-principal-revisions",),
+            )
+            session.commit()
+
+            self.assertEqual(
+                auth_principal_revision(session, tenant_id="tenant-1").tenant,
+                expected_revision,
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()