110 lines
3.7 KiB
Python
110 lines
3.7 KiB
Python
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()
|