120 lines
4.0 KiB
Python
120 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.policy import PolicyImpactPopulationRequest
|
|
from govoplan_core.core.views import ViewSurface
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_views.backend.db.models import ViewDefinition
|
|
from govoplan_views.backend.impact_subjects import (
|
|
ViewsPolicyImpactSubjectProvider,
|
|
)
|
|
|
|
|
|
class _Registry:
|
|
def view_surfaces(self) -> tuple[ViewSurface, ...]:
|
|
return (
|
|
ViewSurface(
|
|
id="views.selector",
|
|
module_id="views",
|
|
kind="selector",
|
|
label="View selector",
|
|
),
|
|
)
|
|
|
|
|
|
class ViewsPolicyImpactSubjectTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(self.engine)
|
|
self.session = Session(self.engine)
|
|
self.provider = ViewsPolicyImpactSubjectProvider(_Registry())
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def _definition(self, *, tenant_id: str, name: str) -> ViewDefinition:
|
|
definition = ViewDefinition(
|
|
tenant_id=tenant_id,
|
|
scope_type="tenant",
|
|
scope_id=None,
|
|
scope_key=f"tenant:{tenant_id}",
|
|
definition_key=name.casefold(),
|
|
name=name,
|
|
status="published",
|
|
)
|
|
self.session.add(definition)
|
|
self.session.flush()
|
|
return definition
|
|
|
|
def test_provider_is_tenant_filtered_and_returns_bounded_actions(self) -> None:
|
|
included = self._definition(tenant_id="tenant-1", name="Included")
|
|
self._definition(tenant_id="tenant-2", name="Hidden")
|
|
|
|
batch = self.provider.collect_policy_impact_subjects(
|
|
self.session,
|
|
request=PolicyImpactPopulationRequest(
|
|
tenant_id="tenant-1",
|
|
policy_family="view",
|
|
selector={
|
|
"actions": ["view", "edit"],
|
|
"include_surfaces": False,
|
|
},
|
|
limit=10,
|
|
allow_sensitive_details=True,
|
|
),
|
|
)
|
|
|
|
self.assertEqual("complete", batch.state)
|
|
self.assertEqual(2, batch.total_available)
|
|
self.assertEqual({included.id}, {item.resource_id for item in batch.subjects})
|
|
self.assertEqual({"view", "edit"}, {item.action for item in batch.subjects})
|
|
self.assertEqual({"Included"}, {item.label for item in batch.subjects})
|
|
|
|
def test_provider_reports_truncation_and_hides_labels(self) -> None:
|
|
self._definition(tenant_id="tenant-1", name="One")
|
|
self._definition(tenant_id="tenant-1", name="Two")
|
|
|
|
batch = self.provider.collect_policy_impact_subjects(
|
|
self.session,
|
|
request=PolicyImpactPopulationRequest(
|
|
tenant_id="tenant-1",
|
|
policy_family="view",
|
|
selector={"include_surfaces": True},
|
|
limit=2,
|
|
),
|
|
)
|
|
|
|
self.assertEqual("truncated", batch.state)
|
|
self.assertGreater(batch.total_available or 0, len(batch.subjects))
|
|
self.assertEqual(2, len(batch.subjects))
|
|
self.assertTrue(all(subject.label is None for subject in batch.subjects))
|
|
|
|
def test_explicit_empty_population_never_falls_back_to_catalogue_scan(self) -> None:
|
|
self._definition(tenant_id="tenant-1", name="Not requested")
|
|
|
|
batch = self.provider.collect_policy_impact_subjects(
|
|
self.session,
|
|
request=PolicyImpactPopulationRequest(
|
|
tenant_id="tenant-1",
|
|
policy_family="view",
|
|
selector={
|
|
"view_ids": [],
|
|
"include_surfaces": False,
|
|
},
|
|
limit=10,
|
|
),
|
|
)
|
|
|
|
self.assertEqual("complete", batch.state)
|
|
self.assertEqual(0, batch.total_available)
|
|
self.assertEqual((), batch.subjects)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|