"""Transfer-ownership tests for /participant//* endpoints. Global participant 611a71d6-b9a9-4598-a83f-6aed1bf73904 may appear on products beyond 5244974, so most participant endpoints aggregate well beyond product 5244974 and cannot isolate its transfer with a query param. Two endpoints *are* precise time-slice probes — both isolate product 5244974 and assert against conftest._VISIBILITY exactly like test_product.py, scoped by the viewer's permission window: * `/participant//timeseries` — `type=TRACK_STREAMS_BY_PRODUCT&ids=5244974` isolates the product server-side. * `/participant//summary` — has no `ids` param, so `type=PRODUCT` is requested with a high `limit` and product 5244974's row is selected from the per-product list client-side. The rest — aggregated-streams, track-streams-all, track-streams-by-store, demographics and /participant-metrics — are whole-catalogue endpoints. They are tested with the FF-pair (the FF-ON profile and its same-access FF-OFF twin, see conftest.PROFILE_PAIRS): with the flag on the viewer must see at least as much as legacy (the v2 path never drops data), and exactly the same for the employee, who sees current attribution under both flags. """ from __future__ import annotations import pytest from analytics import config from tests.integration.endpoints.conftest import PAGINATION, assert_endpoint from tests.integration.transfer_ownership.conftest import ( GLOBAL_PARTICIPANT_ID, PROBES, PRODUCT_ID, TRANSFER_PROFILE_PAIRS, TRANSFER_PROFILES, WINDOW_PROBES, assert_ff_monotonic, assert_ff_noop, assert_visibility, find_row, profile_key_for, sum_metric, ) # aggregated-streams / participant timeseries share this object shape. _AGGREGATED_KEYS = ( "all_other_rollup", "all_other_timeseries", "topn_timeseries", "topn_rollup", "total", ) def _participant_url( template: str, *, params: str = "", with_base: bool = False ) -> str: """Substitute the global participant id; optionally prefix BASE_URL.""" path = template.replace("", GLOBAL_PARTICIPANT_ID) url = f"{config.BASE_URL}{path}" if with_base else path return f"{url}?{params}" if params else url def _demographics_total(payload: dict) -> float: """Sum every age + gender bucket count in a demographics payload.""" demographics = payload.get("demographics") or {} buckets = { **(demographics.get("age") or {}), **(demographics.get("gender") or {}), } return sum((value or 0) for value in buckets.values()) def _store_streams_total(payload: dict) -> float: """Sum streams across every store's daily series in a by-store payload.""" return sum( sum_metric(store.get("items"), "streams") for store in (payload.get("stores") or []) ) class TestParticipantSummary: """/participant//summary — DAILY grain, v2 permissions_filter. The endpoint takes no per-product `ids` param, but `type=PRODUCT` returns one row per product. Requesting the participant's whole catalogue (`limit` set high enough to defeat pagination) and selecting product 5244974's row client-side isolates the transferred product — a precise time-slice probe: product 5244974's streams scoped by the viewer's permission window. """ @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_product_row(self, hdrs, probe): """Product 5244974's per-product summary row, time-sliced per profile.""" url = _participant_url( config.PARTICIPANT_SUMMARY_PATH, params=( f"type=PRODUCT&{PROBES[probe]}" "&order_by=streams&order_dir=desc&limit=100000" ), with_base=True, ) payload = assert_endpoint(url, headers=hdrs) # No `ids` param on this endpoint — pick product 5244974's row out of # the per-product list. The high `limit` returns the participant's # whole catalogue, so an absent row means the product was time-sliced # out, not paginated away. rows = [item for item in payload["items"] if str(item.get("id")) == PRODUCT_ID] assert_visibility( {"items": rows}, profile_key=profile_key_for(hdrs), probe=probe, ) class TestParticipantTimeseries: """/participant//timeseries — DAILY grain, v2 permissions_filter. `type=TRACK_STREAMS_BY_PRODUCT&ids=5244974` isolates the transferred product server-side, so this is a precise time-slice probe: product 5244974's track streams scoped by the viewer's permission window. """ @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_streams(self, hdrs, probe): """Product 5244974 track-stream timeseries per profile x window probe.""" url = _participant_url( config.PARTICIPANT_TIMESERIES_PATH, params=f"type=TRACK_STREAMS_BY_PRODUCT&ids={PRODUCT_ID}&{PROBES[probe]}", with_base=True, ) payload = assert_endpoint(url, headers=hdrs) assert_visibility( payload, profile_key=profile_key_for(hdrs), probe=probe, metric_key="value", ) class TestParticipantAggregatedStreams: """/participant//aggregated-streams — ROLLUP, recent window. Whole-catalogue ROLLUP aggregate; no per-product oracle, so the FF-pair differential is the guard. FF-ON reads the participant `_PRODUCT_TRANSFER` rollup through the v2 permissions_filter; FF-OFF reads the legacy rollup through dim_release. The summed catalogue total must be monotonic under the flag and flag-invariant for the employee. """ @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS) @pytest.mark.parametrize("dimension", ["store", "country", "sos"]) def test_dimension(self, pair, dimension): """Aggregated-streams windowed total is monotonic / flag-neutral.""" url = _participant_url( config.PARTICIPANT_AGGREGATED_STREAMS_URL, params=f"dimension={dimension}&days_back=28", ) on = assert_endpoint(url, headers=pair.ff_on, items_key=None) off = assert_endpoint(url, headers=pair.ff_off, items_key=None) for key in _AGGREGATED_KEYS: assert key in on and key in off, f"missing aggregated-streams field: {key}" on_total = (on.get("total") or {}).get("streams_28_days") or 0 off_total = (off.get("total") or {}).get("streams_28_days") or 0 label = f"{pair.key} /participant/aggregated-streams {dimension}" assert_ff_monotonic(on_total, off_total, label=label) if pair.key == "employee": assert_ff_noop(on_total, off_total, label=label) class TestParticipantTrackStreams: """/participant//track-streams-all — DAILY whole-catalogue timeseries. Sums every product the participant appears on, so it cannot isolate product 5244974. The FF-pair guards the v2 permissions_filter path: the summed daily streams must be monotonic under the flag and flag-invariant for the employee, at every window probe. """ @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_track_streams(self, pair, probe): """Per-day catalogue streams are monotonic / flag-neutral per probe.""" url = _participant_url( config.PARTICIPANT_TRACK_STREAMS_ALL_URL, params=f"{PROBES[probe]}&{PAGINATION}", ) on = assert_endpoint(url, headers=pair.ff_on) off = assert_endpoint(url, headers=pair.ff_off) on_total = sum_metric(on["items"], "streams") off_total = sum_metric(off["items"], "streams") label = f"{pair.key} /participant/track-streams-all probe={probe}" assert_ff_monotonic(on_total, off_total, label=label) if pair.key == "employee": assert_ff_noop(on_total, off_total, label=label) class TestParticipantTrackStreamsByStore: """/participant//track-streams-by-store — DAILY per-store timeseries. Whole-catalogue, so the FF-pair guards the v2 permissions_filter path: streams summed across every store must be monotonic under the flag and flag-invariant for the employee, at every window probe. """ @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_track_streams_by_store(self, pair, probe): """Per-store catalogue streams are monotonic / flag-neutral per probe.""" url = _participant_url( config.PARTICIPANT_TRACK_STREAMS_STORE_URL, params=PROBES[probe] ) on = assert_endpoint(url, headers=pair.ff_on, items_key=None) off = assert_endpoint(url, headers=pair.ff_off, items_key=None) assert "stores" in on and isinstance(on["stores"], list) assert "stores" in off and isinstance(off["stores"], list) on_total = _store_streams_total(on) off_total = _store_streams_total(off) label = f"{pair.key} /participant/track-streams-by-store probe={probe}" assert_ff_monotonic(on_total, off_total, label=label) if pair.key == "employee": assert_ff_noop(on_total, off_total, label=label) class TestParticipantDemographics: """/participant//demographics — DAILY, v2 permissions_filter. Whole-catalogue demographics; the FF-pair guards the v2 path — the summed age/gender counts must be monotonic under the flag and flag-invariant for the employee. """ @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS) def test_default(self, pair): """Demographic counts are monotonic / flag-neutral across the FF-pair.""" url = _participant_url(config.PARTICIPANT_DEMOGRAPHICS_PATH, with_base=True) on = assert_endpoint(url, headers=pair.ff_on, items_key=None) off = assert_endpoint(url, headers=pair.ff_off, items_key=None) on_total = _demographics_total(on) off_total = _demographics_total(off) label = f"{pair.key} /participant/demographics" assert_ff_monotonic(on_total, off_total, label=label) if pair.key == "employee": assert_ff_noop(on_total, off_total, label=label) class TestParticipantMetrics: """/participant-metrics — ROLLUP listing, keyed by participant. The scenario participant's blended `streams_all_time` row spans every product it appears on, so it cannot isolate product 5244974. The FF-pair still guards the v2 ROLLUP path: the participant's all-time streams can only grow under the flag (a former owner regains transferred-away products) and are exactly flag-invariant for the employee. """ @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS) def test_participant_row(self, pair): """The participant's all-time streams are monotonic / flag-neutral.""" url = ( f"{config.PARTICIPANT_METRICS_URL}" f"?global_participant_ids={GLOBAL_PARTICIPANT_ID}&limit=50" ) on = assert_endpoint(url, headers=pair.ff_on, items_key=None) off = assert_endpoint(url, headers=pair.ff_off, items_key=None) for payload in (on, off): assert "metrics" in payload and isinstance(payload["metrics"], list) assert "total_participants" in payload on_row = find_row(on["metrics"], "id", GLOBAL_PARTICIPANT_ID) off_row = find_row(off["metrics"], "id", GLOBAL_PARTICIPANT_ID) on_total = (on_row or {}).get("streams_all_time") or 0 off_total = (off_row or {}).get("streams_all_time") or 0 label = f"{pair.key} /participant-metrics streams_all_time" assert_ff_monotonic(on_total, off_total, label=label) if pair.key == "employee": assert_ff_noop(on_total, off_total, label=label)