from __future__ import annotations from datetime import UTC, datetime import unittest from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef from govoplan_core.core.concurrency import RevisionConflictError from govoplan_core.core.dataflows import ( DataflowDatasetDescriptor, DataflowDatasetResult, ) from govoplan_core.core.distribution_lists import ( DistributionChannelPolicyDecision, DistributionExpansionLimits, DistributionExpansionRequest, DistributionListConflictError, ) from govoplan_core.db.base import Base from govoplan_core.db.session import configure_database, reset_database from govoplan_dist_lists.backend.db.models import ( DistributionList, DistributionListEntry, DistributionListRevision, DistributionListSnapshot, ) from govoplan_dist_lists.backend.capabilities import SqlDistributionListCapabilities from govoplan_dist_lists.backend.expansion import expand_distribution_list from govoplan_dist_lists.backend.schemas import ( DistributionListCreateRequest, DistributionListUpdateRequest, ) from govoplan_dist_lists.backend.service import ( create_distribution_list, get_distribution_list, update_distribution_list, ) def principal( tenant_id: str = "tenant-1", *, account_id: str = "account-1", group_ids: frozenset[str] = frozenset(), admin: bool = True, ) -> ApiPrincipal: scopes = { "dist_lists:list:read", "dist_lists:list:write", } if admin: scopes.add("dist_lists:list:admin") return ApiPrincipal( principal=PrincipalRef( account_id=account_id, membership_id="membership-1", tenant_id=tenant_id, identity_id="identity-1", scopes=frozenset(scopes), group_ids=group_ids, ), account=object(), user=object(), ) def raw_payload(name: str, email: str) -> DistributionListCreateRequest: return DistributionListCreateRequest.model_validate( { "name": name, "constraints": {"default_channel": "email"}, "entries": [ { "entry_key": "primary-email", "kind": "raw_email", "source": { "provider": "local", "resource_type": "email", "resource_id": email, }, } ], } ) class _Dataflow: def list_outputs(self, session, principal, *, query="", limit=100): del session, principal, query, limit return ( DataflowDatasetDescriptor( pipeline_ref="pipeline-1", name="Audience", revision=3, definition_hash="definition-a", status="active", ), ) def read_output(self, session, principal, *, request): del session, principal self.request = request return DataflowDatasetResult( pipeline_ref=request.pipeline_ref, revision=request.revision, definition_hash="definition-a", rows=( { "recipient_key": "person-1", "display_name": "Ada Example", "email": "ada@example.test", "postal_address": "Example Street 1", "identity_id": "identity-a", }, { "recipient_key": "person-2", "display_name": "Blocked Example", "email": "blocked@example.test", "identity_id": "identity-b", }, { "recipient_key": "person-3", "display_name": "Suppressed Example", "postal_address": "Example Street 3", "identity_id": "identity-c", "selected_channel": "postal", "contact_point_id": "contact-postal-c", "distribution_status": "suppressed", "exclusion_reason": "preference.suppressed", "policy_decision": "preference.suppressed", }, ), total_rows=2, truncated=False, output_hash="output-a", executor_version="test", source_fingerprints=({"source": "fixture", "fingerprint": "input-a"},), generated_at=datetime(2026, 1, 1, tzinfo=UTC), ) class _Policy: def resolve_distribution_channel(self, session, principal, *, request): del session, principal allowed = "blocked" not in request.candidate.target return DistributionChannelPolicyDecision( allowed=allowed, reason_code="allowed" if allowed else "policy.suppressed", explanation="Allowed by test Policy." if allowed else "Suppressed by test Policy.", source_path=({"scope_type": "tenant", "scope_id": request.tenant_id},), ) class _Registry: def __init__(self, capabilities=None) -> None: self.capabilities = dict(capabilities or {}) def has_capability(self, name: str) -> bool: return name in self.capabilities def capability(self, name: str): return self.capabilities.get(name) class DistributionListServiceTests(unittest.TestCase): def setUp(self) -> None: self.database = configure_database("sqlite:///:memory:") Base.metadata.create_all( self.database.engine, tables=[ DistributionList.__table__, DistributionListRevision.__table__, DistributionListEntry.__table__, DistributionListSnapshot.__table__, ], ) def tearDown(self) -> None: reset_database(dispose=True) def test_local_list_freezes_reproducible_snapshot_and_preserves_history(self) -> None: effective_at = datetime(2026, 1, 1, tzinfo=UTC) with self.database.session() as session: item, revision = create_distribution_list( session, principal(), raw_payload("Monthly recipients", "first@example.test"), ) session.flush() first = expand_distribution_list( session, principal(), registry=_Registry(), request=DistributionExpansionRequest( list_id=item.id, effective_at=effective_at, freeze=True, idempotency_key="monthly-2026-01", ), ) session.commit() self.assertEqual(1, len(first.recipients)) self.assertIsNotNone(first.snapshot_id) self.assertEqual("first@example.test", first.recipients[0].channels[0].target) preview = expand_distribution_list( session, principal(), registry=_Registry(), request=DistributionExpansionRequest( list_id=item.id, effective_at=effective_at, preview=True, ), ) self.assertEqual(first.expansion_hash, preview.expansion_hash) update = DistributionListUpdateRequest.model_validate( { **raw_payload("Monthly recipients", "second@example.test").model_dump(mode="json"), "base_revision": 1, } ) item, current = update_distribution_list( session, principal(), item, update, ) session.commit() snapshot = session.get(DistributionListSnapshot, first.snapshot_id) self.assertEqual(1, snapshot.revision_number) self.assertEqual("first@example.test", snapshot.recipients[0]["channels"][0]["target"]) self.assertEqual(2, current.revision) self.assertEqual(2, item.resource_revision) self.assertNotEqual(revision.definition_hash, current.definition_hash) def test_stale_resource_revision_is_rejected(self) -> None: with self.database.session() as session: item, _ = create_distribution_list( session, principal(), raw_payload("OCC", "one@example.test"), ) session.flush() payload = DistributionListUpdateRequest.model_validate( { **raw_payload("OCC", "two@example.test").model_dump(mode="json"), "base_revision": 1, } ) update_distribution_list(session, principal(), item, payload) session.flush() with self.assertRaises(RevisionConflictError): update_distribution_list(session, principal(), item, payload) def test_expansion_applies_an_aggregate_candidate_budget(self) -> None: with self.database.session() as session: item, _ = create_distribution_list( session, principal(), DistributionListCreateRequest.model_validate( { "name": "Bounded", "entries": [ { "entry_key": f"email-{index}", "kind": "raw_email", "source": { "provider": "local", "resource_type": "email", "resource_id": f"person-{index}@example.test", }, } for index in range(4) ], } ), ) result = expand_distribution_list( session, principal(), registry=_Registry(), request=DistributionExpansionRequest( list_id=item.id, limits=DistributionExpansionLimits( max_entries=10, max_results=2, max_provider_results=1, ), ), ) self.assertTrue(result.truncated) self.assertEqual(2, len(result.recipients)) self.assertIn( "expansion.candidate_limit", {diagnostic.code for diagnostic in result.diagnostics}, ) def test_non_admin_cannot_claim_another_user_or_group_scope(self) -> None: actor = principal(group_ids=frozenset({"group-own"}), admin=False) with self.database.session() as session: own_user, _ = create_distribution_list( session, actor, DistributionListCreateRequest( name="Mine", scope_type="user", scope_id=actor.account_id, ), ) own_group, _ = create_distribution_list( session, actor, DistributionListCreateRequest( name="Ours", scope_type="group", scope_id="group-own", ), ) self.assertEqual(actor.account_id, own_user.scope_id) self.assertEqual("group-own", own_group.scope_id) for scope_type, scope_id in ( ("user", "account-other"), ("group", "group-other"), ): with self.assertRaises(DistributionListConflictError): create_distribution_list( session, actor, DistributionListCreateRequest( name=f"Blocked {scope_type}", scope_type=scope_type, scope_id=scope_id, ), ) other_user, _ = create_distribution_list( session, principal(account_id="admin"), DistributionListCreateRequest( name="Delegated", scope_type="user", scope_id="account-other", ), ) self.assertEqual( "account-other", get_distribution_list( session, principal(account_id="admin"), other_user.id, ).scope_id, ) decision = SqlDistributionListCapabilities().explain_write( session, principal(account_id="admin"), list_id=other_user.id, operation="update", ) self.assertTrue(decision.allowed) def test_tenant_isolation_and_nested_cycle_diagnostic(self) -> None: with self.database.session() as session: first, _ = create_distribution_list( session, principal(), DistributionListCreateRequest(name="First"), ) second, _ = create_distribution_list( session, principal(), DistributionListCreateRequest.model_validate( { "name": "Second", "entries": [_nested_entry(first.id, "first")], } ), ) first_update = DistributionListUpdateRequest.model_validate( { "name": "First", "base_revision": 1, "entries": [_nested_entry(second.id, "second")], } ) update_distribution_list(session, principal(), first, first_update) session.flush() result = expand_distribution_list( session, principal(), registry=_Registry(), request=DistributionExpansionRequest(list_id=first.id), ) self.assertIn("expansion.nested_cycle", {item.code for item in result.diagnostics}) with self.assertRaises(ValueError): get_distribution_list(session, principal("tenant-2"), first.id) def test_dataflow_parameters_policy_and_missing_providers_are_explained(self) -> None: dataflow = _Dataflow() registry = _Registry( { "dataflow.dataset_output": dataflow, "policy.distribution_channels": _Policy(), } ) with self.database.session() as session: item, _ = create_distribution_list( session, principal(), DistributionListCreateRequest.model_validate( { "name": "Dynamic", "definition_kind": "parameterized", "parameters": [ { "key": "region", "value_type": "string", "required": True, "allowed_values": ["north", "south"], } ], "entries": [ { "entry_key": "flow", "kind": "dataflow_result", "source": { "provider": "dataflow", "resource_type": "pipeline_output", "resource_id": "pipeline-1", "revision": "3", "fingerprint": "definition-a", }, }, { "entry_key": "group", "kind": "idm_group", "source": { "provider": "idm", "resource_type": "group", "resource_id": "group-1", }, }, ], } ), ) session.flush() result = expand_distribution_list( session, principal(), registry=registry, request=DistributionExpansionRequest( list_id=item.id, parameters={"region": "north"}, effective_at=datetime(2026, 1, 1, tzinfo=UTC), ), ) self.assertEqual({"person-1"}, {item.recipient_key for item in result.recipients}) self.assertIn("policy.suppressed", {item.channels[0].reason_code for item in result.excluded if item.channels}) explicitly_suppressed = next( row for row in result.excluded if row.recipient_key == "person-3" ) self.assertEqual("suppressed", explicitly_suppressed.status) self.assertEqual( "contact-postal-c", explicitly_suppressed.channels[0].contact_point_id, ) self.assertEqual( "preference.suppressed", explicitly_suppressed.explanations[0].code, ) self.assertIn("provider_unavailable", {item.status for item in result.excluded}) self.assertEqual("north", dataflow.request.parameters["region"]) postal = expand_distribution_list( session, principal(), registry=registry, request=DistributionExpansionRequest( list_id=item.id, parameters={"region": "north"}, requested_channels=("postal",), effective_at=datetime(2026, 1, 1, tzinfo=UTC), ), ) ada = next(row for row in postal.recipients if row.recipient_key == "person-1") by_channel = {candidate.channel: candidate for candidate in ada.channels} self.assertEqual("suppressed", by_channel["email"].status) self.assertEqual("channel.not_requested", by_channel["email"].reason_code) self.assertEqual("usable", by_channel["postal"].status) def _nested_entry(list_id: str, key: str) -> dict[str, object]: return { "entry_key": key, "kind": "distribution_list", "source": { "provider": "dist_lists", "resource_type": "distribution_list", "resource_id": list_id, }, } if __name__ == "__main__": unittest.main()