147 lines
5.0 KiB
Python
147 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.distribution_lists import (
|
|
DistributionChannelCandidate,
|
|
DistributionChannelPolicyRequest,
|
|
DistributionRecipientRef,
|
|
)
|
|
from govoplan_policy.backend.db.models import PolicyOverride
|
|
from govoplan_policy.backend.distribution_channels import (
|
|
DistributionChannelPolicyProvider,
|
|
resolve_distribution_channel_policy_rows,
|
|
validate_distribution_channel_policy,
|
|
)
|
|
|
|
|
|
class _Principal:
|
|
account_id = "account-1"
|
|
membership_id = "membership-1"
|
|
group_ids = frozenset({"group-1"})
|
|
|
|
|
|
def _request(channel: str = "email") -> DistributionChannelPolicyRequest:
|
|
candidate = DistributionChannelCandidate(
|
|
channel=channel, # type: ignore[arg-type]
|
|
target="recipient@example.test",
|
|
target_key=f"{channel}:recipient@example.test",
|
|
)
|
|
return DistributionChannelPolicyRequest(
|
|
tenant_id="tenant-1",
|
|
list_id="list-1",
|
|
purpose="monthly-notice",
|
|
effective_at=datetime(2026, 1, 1, tzinfo=UTC),
|
|
recipient=DistributionRecipientRef(
|
|
recipient_key="recipient-1",
|
|
display_name="Recipient",
|
|
status="usable",
|
|
channels=(candidate,),
|
|
),
|
|
candidate=candidate,
|
|
)
|
|
|
|
|
|
class DistributionChannelPolicyTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
PolicyOverride.__table__.create(self.engine)
|
|
|
|
def tearDown(self) -> None:
|
|
self.engine.dispose()
|
|
|
|
def test_hierarchy_can_only_reduce_permitted_channels(self) -> None:
|
|
with Session(self.engine) as session:
|
|
session.add_all(
|
|
(
|
|
PolicyOverride(
|
|
policy_family="distribution_channels",
|
|
target_key="*",
|
|
tenant_id=None,
|
|
scope_type="system",
|
|
scope_id=None,
|
|
scope_key="system",
|
|
policy={"allowed_channels": ["email", "postal"]},
|
|
),
|
|
PolicyOverride(
|
|
policy_family="distribution_channels",
|
|
target_key="purpose:monthly-notice",
|
|
tenant_id="tenant-1",
|
|
scope_type="tenant",
|
|
scope_id="tenant-1",
|
|
scope_key="tenant:tenant-1",
|
|
policy={"blocked_channels": ["postal"]},
|
|
),
|
|
)
|
|
)
|
|
session.commit()
|
|
provider = DistributionChannelPolicyProvider()
|
|
|
|
email = provider.resolve_distribution_channel(
|
|
session,
|
|
_Principal(),
|
|
request=_request("email"),
|
|
)
|
|
postal = provider.resolve_distribution_channel(
|
|
session,
|
|
_Principal(),
|
|
request=_request("postal"),
|
|
)
|
|
|
|
self.assertTrue(email.allowed)
|
|
self.assertFalse(postal.allowed)
|
|
self.assertEqual("policy.channel_blocked", postal.reason_code)
|
|
self.assertEqual(2, len(postal.source_path))
|
|
|
|
def test_malformed_policy_fails_closed_with_diagnostic(self) -> None:
|
|
row = type(
|
|
"Row",
|
|
(),
|
|
{
|
|
"policy": {"allowed_channels": "email"},
|
|
"target_key": "*",
|
|
"scope_type": "tenant",
|
|
"scope_id": "tenant-1",
|
|
"scope_key": "tenant:tenant-1",
|
|
},
|
|
)()
|
|
result = resolve_distribution_channel_policy_rows((row,))
|
|
|
|
self.assertEqual(frozenset(), result.allowed_channels)
|
|
self.assertEqual("distribution_channel_policy.invalid", result.diagnostics[0]["code"])
|
|
|
|
def test_schema_rejects_unknown_channels_and_fields(self) -> None:
|
|
self.assertEqual(
|
|
({"allowed_channels": ("email",)}, False),
|
|
validate_distribution_channel_policy({"allowed_channels": ["email"]}),
|
|
)
|
|
self.assertTrue(validate_distribution_channel_policy({"allowed_channels": ["fax"]})[1])
|
|
self.assertTrue(validate_distribution_channel_policy({"unknown": []})[1])
|
|
|
|
def test_long_purpose_is_resolved_without_exceeding_storage_key_limit(self) -> None:
|
|
request = _request("email")
|
|
request = DistributionChannelPolicyRequest(
|
|
tenant_id=request.tenant_id,
|
|
list_id=request.list_id,
|
|
purpose="purpose-" + "x" * 120,
|
|
effective_at=request.effective_at,
|
|
recipient=request.recipient,
|
|
candidate=request.candidate,
|
|
)
|
|
with Session(self.engine) as session:
|
|
decision = DistributionChannelPolicyProvider().resolve_distribution_channel(
|
|
session,
|
|
_Principal(),
|
|
request=request,
|
|
)
|
|
|
|
self.assertTrue(decision.allowed)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|