110 lines
3.2 KiB
Python
110 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import pathlib
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import (
|
|
CAPABILITY_POLICY_ACCESS_EXPLANATION_SUBJECTS,
|
|
AccessExplanationSubjectDecision,
|
|
PrincipalRef,
|
|
)
|
|
from govoplan_access.backend.api.v1.routes import (
|
|
_access_explanation_subject_decision,
|
|
)
|
|
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _principal() -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="user-1",
|
|
tenant_id="tenant-1",
|
|
),
|
|
account=SimpleNamespace(id="account-1"),
|
|
user=SimpleNamespace(id="user-1"),
|
|
)
|
|
|
|
|
|
class _SubjectPolicy:
|
|
def decide_subject_selection(
|
|
self,
|
|
session: object,
|
|
principal: PrincipalRef,
|
|
*,
|
|
tenant_id: str,
|
|
) -> AccessExplanationSubjectDecision:
|
|
del session, principal, tenant_id
|
|
return AccessExplanationSubjectDecision(
|
|
allow_other_users=True,
|
|
reason="Permitted by test policy.",
|
|
source="test.policy",
|
|
)
|
|
|
|
|
|
class _Registry:
|
|
def __init__(self, provider: object | None = None) -> None:
|
|
self.provider = provider
|
|
|
|
def has_capability(self, name: str) -> bool:
|
|
return (
|
|
name == CAPABILITY_POLICY_ACCESS_EXPLANATION_SUBJECTS
|
|
and self.provider is not None
|
|
)
|
|
|
|
def require_capability(self, name: str) -> object:
|
|
if not self.has_capability(name):
|
|
raise KeyError(name)
|
|
return self.provider
|
|
|
|
|
|
class ResourceAccessExplanationSubjectTests(unittest.TestCase):
|
|
def test_missing_policy_defaults_to_current_user(self) -> None:
|
|
with patch(
|
|
"govoplan_access.backend.api.v1.routes.get_registry",
|
|
return_value=None,
|
|
):
|
|
decision = _access_explanation_subject_decision(
|
|
object(), # type: ignore[arg-type]
|
|
_principal(),
|
|
tenant_id="tenant-1",
|
|
)
|
|
|
|
self.assertFalse(decision.allow_other_users)
|
|
self.assertEqual("access.safe_default", decision.source)
|
|
|
|
def test_policy_capability_controls_cross_user_selection(self) -> None:
|
|
with patch(
|
|
"govoplan_access.backend.api.v1.routes.get_registry",
|
|
return_value=_Registry(_SubjectPolicy()),
|
|
):
|
|
decision = _access_explanation_subject_decision(
|
|
object(), # type: ignore[arg-type]
|
|
_principal(),
|
|
tenant_id="tenant-1",
|
|
)
|
|
|
|
self.assertTrue(decision.allow_other_users)
|
|
self.assertEqual("test.policy", decision.source)
|
|
|
|
def test_route_contract_is_tenant_bounded_and_audited(self) -> None:
|
|
source = (
|
|
ROOT / "src/govoplan_access/backend/api/v1/routes.py"
|
|
).read_text(encoding="utf-8")
|
|
|
|
self.assertIn('User.tenant_id == tenant.id', source)
|
|
self.assertIn('User.id == principal.membership_id', source)
|
|
self.assertIn(
|
|
'action="access.resource_explanation.selected_user_viewed"',
|
|
source,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|