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.contact_points import ( CONTACT_POINT_CONTRACT_VERSION, ContactPointCandidate, ContactPointResolution, ContactPointSourcePreview, ) from govoplan_core.core.dataflows import ( DataflowDatasetDescriptor, DataflowDatasetResult, ) from govoplan_core.core.distribution_lists import ( DistributionChannelPolicyDecision, DistributionExpansionLimits, DistributionExpansionRequest, DistributionExplanation, DistributionListConflictError, DistributionSourceReference, ) 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 _ContactPoints: def __init__(self) -> None: self.resolve_request = None self.source_requests = [] source = DistributionSourceReference( provider="addresses", resource_type="contact", resource_id="contact-1", revision="contact-revision-1", fingerprint="contact-fingerprint-1", label="Ada Example", ) self.resolution = ContactPointResolution( contract_version=CONTACT_POINT_CONTRACT_VERSION, subject=source, status="usable", contact_id="contact-1", display_name="Ada Example", candidates=( ContactPointCandidate( channel="postal", target="Ada Example\nMain Street 1\n10115 Berlin\nGermany", target_key="postal:main street 1|10115|berlin||germany", status="usable", contact_point_id="postal-1", address_purpose="official", locale="de-DE", preferred=True, preference_rank=1, source=source, source_revision="facts-revision-1", preference_revision="preference-revision-1", consent_revision="consent-revision-1", value={ "street": "Main Street 1", "postal_code": "10115", "locality": "Berlin", "country": "Germany", }, provenance={ "rule_ids": ["rule-1"], "source_revision": "contact-revision-1", }, ), ), excluded=( ContactPointCandidate( channel="email", target="ada@example.test", target_key="email:ada@example.test", status="suppressed", contact_point_id="email-1", reason_code="addresses.channel.opted_out", explanation="The contact opted out of email delivery.", source=source, source_revision="facts-revision-1", provenance={ "rule_ids": ["rule-2"], "source_revision": "contact-revision-1", }, ), ), explanations=( DistributionExplanation( code="addresses.channel_fact.expired", message="An older preference expired.", severity="info", provider="addresses", source=source, ), ), source_revision="facts-revision-1", source_fingerprint="facts-fingerprint-1", provenance={"active_rule_ids": ["rule-1", "rule-2"]}, ) def resolve_contact_points(self, session, principal, *, request): del session, principal self.resolve_request = request return self.resolution def preview_source(self, session, principal, *, request, offset=0, limit=100): del session, principal self.source_requests.append(request) resolutions = (self.resolution, self.resolution)[offset : offset + limit] return ContactPointSourcePreview( contract_version=CONTACT_POINT_CONTRACT_VERSION, source=DistributionSourceReference( provider="addresses", resource_type="address_list", resource_id="list-1", ), request=request, resolutions=resolutions, total_count=2, usable_count=len(resolutions), excluded_count=len(resolutions), offset=offset, limit=limit, has_more=offset + len(resolutions) < 2, source_revision="list-revision-1", source_fingerprint="list-fingerprint-1", generated_at=datetime(2026, 1, 1, tzinfo=UTC), provenance={"bounded": True}, ) def freeze_source(self, session, principal, *, request): raise AssertionError("Distribution Lists freezes its complete expansion snapshot.") def get_snapshot(self, session, principal, *, snapshot_id): del session, principal, snapshot_id return None 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 test_address_contact_resolution_is_purpose_aware_and_frozen(self) -> None: contact_points = _ContactPoints() registry = _Registry( {"addresses.contact_point_resolution": contact_points} ) with self.database.session() as session: item, _ = create_distribution_list( session, principal(), DistributionListCreateRequest.model_validate( { "name": "Official postal recipients", "entries": [ { "entry_key": "ada", "kind": "address_contact", "purpose": "official_notice", "requested_channels": ["postal"], "source": { "provider": "addresses", "resource_type": "contact", "resource_id": "contact-1", "revision": "contact-revision-1", }, "configuration": { "address_purpose": "official", "fallback_rule": "none", "locale": "de-DE", "postal_format": "international", }, } ], } ), ) result = expand_distribution_list( session, principal(), registry=registry, request=DistributionExpansionRequest( list_id=item.id, effective_at=datetime(2026, 1, 1, tzinfo=UTC), requested_channels=("postal",), freeze=True, idempotency_key="official-postal-1", ), ) session.commit() self.assertEqual(1, len(result.recipients)) recipient = result.recipients[0] self.assertEqual("contact-1", recipient.contact_id) self.assertEqual("postal-1", recipient.channels[0].contact_point_id) self.assertIn("Main Street 1", recipient.channels[0].target) self.assertEqual("suppressed", recipient.channels[1].status) self.assertEqual( "addresses.channel.opted_out", recipient.channels[1].reason_code, ) self.assertEqual("official_notice", contact_points.resolve_request.purpose) self.assertEqual("official", contact_points.resolve_request.address_purpose) self.assertEqual("international", contact_points.resolve_request.postal_format) self.assertEqual( "1.0", recipient.provenance["contact_point_resolution"]["contract_version"], ) snapshot = session.get(DistributionListSnapshot, result.snapshot_id) self.assertIn("Main Street 1", snapshot.recipients[0]["channels"][0]["target"]) self.assertEqual( "postal-1", snapshot.recipients[0]["channels"][0]["contact_point_id"], ) self.assertEqual( "facts-fingerprint-1", snapshot.provider_evidence[0]["actual_fingerprint"], ) def test_address_list_resolution_is_bounded_and_channel_neutral(self) -> None: contact_points = _ContactPoints() registry = _Registry( {"addresses.contact_point_resolution": contact_points} ) with self.database.session() as session: item, _ = create_distribution_list( session, principal(), DistributionListCreateRequest.model_validate( { "name": "Address source", "entries": [ { "entry_key": "address-list", "kind": "address_list", "source": { "provider": "addresses", "resource_type": "address_list", "resource_id": "addresses:address_list:list-1", }, } ], } ), ) result = expand_distribution_list( session, principal(), registry=registry, request=DistributionExpansionRequest( list_id=item.id, effective_at=datetime(2026, 1, 1, tzinfo=UTC), requested_channels=("postal",), limits=DistributionExpansionLimits(max_provider_results=1), ), ) self.assertEqual(1, len(result.recipients)) self.assertTrue(result.truncated) self.assertEqual( "addresses:address_list:list-1", contact_points.source_requests[0].source_id, ) self.assertEqual(("postal",), contact_points.source_requests[0].requested_channels) self.assertEqual("list-revision-1", result.provider_evidence[0].actual_revision) self.assertIn( "provider.result_limit", {item.code for item in result.diagnostics}, ) 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()