feat(policy): resolve datasource visibility overlays

2026-08-21 20:31:52 +02:00
parent 016965917e
commit 72779d0277
7 changed files with 287 additions and 6 deletions
@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.datasources import DatasourceVisibilityPolicyRequest
+from govoplan_core.db.base import Base
+from govoplan_policy.backend.datasource_visibility import (
+    DatasourceVisibilityPolicyProvider,
+)
+from govoplan_policy.backend.db.models import PolicyOverride
+from govoplan_policy.backend.policy_overrides import set_policy_override
+
+
+class DatasourceVisibilityPolicyTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(self.engine, tables=[PolicyOverride.__table__])
+        self.Session = sessionmaker(bind=self.engine)
+        self.session = self.Session()
+        self.provider = DatasourceVisibilityPolicyProvider()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(self.engine, tables=[PolicyOverride.__table__])
+        self.engine.dispose()
+
+    def _request(self, policy_ref: str | None) -> DatasourceVisibilityPolicyRequest:
+        return DatasourceVisibilityPolicyRequest(
+            tenant_id="tenant-1",
+            datasource_ref="datasource:cases",
+            principal=PrincipalRef(
+                account_id="account-1",
+                membership_id="member-1",
+                tenant_id="tenant-1",
+                group_ids=frozenset({"group-1"}),
+            ),
+            action="read",
+            policy_ref=policy_ref,
+        )
+
+    def test_unresolved_explicit_reference_fails_closed(self) -> None:
+        decision = self.provider.decide_datasource_visibility(
+            self.session,
+            request=self._request("missing"),
+        )
+
+        self.assertFalse(decision.allowed)
+        self.assertEqual("reference_unresolved", decision.provenance["status"])
+        self.assertTrue(decision.decision_ref.startswith("datasource-visibility:"))
+
+    def test_global_and_referenced_hierarchy_are_returned_as_overlays(self) -> None:
+        set_policy_override(
+            self.session,
+            policy_family="datasource_visibility",
+            target_key="*",
+            tenant_id="tenant-1",
+            scope_type="system",
+            scope_id=None,
+            policy={"source_acl": {"auth_methods": ["session"]}},
+            actor_id="admin",
+        )
+        set_policy_override(
+            self.session,
+            policy_family="datasource_visibility",
+            target_key="Case-Workers",
+            tenant_id="tenant-1",
+            scope_type="tenant",
+            scope_id="tenant-1",
+            policy={"source_acl": {"group_ids": ["group-1"]}},
+            actor_id="admin",
+        )
+        set_policy_override(
+            self.session,
+            policy_family="datasource_visibility",
+            target_key="case-workers",
+            tenant_id="tenant-1",
+            scope_type="user",
+            scope_id="member-1",
+            policy={
+                "fields": {
+                    "secret": {
+                        "action": "omit",
+                        "allow": {"role_ids": ["privileged"]},
+                    }
+                }
+            },
+            actor_id="admin",
+        )
+
+        decision = self.provider.decide_datasource_visibility(
+            self.session,
+            request=self._request("CASE-WORKERS"),
+        )
+
+        self.assertTrue(decision.allowed)
+        self.assertEqual(3, len(decision.policies))
+        self.assertEqual("resolved", decision.provenance["status"])
+        self.assertEqual(
+            ["system", "tenant", "user"],
+            [source["scope_type"] for source in decision.provenance["sources"]],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()