92 lines
2.7 KiB
Python
92 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from govoplan_core.core.policy import (
|
|
CAPABILITY_POLICY_IMPACT_SUBJECT_PREFIX,
|
|
PolicyImpactPopulationRequest,
|
|
PolicyImpactSubject,
|
|
PolicyImpactSubjectBatch,
|
|
PolicyImpactSubjectProvider,
|
|
policy_impact_subject_provider,
|
|
)
|
|
|
|
|
|
class _Provider:
|
|
provider_id = "example"
|
|
supported_policy_families = ("view",)
|
|
|
|
def collect_policy_impact_subjects(
|
|
self,
|
|
session: object | None = None,
|
|
*,
|
|
request: PolicyImpactPopulationRequest,
|
|
) -> PolicyImpactSubjectBatch:
|
|
del session, request
|
|
return PolicyImpactSubjectBatch(
|
|
provider_id=self.provider_id,
|
|
subjects=(
|
|
PolicyImpactSubject(
|
|
module_id="example",
|
|
resource_type="record",
|
|
resource_id="record-1",
|
|
action="view",
|
|
),
|
|
),
|
|
total_available=1,
|
|
)
|
|
|
|
|
|
class _Registry:
|
|
def __init__(self, capability: object) -> None:
|
|
self._capability = capability
|
|
|
|
def has_capability(self, name: str) -> bool:
|
|
return name == f"{CAPABILITY_POLICY_IMPACT_SUBJECT_PREFIX}example"
|
|
|
|
def capability(self, name: str) -> object:
|
|
del name
|
|
return self._capability
|
|
|
|
|
|
class PolicyImpactContractTests(unittest.TestCase):
|
|
def test_provider_contract_is_runtime_checkable_and_discoverable(self) -> None:
|
|
provider = _Provider()
|
|
self.assertIsInstance(provider, PolicyImpactSubjectProvider)
|
|
self.assertIs(
|
|
policy_impact_subject_provider(_Registry(provider), "example"),
|
|
provider,
|
|
)
|
|
|
|
def test_population_and_provider_batches_are_bounded(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "between 1 and 500"):
|
|
PolicyImpactPopulationRequest(
|
|
tenant_id="tenant-1",
|
|
policy_family="view",
|
|
limit=501,
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "at most 500"):
|
|
PolicyImpactSubjectBatch(
|
|
provider_id="example",
|
|
subjects=tuple(
|
|
PolicyImpactSubject(
|
|
module_id="example",
|
|
resource_type="record",
|
|
resource_id=str(index),
|
|
action="view",
|
|
)
|
|
for index in range(501)
|
|
),
|
|
)
|
|
|
|
def test_unavailable_batches_explain_the_gap(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "need an explanation"):
|
|
PolicyImpactSubjectBatch(
|
|
provider_id="example",
|
|
state="unavailable",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|