179 lines
6.8 KiB
Python
179 lines
6.8 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.admin.models import SystemSettings
|
|
from govoplan_core.core.modules import ModuleContext
|
|
from govoplan_core.core.policy import (
|
|
CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY,
|
|
SchedulingParticipantPrivacyPolicy,
|
|
SchedulingParticipantPrivacyRequest,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.tenancy.scope import Tenant, scope_registry
|
|
from govoplan_policy.backend.manifest import manifest
|
|
from govoplan_policy.backend.scheduling_privacy import (
|
|
SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY,
|
|
SqlSchedulingParticipantPrivacyPolicy,
|
|
)
|
|
|
|
|
|
def _policy_settings(maximum_visibility: object) -> dict[str, object]:
|
|
return {
|
|
SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY: {
|
|
"maximum_visibility": maximum_visibility,
|
|
}
|
|
}
|
|
|
|
|
|
class SchedulingParticipantPrivacyTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(bind=self.engine, tables=[SystemSettings.__table__])
|
|
scope_registry.metadata.create_all(bind=self.engine, tables=[Tenant.__table__])
|
|
self.session = Session(self.engine)
|
|
self.provider = SqlSchedulingParticipantPrivacyPolicy()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def _add_settings(
|
|
self,
|
|
*,
|
|
system: dict[str, object] | None = None,
|
|
tenant: dict[str, object] | None = None,
|
|
) -> None:
|
|
self.session.add(SystemSettings(id="global", settings=system or {}))
|
|
self.session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant 1", settings=tenant or {}))
|
|
self.session.flush()
|
|
|
|
def _resolve(self, requested_visibility: str = "names_and_statuses"):
|
|
return self.provider.resolve_scheduling_participant_visibility(
|
|
self.session,
|
|
request=SchedulingParticipantPrivacyRequest(
|
|
tenant_id="tenant-1",
|
|
scheduling_request_id="request-1",
|
|
participant_id="participant-1",
|
|
requested_visibility=requested_visibility, # type: ignore[arg-type]
|
|
actor_user_id="user-1",
|
|
),
|
|
)
|
|
|
|
def test_manifest_registers_the_core_capability(self) -> None:
|
|
factory = manifest.capability_factories[CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY]
|
|
|
|
provider = factory(ModuleContext(registry=object(), settings=object()))
|
|
|
|
self.assertIsInstance(provider, SchedulingParticipantPrivacyPolicy)
|
|
|
|
def test_missing_policy_preserves_scheduling_visibility(self) -> None:
|
|
self._add_settings()
|
|
|
|
decision = self._resolve()
|
|
|
|
self.assertEqual("names_and_statuses", decision.effective_visibility)
|
|
self.assertIsNone(decision.reason)
|
|
self.assertEqual((), decision.source_path)
|
|
self.assertEqual([], decision.details["configured_scopes"])
|
|
|
|
def test_missing_system_settings_is_an_unrestricted_read_only_default(self) -> None:
|
|
self.session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant 1", settings={}))
|
|
self.session.flush()
|
|
|
|
decision = self._resolve()
|
|
|
|
self.assertEqual("names_and_statuses", decision.effective_visibility)
|
|
self.assertIsNone(self.session.get(SystemSettings, "global"))
|
|
|
|
def test_system_ceiling_restricts_participant_roster(self) -> None:
|
|
self._add_settings(system=_policy_settings("aggregates_only"))
|
|
|
|
decision = self._resolve()
|
|
|
|
self.assertEqual("aggregates_only", decision.effective_visibility)
|
|
self.assertEqual("Participant roster visibility is restricted by policy.", decision.reason)
|
|
self.assertEqual(["system"], [step.path for step in decision.source_path])
|
|
self.assertEqual("aggregates_only", decision.source_path[0].policy["maximum_visibility"])
|
|
|
|
def test_tenant_ceiling_can_narrow_system_but_cannot_widen_it(self) -> None:
|
|
self._add_settings(
|
|
system=_policy_settings("aggregates_only"),
|
|
tenant=_policy_settings("names_and_statuses"),
|
|
)
|
|
|
|
decision = self._resolve()
|
|
|
|
self.assertEqual("aggregates_only", decision.effective_visibility)
|
|
self.assertEqual(["system", "tenant:tenant-1"], [step.path for step in decision.source_path])
|
|
self.assertEqual("aggregates_only", decision.details["policy_ceiling"])
|
|
|
|
def test_policy_never_broadens_an_aggregate_only_request(self) -> None:
|
|
self._add_settings(
|
|
system=_policy_settings("names_and_statuses"),
|
|
tenant=_policy_settings("names_and_statuses"),
|
|
)
|
|
|
|
decision = self._resolve("aggregates_only")
|
|
|
|
self.assertEqual("aggregates_only", decision.effective_visibility)
|
|
self.assertIsNone(decision.reason)
|
|
|
|
def test_invalid_explicit_policy_fails_closed_without_echoing_value(self) -> None:
|
|
self._add_settings(tenant=_policy_settings("surprise"))
|
|
|
|
decision = self._resolve()
|
|
|
|
self.assertEqual("aggregates_only", decision.effective_visibility)
|
|
self.assertIn("could not be validated", decision.reason or "")
|
|
self.assertEqual(
|
|
[{"scope": "tenant", "code": "invalid_maximum_visibility"}],
|
|
decision.details["configuration_errors"],
|
|
)
|
|
self.assertNotIn("surprise", repr(decision.to_dict()))
|
|
self.assertEqual("invalid_fail_closed", decision.source_path[0].policy["configuration_status"])
|
|
|
|
def test_invalid_policy_shape_fails_closed(self) -> None:
|
|
self._add_settings(
|
|
system={SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY: "not-an-object"},
|
|
)
|
|
|
|
decision = self._resolve()
|
|
|
|
self.assertEqual("aggregates_only", decision.effective_visibility)
|
|
self.assertEqual(
|
|
[{"scope": "system", "code": "invalid_policy_shape"}],
|
|
decision.details["configuration_errors"],
|
|
)
|
|
|
|
def test_missing_tenant_fails_closed(self) -> None:
|
|
self.session.add(SystemSettings(id="global", settings={}))
|
|
self.session.flush()
|
|
|
|
decision = self._resolve()
|
|
|
|
self.assertEqual("aggregates_only", decision.effective_visibility)
|
|
self.assertEqual(
|
|
[{"scope": "tenant", "code": "tenant_not_found"}],
|
|
decision.details["configuration_errors"],
|
|
)
|
|
self.assertEqual(["tenant:tenant-1"], [step.path for step in decision.source_path])
|
|
|
|
def test_invalid_requested_visibility_fails_closed(self) -> None:
|
|
self._add_settings()
|
|
|
|
decision = self._resolve("invalid")
|
|
|
|
self.assertEqual("aggregates_only", decision.effective_visibility)
|
|
self.assertEqual(
|
|
[{"scope": "request", "code": "invalid_requested_visibility"}],
|
|
decision.details["configuration_errors"],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|