diff --git a/src/adcp/reporting/subscriptions/facade.py b/src/adcp/reporting/subscriptions/facade.py index 5c504c8a1..66e57ac7c 100644 --- a/src/adcp/reporting/subscriptions/facade.py +++ b/src/adcp/reporting/subscriptions/facade.py @@ -217,15 +217,41 @@ async def guard_sync_principal( ) +#: Subscription arrays that are ``uniqueItems`` sets: a seller may return them +#: in any order, such as sorted, so the readback compares them as sets. +_SET_ARRAYS = frozenset({"account_ids", "exclude_account_ids", "operator_domains", "dimensions"}) +#: Arrays of objects whose order carries no meaning: ``destination_overrides`` +#: select disjoint accounts, so each sent entry matches any stored one. +_UNORDERED_ARRAYS = frozenset({"destination_overrides"}) + + def differing_path(sent: Any, stored: Any, path: str = "") -> str | None: """The first JSON Pointer at which ``stored`` lacks or changes a value the buyer sent. - Fields the seller adds (defaults it fills) are ignored; arrays must match - element for element. + Fields the seller adds (defaults it fills) are ignored. Set-valued arrays + (``account_ids``, ``exclude_account_ids``, ``operator_domains``, + ``dimensions``) and ``destination_overrides`` compare without regard to + order; every other array must match element for element. """ if isinstance(sent, list): if not isinstance(stored, list) or len(stored) != len(sent): return path or "/" + key = path.rsplit("/", 1)[-1] + if key in _SET_ARRAYS and all(isinstance(item, (str, int)) for item in sent): + if sorted(map(str, sent)) != sorted(map(str, stored)): + return path or "/" + return None + if key in _UNORDERED_ARRAYS: + unmatched = list(stored) + for index, item in enumerate(sent): + match = next( + (i for i, other in enumerate(unmatched) if differing_path(item, other) is None), + None, + ) + if match is None: + return f"{path}/{index}" + unmatched.pop(match) + return None for index, item in enumerate(sent): found = differing_path(item, stored[index], f"{path}/{index}") if found is not None: diff --git a/tests/conformance/reporting/test_reporting_row_storage_object_pg.py b/tests/conformance/reporting/test_reporting_row_storage_object_pg.py index aa2410045..f4cab4017 100644 --- a/tests/conformance/reporting/test_reporting_row_storage_object_pg.py +++ b/tests/conformance/reporting/test_reporting_row_storage_object_pg.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -442,13 +443,18 @@ async def test_the_producer_stages_object_uploads_outside_its_publication_transa from tests.test_reporting_settling import _capabilities, _harness, _revisions async with isolated_reporting_pool() as pool: - producer, store, _, _ = await _harness( + producer, store, _, clock = await _harness( _capabilities(restatement_window="P3D"), store_factory=lambda clock: PgReportingLedgerStore( pool=pool, clock=clock, row_storage=options(tmp_path) ), ) assert isinstance(producer, ReportingProducer) + # The row upload checks the worker lease against the database's + # clock_timestamp(), so the lease the injected clock stamps must lie in + # the real future: run the turn at the wall clock, not the harness's + # fixed instant, which is already in the past. + clock[0] = datetime.now(timezone.utc) turn = await producer.run_worker() assert turn.revisions_committed (revision,) = await _revisions(store) diff --git a/tests/test_reporting_subscription_facade.py b/tests/test_reporting_subscription_facade.py index c5787c009..4a03c3711 100644 --- a/tests/test_reporting_subscription_facade.py +++ b/tests/test_reporting_subscription_facade.py @@ -272,6 +272,68 @@ async def test_a_readback_mismatch_is_terminal() -> None: assert mismatch.value.recovery == "terminal" and mismatch.value.readback is not None +class Sorting(Client): + """A conformant seller that returns every set-valued array in sorted order.""" + + async def sync_principal(self, request): + result = await super().sync_principal(request) + for state in result.data.result.configuration.reporting_subscriptions: + configuration = state.configuration + dumped = configuration.model_dump(mode="json", exclude_none=True) + for name in ("exclude_account_ids", "operator_domains"): + if name in dumped: + dumped[name] = sorted(dumped[name]) + if "account_ids" in dumped.get("scope", {}): + dumped["scope"]["account_ids"] = sorted(dumped["scope"]["account_ids"]) + if "dimensions" in dumped.get("selector", {}): + dumped["selector"]["dimensions"] = sorted(dumped["selector"]["dimensions"]) + state.configuration = type(configuration).model_validate(dumped) + return result + + +async def test_the_readback_compares_set_valued_arrays_without_order() -> None: + reporting = ReportingClient(Sorting(Seller())) + selector = {**SUBSCRIPTION["selector"], "dimensions": ["package", "creative"]} + result = await reporting.subscribe( + _subscription( + scope={"account_ids": ["acct-2", "acct-1"]}, + operator_domains=["z.example", "a.example"], + selector=selector, + ) + ) + assert result.action == "created" + + +def test_differing_path_orders_only_lists_that_carry_order() -> None: + from adcp.reporting.subscriptions.facade import differing_path + + assert ( + differing_path({"operator_domains": ["b", "a"]}, {"operator_domains": ["a", "b"]}) is None + ) + assert differing_path({"operator_domains": ["b", "a"]}, {"operator_domains": ["a", "c"]}) == ( + "/operator_domains" + ) + overrides = [ + {"destination_id": "x", "account_ids": ["1", "2"]}, + {"destination_id": "y", "operator_domains": ["o.example"]}, + ] + stored = [ + {"destination_id": "y", "operator_domains": ["o.example"], "added": True}, + {"destination_id": "x", "account_ids": ["2", "1"]}, + ] + assert ( + differing_path({"destination_overrides": overrides}, {"destination_overrides": stored}) + is None + ) + stored[0]["destination_id"] = "z" + assert ( + differing_path({"destination_overrides": overrides}, {"destination_overrides": stored}) + == "/destination_overrides/1" + ) + # Any other array still compares element for element. + assert differing_path({"other": ["a", "b"]}, {"other": ["b", "a"]}) == "/other/0" + + class Roster(Client): def __init__(self, seller: Seller, accounts: list[dict[str, Any]], offerings=None) -> None: super().__init__(seller)