973 lines
37 KiB
Python
973 lines
37 KiB
Python
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.core.identity import (
|
|
CAPABILITY_IDENTITY_DIRECTORY,
|
|
IdentityAccountLinkRef,
|
|
IdentityRef,
|
|
)
|
|
from govoplan_core.core.idm import (
|
|
CAPABILITY_IDM_RELATIONSHIPS,
|
|
IdentityRelationshipDecisionRef,
|
|
IdentityRelationshipRef,
|
|
TypedGroupMembershipResolutionRef,
|
|
TypedGroupRef,
|
|
)
|
|
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.catalogue import provider_catalogue
|
|
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 _IdentityDirectory:
|
|
def __init__(self) -> None:
|
|
self.identity = IdentityRef(
|
|
id="identity-active",
|
|
display_name="Ada Example",
|
|
primary_account_id="account-active",
|
|
account_ids=("account-active",),
|
|
status="active",
|
|
)
|
|
|
|
def get_identity(self, identity_id: str):
|
|
return self.identity if identity_id == self.identity.id else None
|
|
|
|
def identity_for_account(self, account_id: str):
|
|
return self.identity if account_id == "account-active" else None
|
|
|
|
def identities_for_accounts(self, account_ids):
|
|
return (self.identity,) if "account-active" in account_ids else ()
|
|
|
|
def accounts_for_identity(self, identity_id: str):
|
|
if identity_id != self.identity.id:
|
|
return ()
|
|
return (
|
|
IdentityAccountLinkRef(
|
|
id="link-1",
|
|
identity_id=identity_id,
|
|
account_id="account-active",
|
|
is_primary=True,
|
|
),
|
|
)
|
|
|
|
|
|
class _RelationshipDirectory:
|
|
def __init__(self, effective_at: datetime) -> None:
|
|
self.group = TypedGroupRef(
|
|
id="group-1",
|
|
tenant_id="tenant-1",
|
|
key="permit-holder",
|
|
name="Permit holders",
|
|
group_type="business_status",
|
|
source_provider="ldap",
|
|
source_resource_type="group",
|
|
source_resource_id="cn=permit-holders,dc=example",
|
|
source_revision="directory-42",
|
|
properties={"classification": "resident"},
|
|
provenance={"sync_run_id": "sync-1"},
|
|
revision=3,
|
|
)
|
|
active = IdentityRelationshipRef(
|
|
id="relationship-active",
|
|
tenant_id="tenant-1",
|
|
relationship_kind="member",
|
|
subject_identity_id="identity-active",
|
|
target_group_id="group-1",
|
|
source_provider="ldap",
|
|
source_resource_type="membership",
|
|
source_resource_id="member:active",
|
|
source_revision="directory-42",
|
|
properties={"rank": 1},
|
|
provenance={"sync_run_id": "sync-1"},
|
|
revision=2,
|
|
)
|
|
expired = IdentityRelationshipRef(
|
|
id="relationship-expired",
|
|
tenant_id="tenant-1",
|
|
relationship_kind="member",
|
|
subject_identity_id="identity-expired",
|
|
target_group_id="group-1",
|
|
valid_until=effective_at,
|
|
source_provider="ldap",
|
|
source_revision="directory-41",
|
|
)
|
|
self.resolution = TypedGroupMembershipResolutionRef(
|
|
group=self.group,
|
|
effective_at=effective_at,
|
|
decisions=(
|
|
IdentityRelationshipDecisionRef(
|
|
relationship=active,
|
|
included=True,
|
|
code="relationship.effective",
|
|
explanation="The relationship is effective.",
|
|
identity_status="active",
|
|
),
|
|
IdentityRelationshipDecisionRef(
|
|
relationship=expired,
|
|
included=False,
|
|
code="relationship.expired",
|
|
explanation="The relationship has expired.",
|
|
identity_status="active",
|
|
),
|
|
),
|
|
)
|
|
|
|
def get_typed_group(self, group_id, *, tenant_id=None):
|
|
return self.group if group_id == self.group.id else None
|
|
|
|
def list_typed_groups(self, *, tenant_id, query=None, group_types=(), include_inactive=False, limit=100):
|
|
del group_types, include_inactive, limit
|
|
if tenant_id != "tenant-1" or (query and query.casefold() not in self.group.name.casefold()):
|
|
return ()
|
|
return (self.group,)
|
|
|
|
def identity_relationships_for_identity(self, identity_id, **kwargs):
|
|
return ()
|
|
|
|
def identity_relationships_for_identities(self, identity_ids, **kwargs):
|
|
return {identity_id: () for identity_id in identity_ids}
|
|
|
|
def identity_relationships_for_group(self, group_id, **kwargs):
|
|
return tuple(
|
|
item.relationship
|
|
for item in self.resolution.decisions
|
|
if item.included
|
|
)
|
|
|
|
def identity_relationships_for_groups(self, group_ids, **kwargs):
|
|
return {
|
|
group_id: self.identity_relationships_for_group(group_id, **kwargs)
|
|
for group_id in group_ids
|
|
}
|
|
|
|
def resolve_typed_group_memberships(self, group_ids, **kwargs):
|
|
return {
|
|
group_id: self.resolution
|
|
for group_id in group_ids
|
|
if group_id == self.group.id
|
|
}
|
|
|
|
|
|
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_idm_typed_group_expansion_preserves_decisions_and_provider_evidence(self) -> None:
|
|
effective_at = datetime(2026, 8, 2, 12, tzinfo=UTC)
|
|
relationships = _RelationshipDirectory(effective_at)
|
|
registry = _Registry(
|
|
{
|
|
CAPABILITY_IDM_RELATIONSHIPS: relationships,
|
|
CAPABILITY_IDENTITY_DIRECTORY: _IdentityDirectory(),
|
|
}
|
|
)
|
|
with self.database.session() as session:
|
|
item, _ = create_distribution_list(
|
|
session,
|
|
principal(),
|
|
DistributionListCreateRequest.model_validate(
|
|
{
|
|
"name": "Permit holders",
|
|
"entries": [
|
|
{
|
|
"entry_key": "permit-holders",
|
|
"kind": "idm_group",
|
|
"source": {
|
|
"provider": "idm",
|
|
"resource_type": "typed_group",
|
|
"resource_id": "group-1",
|
|
"revision": "directory-42",
|
|
},
|
|
}
|
|
],
|
|
}
|
|
),
|
|
)
|
|
session.flush()
|
|
result = expand_distribution_list(
|
|
session,
|
|
principal(),
|
|
registry=registry,
|
|
request=DistributionExpansionRequest(
|
|
list_id=item.id,
|
|
effective_at=effective_at,
|
|
),
|
|
)
|
|
catalogue = provider_catalogue(
|
|
session,
|
|
principal(),
|
|
registry=registry,
|
|
query="Permit",
|
|
)
|
|
|
|
self.assertEqual(("identity:identity-active",), tuple(row.recipient_key for row in result.recipients))
|
|
self.assertEqual("account-active", result.recipients[0].account_id)
|
|
self.assertEqual(
|
|
"relationship.effective",
|
|
result.recipients[0].explanations[0].code,
|
|
)
|
|
expired = next(
|
|
row for row in result.excluded if row.identity_id == "identity-expired"
|
|
)
|
|
self.assertEqual("relationship.expired", expired.explanations[0].code)
|
|
self.assertEqual(
|
|
"directory-42",
|
|
result.provider_evidence[0].actual_revision,
|
|
)
|
|
self.assertEqual(
|
|
"relationship-active",
|
|
result.recipients[0].provenance["relationship_id"],
|
|
)
|
|
self.assertIn("idm-group:group-1", {row.key for row in catalogue.items})
|
|
self.assertNotIn("idm", {row.provider for row in catalogue.unavailable_providers})
|
|
|
|
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()
|