feat: complete postbox access evidence
This commit is contained in:
@@ -0,0 +1,187 @@
|
||||
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,
|
||||
):
|
||||
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]
|
||||
),
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
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",
|
||||
"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_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()
|
||||
+120
-1
@@ -49,7 +49,7 @@ from govoplan_postbox.backend.db.models import (
|
||||
PostboxTemplate,
|
||||
PostboxTemplateRevision,
|
||||
)
|
||||
from govoplan_postbox.backend.service import PostboxService
|
||||
from govoplan_postbox.backend.service import PostboxError, PostboxService
|
||||
|
||||
|
||||
POSTBOX_TABLES = (
|
||||
@@ -915,6 +915,12 @@ class PostboxServiceTests(unittest.TestCase):
|
||||
actor=self.actor,
|
||||
state="acknowledged",
|
||||
)
|
||||
receipt_summary = service.delivery_receipt_summaries(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
producer_module="campaigns",
|
||||
delivery_ids=(first.delivery_id,),
|
||||
)[first.delivery_id]
|
||||
session.commit()
|
||||
|
||||
self.assertFalse(first.duplicate)
|
||||
@@ -922,6 +928,20 @@ class PostboxServiceTests(unittest.TestCase):
|
||||
self.assertEqual(first.message_id, second.message_id)
|
||||
self.assertIsNotNone(marked.read_at)
|
||||
self.assertIsNotNone(marked.acknowledged_at)
|
||||
self.assertTrue(receipt_summary.currently_readable)
|
||||
self.assertEqual(1, receipt_summary.current_holder_count)
|
||||
self.assertEqual(1, receipt_summary.read_receipt_count)
|
||||
self.assertEqual(1, receipt_summary.acknowledged_receipt_count)
|
||||
self.assertEqual(0, receipt_summary.withdrawn_message_count)
|
||||
self.assertEqual(
|
||||
{},
|
||||
service.delivery_receipt_summaries(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
producer_module="another-module",
|
||||
delivery_ids=(first.delivery_id,),
|
||||
),
|
||||
)
|
||||
self.assertEqual(
|
||||
1,
|
||||
session.query(PostboxMessage).count(),
|
||||
@@ -971,6 +991,95 @@ class PostboxServiceTests(unittest.TestCase):
|
||||
[event.type for event in events],
|
||||
)
|
||||
|
||||
def test_withdrawn_and_expired_messages_keep_metadata_but_hide_content(
|
||||
self,
|
||||
) -> None:
|
||||
self.idm.assignments.append(self.assignment)
|
||||
with Session(self.engine) as session:
|
||||
postbox = self._create_exact(session)
|
||||
withdrawn = self.service.deliver(
|
||||
session,
|
||||
PostboxDeliveryRequest(
|
||||
tenant_id="tenant-1",
|
||||
target=PostboxTargetRef(postbox_id=postbox.id),
|
||||
producer_module="campaigns",
|
||||
producer_resource_type="campaign_recipient",
|
||||
producer_resource_id="recipient-withdrawn",
|
||||
idempotency_key="withdrawn-message",
|
||||
subject="Withdrawn decision",
|
||||
body_text="Sensitive withdrawn content",
|
||||
),
|
||||
)
|
||||
expired = self.service.deliver(
|
||||
session,
|
||||
PostboxDeliveryRequest(
|
||||
tenant_id="tenant-1",
|
||||
target=PostboxTargetRef(postbox_id=postbox.id),
|
||||
producer_module="campaigns",
|
||||
producer_resource_type="campaign_recipient",
|
||||
producer_resource_id="recipient-expired",
|
||||
idempotency_key="expired-message",
|
||||
subject="Expired decision",
|
||||
body_text="Sensitive expired content",
|
||||
expires_at=utc_now() - timedelta(minutes=1),
|
||||
),
|
||||
)
|
||||
withdrawn_model = session.get(PostboxMessage, withdrawn.message_id)
|
||||
assert withdrawn_model is not None
|
||||
withdrawn_model.withdrawn_at = utc_now()
|
||||
session.flush()
|
||||
|
||||
withdrawn_ref = self.service.get_message(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
message_id=withdrawn.message_id,
|
||||
actor=self.actor,
|
||||
)
|
||||
expired_ref = self.service.get_message(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
message_id=expired.message_id,
|
||||
actor=self.actor,
|
||||
)
|
||||
|
||||
assert withdrawn_ref is not None
|
||||
assert expired_ref is not None
|
||||
self.assertEqual("withdrawn", withdrawn_ref.availability)
|
||||
self.assertIsNone(withdrawn_ref.body_text)
|
||||
self.assertEqual((), withdrawn_ref.attachments)
|
||||
self.assertEqual("expired", expired_ref.availability)
|
||||
self.assertIsNone(expired_ref.body_text)
|
||||
with self.assertRaisesRegex(PostboxError, "withdrawn"):
|
||||
self.service.mark_message(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
message_id=withdrawn.message_id,
|
||||
actor=self.actor,
|
||||
state="read",
|
||||
)
|
||||
with self.assertRaisesRegex(PostboxError, "expired"):
|
||||
self.service.mark_message(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
message_id=expired.message_id,
|
||||
actor=self.actor,
|
||||
state="acknowledged",
|
||||
)
|
||||
|
||||
summaries = self.service.delivery_receipt_summaries(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
producer_module="campaigns",
|
||||
delivery_ids=(withdrawn.delivery_id, expired.delivery_id),
|
||||
)
|
||||
self.assertFalse(summaries[withdrawn.delivery_id].currently_readable)
|
||||
self.assertEqual(
|
||||
1,
|
||||
summaries[withdrawn.delivery_id].withdrawn_message_count,
|
||||
)
|
||||
self.assertFalse(summaries[expired.delivery_id].currently_readable)
|
||||
self.assertEqual(1, summaries[expired.delivery_id].expired_message_count)
|
||||
|
||||
def test_hierarchy_linked_copy_snapshots_path_and_independent_state(
|
||||
self,
|
||||
) -> None:
|
||||
@@ -1023,6 +1132,16 @@ class PostboxServiceTests(unittest.TestCase):
|
||||
session.commit()
|
||||
receipts = session.query(PostboxMessageReceipt).all()
|
||||
self.assertEqual([route.target_message_id], [item.message_id for item in receipts])
|
||||
summary = service.delivery_receipt_summaries(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
producer_module="campaigns",
|
||||
delivery_ids=(result.delivery_id,),
|
||||
)[result.delivery_id]
|
||||
self.assertEqual(2, summary.message_count)
|
||||
self.assertEqual(1, summary.routed_message_count)
|
||||
self.assertEqual(1, summary.read_receipt_count)
|
||||
self.assertEqual({"accepted": 1}, summary.route_status_counts)
|
||||
|
||||
def test_hierarchy_dry_run_explains_gates_depth_and_parallel_structure(
|
||||
self,
|
||||
|
||||
Reference in New Issue
Block a user