from __future__ import annotations import unittest from govoplan_core.core.idm import OrganizationFunctionAssignmentRef from govoplan_core.core.postbox import PostboxActorRef from govoplan_postbox.backend.access_decisions import ( ACCESS_DECISION_TABLE, evaluate_postbox_access, ) def assignment( source: str = "direct", *, assignment_id: str | None = None, tenant_id: str = "tenant-1", acting_for_account_id: str | None = None, ) -> OrganizationFunctionAssignmentRef: return OrganizationFunctionAssignmentRef( id=assignment_id or f"{source}-assignment", tenant_id=tenant_id, identity_id="identity-1", account_id="account-1", function_id="function-1", organization_unit_id="unit-1", source=source, # type: ignore[arg-type] acting_for_account_id=acting_for_account_id, ) def decide( *, status: str = "active", action: str = "read", authorized_actions: frozenset[str] = frozenset({"read"}), binding_available: bool = True, assignments=(), selected_assignment_id: str | None = None, acting_for_account_id: str | None = None, classification: str = "internal", authorized_classifications: frozenset[str] = frozenset( {"public", "internal"} ), binding_status: str | None = None, ): return evaluate_postbox_access( postbox_id="postbox-1", postbox_active=status == "active", action=action, # type: ignore[arg-type] actor=PostboxActorRef( account_id="account-1", identity_id="identity-1", selected_assignment_id=selected_assignment_id, acting_for_account_id=acting_for_account_id, authorized_actions=authorized_actions, # type: ignore[arg-type] authorized_classifications=authorized_classifications, # type: ignore[arg-type] ), organization_unit_id="unit-1" if binding_available else None, function_id="function-1" if binding_available else None, holder_count=len(assignments), binding_available=binding_available, binding_assignments=assignments, binding_status=binding_status, # type: ignore[arg-type] classification=classification, ) class PostboxAccessDecisionTableTests(unittest.TestCase): def test_rule_order_is_fail_closed(self) -> None: self.assertEqual( [rule.name for rule in ACCESS_DECISION_TABLE], [ "inactive_postbox", "generic_permission", "administrator", "classification_clearance", "active_function_binding", ], ) inactive_admin = decide( status="archived", action="administer", authorized_actions=frozenset({"administer"}), ) self.assertFalse(inactive_admin.allowed) self.assertEqual(inactive_admin.reason_code, "postbox_inactive") def test_permission_and_binding_denials_have_stable_provenance(self) -> None: missing_permission = decide( assignments=(assignment(),), authorized_actions=frozenset(), ) missing_binding = decide(binding_available=False) self.assertEqual( missing_permission.reason_code, "generic_permission_missing", ) self.assertEqual( missing_binding.reason_code, "function_binding_missing", ) def test_administration_is_generic_but_still_requires_active_postbox(self) -> None: decision = decide( action="administer", authorized_actions=frozenset({"administer"}), binding_available=False, ) self.assertTrue(decision.allowed) self.assertEqual(decision.reason_code, "generic_administrator") def test_reply_is_a_distinct_permission_decision(self) -> None: missing = decide( action="reply", authorized_actions=frozenset({"send"}), assignments=(assignment(),), ) allowed = decide( action="reply", authorized_actions=frozenset({"reply"}), assignments=(assignment(),), ) self.assertEqual(missing.reason_code, "generic_permission_missing") self.assertTrue(allowed.allowed) def test_classification_is_fail_closed_and_explained(self) -> None: missing_clearance = decide( classification="confidential", assignments=(assignment(),), ) allowed = decide( classification="confidential", authorized_classifications=frozenset( {"public", "internal", "confidential"} ), assignments=(assignment(),), ) unsupported = decide( classification="secret", authorized_classifications=frozenset( {"public", "internal", "confidential", "restricted"} ), assignments=(assignment(),), ) self.assertEqual( missing_clearance.reason_code, "classification_clearance_missing", ) self.assertFalse(missing_clearance.classification_allowed) self.assertTrue(allowed.allowed) self.assertEqual(unsupported.reason_code, "classification_unsupported") def test_binding_failures_have_stable_transition_reasons(self) -> None: expected = { "unit_inactive": "organization_unit_inactive", "function_inactive": "organization_function_inactive", "function_reassigned": "organization_function_reassigned", "directory_unavailable": "organization_directory_unavailable", } for binding_status, reason_code in expected.items(): with self.subTest(binding_status=binding_status): decision = decide( assignments=(assignment(),), binding_status=binding_status, ) self.assertFalse(decision.allowed) self.assertEqual(decision.reason_code, reason_code) def test_direct_delegated_directory_governance_and_system_sources_are_allowed(self) -> None: for source in ( "direct", "delegated", "directory", "governance", "system", ): with self.subTest(source=source): decision = decide(assignments=(assignment(source),)) self.assertTrue(decision.allowed) self.assertEqual( decision.reason_code, f"effective_{source}_assignment", ) def test_selected_direct_context_is_preferred_without_hiding_other_matches(self) -> None: direct = assignment("direct", assignment_id="direct-1") delegated = assignment("delegated", assignment_id="delegated-1") decision = decide( assignments=(direct, delegated), selected_assignment_id=delegated.id, ) self.assertTrue(decision.allowed) self.assertEqual(decision.selected_assignment_id, delegated.id) self.assertEqual( decision.assignment_ids, ("direct-1", "delegated-1"), ) def test_acting_access_requires_exact_assignment_and_represented_account(self) -> None: acting = assignment( "acting_for", assignment_id="acting-1", acting_for_account_id="represented-1", ) missing_context = decide(assignments=(acting,)) wrong_assignment = decide( assignments=(acting,), selected_assignment_id="acting-other", ) wrong_account = decide( assignments=(acting,), selected_assignment_id=acting.id, acting_for_account_id="represented-other", ) allowed = decide( assignments=(acting,), selected_assignment_id=acting.id, acting_for_account_id="represented-1", ) self.assertEqual( missing_context.reason_code, "acting_context_required", ) self.assertEqual( wrong_assignment.reason_code, "acting_assignment_not_selected", ) self.assertEqual( wrong_account.reason_code, "acting_account_mismatch", ) self.assertTrue(allowed.allowed) self.assertEqual( allowed.reason_code, "effective_acting_for_assignment", ) def test_vacancy_is_provenance_not_an_implicit_access_override(self) -> None: denied = decide(assignments=()) self.assertFalse(denied.allowed) self.assertTrue(denied.vacant) self.assertEqual( denied.reason_code, "effective_assignment_missing", ) if __name__ == "__main__": unittest.main()