139 lines
4.8 KiB
Python
139 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.people import (
|
|
CAPABILITY_ACCESS_PEOPLE_SEARCH,
|
|
PeopleSearchGroup,
|
|
PersonSearchCandidate,
|
|
)
|
|
from govoplan_scheduling.backend.manifest import WRITE_SCOPE
|
|
from govoplan_scheduling.backend.router import api_search_scheduling_people
|
|
from govoplan_scheduling.backend.runtime import configure_runtime
|
|
|
|
|
|
class _Registry:
|
|
def __init__(self, capabilities: dict[str, object] | None = None) -> None:
|
|
self.capabilities = capabilities or {}
|
|
|
|
def has_capability(self, name: str) -> bool:
|
|
return name in self.capabilities
|
|
|
|
def capability(self, name: str) -> object:
|
|
return self.capabilities[name]
|
|
|
|
|
|
class _PeopleProvider:
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[object, object, str, int]] = []
|
|
|
|
def search_people(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
query: str,
|
|
limit: int = 25,
|
|
) -> tuple[PeopleSearchGroup, ...]:
|
|
self.calls.append((session, principal, query, limit))
|
|
return (
|
|
PeopleSearchGroup(
|
|
key="accounts",
|
|
label="Accounts",
|
|
candidates=(
|
|
PersonSearchCandidate(
|
|
selection_key="account:account-2",
|
|
kind="account",
|
|
reference_id="account-2",
|
|
display_name="Ada Lovelace",
|
|
email="ada@example.test",
|
|
source_module="access",
|
|
source_label="Accounts",
|
|
source_ref="access:account:account-2",
|
|
source_revision="revision-1",
|
|
description="Research",
|
|
provenance={"tenant_id": "tenant-1", "internal": "secret"},
|
|
metadata={"internal_group_ids": ["group-1"]},
|
|
),
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
def _principal(*scopes: str) -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="membership-1",
|
|
tenant_id="tenant-1",
|
|
email="organizer@example.test",
|
|
display_name="Organizer",
|
|
scopes=frozenset(scopes),
|
|
),
|
|
account=SimpleNamespace(id="account-1"),
|
|
user=SimpleNamespace(id="membership-1"),
|
|
)
|
|
|
|
|
|
class SchedulingPeopleSearchTests(unittest.TestCase):
|
|
def tearDown(self) -> None:
|
|
configure_runtime(registry=_Registry())
|
|
|
|
def test_search_uses_principal_aware_core_aggregator_and_redacts_provider_internals(self) -> None:
|
|
provider = _PeopleProvider()
|
|
registry = _Registry({CAPABILITY_ACCESS_PEOPLE_SEARCH: provider})
|
|
configure_runtime(registry=registry)
|
|
session = object()
|
|
principal = _principal(WRITE_SCOPE)
|
|
|
|
response = api_search_scheduling_people(
|
|
query="ada",
|
|
limit=12,
|
|
session=session, # type: ignore[arg-type] - provider contract is intentionally generic
|
|
principal=principal,
|
|
)
|
|
|
|
self.assertEqual([(session, principal, "ada", 12)], provider.calls)
|
|
payload = response.model_dump()
|
|
self.assertEqual("account:account-2", payload["groups"][0]["candidates"][0]["selection_key"])
|
|
self.assertEqual("revision-1", payload["groups"][0]["candidates"][0]["source_revision"])
|
|
self.assertNotIn("source_ref", payload["groups"][0]["candidates"][0])
|
|
self.assertNotIn("provenance", payload["groups"][0]["candidates"][0])
|
|
self.assertNotIn("metadata", payload["groups"][0]["candidates"][0])
|
|
|
|
def test_search_is_empty_when_no_optional_directory_provider_is_installed(self) -> None:
|
|
configure_runtime(registry=_Registry())
|
|
|
|
response = api_search_scheduling_people(
|
|
query="ada",
|
|
limit=25,
|
|
session=object(), # type: ignore[arg-type]
|
|
principal=_principal(WRITE_SCOPE),
|
|
)
|
|
|
|
self.assertEqual([], response.groups)
|
|
|
|
def test_search_requires_scheduling_write_or_admin_access(self) -> None:
|
|
provider = _PeopleProvider()
|
|
configure_runtime(registry=_Registry({CAPABILITY_ACCESS_PEOPLE_SEARCH: provider}))
|
|
|
|
with self.assertRaises(HTTPException) as raised:
|
|
api_search_scheduling_people(
|
|
query="ada",
|
|
limit=25,
|
|
session=object(), # type: ignore[arg-type]
|
|
principal=_principal("scheduling:schedule:read"),
|
|
)
|
|
|
|
self.assertEqual(403, raised.exception.status_code)
|
|
self.assertEqual([], provider.calls)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|