126 lines
5.5 KiB
Python
126 lines
5.5 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from govoplan_core.core.configuration_safety import (
|
|
classify_configuration_field,
|
|
configuration_safety_catalog,
|
|
high_impact_configuration_fields,
|
|
plan_configuration_change,
|
|
ui_managed_configuration_fields_requiring_approval,
|
|
)
|
|
from govoplan_core.core.policy import (
|
|
PolicyDecision,
|
|
PolicySourceStep,
|
|
parse_policy_source_path,
|
|
policy_source_path,
|
|
policy_source_step,
|
|
)
|
|
|
|
|
|
class PolicyContractTests(unittest.TestCase):
|
|
def test_policy_source_paths_are_stable_and_round_trip(self) -> None:
|
|
self.assertEqual(policy_source_path("system"), "system")
|
|
self.assertEqual(policy_source_path("tenant", "tenant-1"), "tenant:tenant-1")
|
|
self.assertEqual(policy_source_path("campaign", "campaign/with space"), "campaign:campaign%2Fwith%20space")
|
|
|
|
tenant_ref = parse_policy_source_path("tenant:tenant-1")
|
|
self.assertEqual(tenant_ref.scope_type, "tenant")
|
|
self.assertEqual(tenant_ref.scope_id, "tenant-1")
|
|
self.assertEqual(tenant_ref.path, "tenant:tenant-1")
|
|
|
|
campaign_ref = parse_policy_source_path("campaign:campaign%2Fwith%20space")
|
|
self.assertEqual(campaign_ref.scope_id, "campaign/with space")
|
|
|
|
def test_policy_source_step_and_decision_serialize_for_api_responses(self) -> None:
|
|
source = policy_source_step(
|
|
"tenant",
|
|
"Tenant",
|
|
"tenant-1",
|
|
applied_fields=("allow_lower_level_limits",),
|
|
policy={"allow_lower_level_limits": {"audit_detail_level": False}},
|
|
)
|
|
self.assertEqual(source["path"], "tenant:tenant-1")
|
|
self.assertEqual(source["applied_fields"], ["allow_lower_level_limits"])
|
|
|
|
decision = PolicyDecision(
|
|
allowed=False,
|
|
reason="Parent policy locks lower-level changes.",
|
|
source_path=(PolicySourceStep.from_mapping(source),),
|
|
requirements=("audit_detail_level",),
|
|
details={"blocked_fields": ["audit_detail_level"]},
|
|
)
|
|
payload = decision.to_dict()
|
|
self.assertFalse(payload["allowed"])
|
|
self.assertEqual(payload["source_path"][0]["path"], "tenant:tenant-1")
|
|
self.assertEqual(payload["requirements"], ["audit_detail_level"])
|
|
|
|
def test_configuration_safety_catalog_classifies_ui_and_env_managed_fields(self) -> None:
|
|
catalog = {item.key: item for item in configuration_safety_catalog()}
|
|
|
|
module_plan = catalog["module_management.install_plan"]
|
|
self.assertTrue(module_plan.ui_managed)
|
|
self.assertEqual(module_plan.risk, "destructive")
|
|
self.assertTrue(module_plan.dry_run_required)
|
|
self.assertTrue(module_plan.maintenance_required)
|
|
self.assertTrue(module_plan.two_person_approval_required)
|
|
self.assertTrue(module_plan.rollback_history_required)
|
|
|
|
connector_profiles = catalog["files.connector_profiles"]
|
|
self.assertEqual(connector_profiles.secret_handling, "reference_only")
|
|
self.assertTrue(connector_profiles.policy_explanation_required)
|
|
self.assertIn("files:file:admin", connector_profiles.required_scopes)
|
|
|
|
database_url = catalog["DATABASE_URL"]
|
|
self.assertFalse(database_url.ui_managed)
|
|
self.assertEqual(database_url.secret_handling, "env_only")
|
|
self.assertEqual(database_url.storage, "environment")
|
|
|
|
self.assertIs(classify_configuration_field("MASTER_KEY_B64"), catalog["MASTER_KEY_B64"])
|
|
self.assertIsNone(classify_configuration_field("missing.setting"))
|
|
self.assertNotIn("DATABASE_URL", {item.key for item in configuration_safety_catalog(include_env_only=False)})
|
|
|
|
def test_configuration_safety_helpers_surface_guardrail_sets(self) -> None:
|
|
high_impact = {item.key for item in high_impact_configuration_fields()}
|
|
approval = {item.key for item in ui_managed_configuration_fields_requiring_approval()}
|
|
|
|
self.assertIn("module_management.install_plan", high_impact)
|
|
self.assertIn("privacy_retention_policy", approval)
|
|
self.assertIn("files.connector_profiles", approval)
|
|
self.assertNotIn("DATABASE_URL", approval)
|
|
|
|
def test_configuration_change_safety_plan_enforces_guardrails(self) -> None:
|
|
blocked = plan_configuration_change(
|
|
"files.connector_profiles",
|
|
actor_scopes=("tenant:*",),
|
|
value={"password": "plain-secret"},
|
|
dry_run=False,
|
|
maintenance_mode=False,
|
|
approval_count=1,
|
|
)
|
|
self.assertFalse(blocked.allowed)
|
|
self.assertIn("dry_run_required", blocked.blockers)
|
|
self.assertIn("two_person_approval_required", blocked.blockers)
|
|
self.assertIn("secret_reference_required", blocked.blockers)
|
|
self.assertEqual(blocked.audit_event, "files.connector_profile.updated")
|
|
self.assertTrue(blocked.rollback_history_required)
|
|
|
|
allowed = plan_configuration_change(
|
|
"files.connector_profiles",
|
|
actor_scopes=("tenant:*",),
|
|
value={"secret_ref": "vault://files/smb"},
|
|
dry_run=True,
|
|
approval_count=2,
|
|
)
|
|
self.assertTrue(allowed.allowed, allowed.to_dict())
|
|
self.assertEqual(allowed.secret_handling, "reference_only")
|
|
|
|
env_only = plan_configuration_change("DATABASE_URL", actor_scopes=("system:*",), value="postgresql://example")
|
|
self.assertFalse(env_only.allowed)
|
|
self.assertIn("deployment_managed", env_only.blockers)
|
|
self.assertIn("env_only_secret", env_only.blockers)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|