"""Participation-axis transfer-ownership tests — the two ``permission_label_participant_ids`` branches of ``_permissions_filter_v2``. When product 5244974 is moved from vendor 81790 to 797716, the move also moves the artist's *participation* (lambda-kinesis-to-neo4j ``move_project.py``): it re-points the ``PARTICIPATED_IN`` edge onto a freshly-created destination-vendor ``LabelParticipant`` and hard-deletes the originating LP's edge. The two LP branches of the v2 filter are therefore exercised by two DIFFERENT artist profiles, both reaching global participant 611a71d6 (which represents both LPs): * **destination_artist** (LP 1567012701053, vendor 797716) — the *current* participant. Its row is live in ``LABEL_PARTICIPANT_PARTICIPATED_IN_ORCHARD_PRODUCT``, so the current LP arm matches and it sees the product's full lifetime in every window. The flag is a no-op (legacy reads the same current participation row, untime-sliced). * **originating_artist** (LP 1702386309417, vendor 81790) — the *former* participant. The move deleted its ``PARTICIPATED_IN`` row, so both the current arm and the legacy path match nothing; only the ``USED_TO_PARTICIPATE_IN`` shadow re-grants access, frozen to the former-owner window by the historical arm. Same shape as the ``ala_vendor`` former owner, via the participation axis: W1 data, **no W2 leak**, full history. The originating grant depends on the ``permissions_filter.sql`` subaccount normalization (the USED_TO row carries ``subaccount_id = 0`` for a vendor-level participant while ``PRODUCT_OWNERSHIP_ACCESS`` stores ``NULL``; the macro collapses both to "no subaccount"). Until that macro deploys to a given env the FF-ON originating profile reads 0 there — deploy-lag, the same class as README §10 cluster A. The suite passes against a local dev server, which runs the patched macro over ``FACTS.QA``. """ from __future__ import annotations from datetime import date, timedelta import pytest from analytics import config from tests.integration.endpoints.conftest import PAGINATION, assert_endpoint from tests.integration.transfer_ownership.conftest import ( ALL_PROBES, ARTIST_PROFILE_PAIRS, DESTINATION_ARTIST, GLOBAL_PARTICIPANT_ID, ORIGINATING_ARTIST, ORIGINATING_BOUNDARY, PROBES, PRODUCT_ID, WINDOW_PROBES, assert_ff_grants_transfer, assert_ff_monotonic, assert_ff_noop, assert_visibility, find_row, sum_metric, ) def _participant_url( template: str, *, params: str = "", with_base: bool = False ) -> str: """Substitute the global participant id; optionally prefix BASE_URL. The ``*_PATH`` config constants are bare paths (pass ``with_base=True``); the ``*_URL`` constants already include BASE_URL (the default). """ 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 _product_timeseries_total(headers: dict, probe_params: str) -> tuple[dict, float]: """Product 5244974's track-stream timeseries, isolated server-side. Returns the raw payload and the summed daily ``value`` over the window. """ url = _participant_url( config.PARTICIPANT_TIMESERIES_PATH, params=f"type=TRACK_STREAMS_BY_PRODUCT&ids={PRODUCT_ID}&{probe_params}", with_base=True, ) payload = assert_endpoint(url, headers=headers) return payload, sum_metric(payload.get("items"), "value") class TestParticipationTimeseries: """``/participant//timeseries`` — ``type=TRACK_STREAMS_BY_PRODUCT``. ``ids=5244974`` isolates the moved product server-side, so this is a precise time-slice probe of each artist's permission window. The destination (current participant) sees every window; the originating (USED_TO former participant) sees W1 (early/mid) and the spanning ``full`` window, but nothing in W2 (``late``) — no forward leak past 2026-04-30. """ @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS) @pytest.mark.parametrize("probe", ALL_PROBES) def test_product_streams(self, pair, probe): payload, _ = _product_timeseries_total(pair.ff_on, PROBES[probe]) assert_visibility( payload, profile_key=pair.key, probe=probe, metric_key="value" ) class TestParticipationSummary: """``/participant//summary`` — ``type=PRODUCT``, per-product rows. No per-product ``ids`` param, so request the whole catalogue (high ``limit``) and select product 5244974's row client-side — an absent row at a W2 probe means the product was time-sliced out, not paginated away. """ @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_product_row(self, pair, probe): 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=pair.ff_on) rows = [item for item in payload["items"] if str(item.get("id")) == PRODUCT_ID] assert_visibility({"items": rows}, profile_key=pair.key, probe=probe) class TestParticipationFlagDifferential: """FF-pair differential on the per-product timeseries probe. The flag is a differential oracle on the participation axis: * **originating_artist — grant.** FF-OFF runs the legacy LP arm, which reads the *current* ``PARTICIPATED_IN`` table; the move deleted the row, so it sees 0 (a former participant cannot see the moved-away product without the flag). FF-ON's USED_TO arm re-grants the W1 slice (> 0). Measured at a W1 probe, where the frozen window has data. * **destination_artist — no-op.** The current participation row is live under both flags (legacy untime-sliced, v2 current arm), so the product's figure is identical. Measured at the spanning ``full`` window. """ def test_originating_grant(self): _, on_total = _product_timeseries_total( ORIGINATING_ARTIST.ff_on, PROBES["early"] ) _, off_total = _product_timeseries_total( ORIGINATING_ARTIST.ff_off, PROBES["early"] ) assert_ff_grants_transfer( on_total, off_total, label=( f"originating_artist /participant/timeseries product={PRODUCT_ID} W1" ), ) def test_destination_noop(self): _, on_total = _product_timeseries_total( DESTINATION_ARTIST.ff_on, PROBES["full"] ) _, off_total = _product_timeseries_total( DESTINATION_ARTIST.ff_off, PROBES["full"] ) label = f"destination_artist /participant/timeseries product={PRODUCT_ID} full" assert_ff_monotonic(on_total, off_total, label=label) assert_ff_noop(on_total, off_total, label=label) # --------------------------------------------------------------------------- # Whole-catalogue participant endpoints — FF-pair monotonicity guards # --------------------------------------------------------------------------- # # Global participant 611a71d6 may appear on products beyond 5244974, so these # endpoints aggregate the participant's whole catalogue and cannot isolate the # transfer with a query param. The precise grant/no-op is already pinned by the # per-product probes above (and the product / sound-recording suites); here the # FF-pair is the monotonicity guard on the v2 participation path — FF-ON (the # participant `_PRODUCT_TRANSFER` rollup / DAILY participation filter) must see # at least as much as FF-OFF (legacy). # # Neither artist is full-access, so there is no strict no-op oracle on these # whole-catalogue aggregates: the legacy and `_PRODUCT_TRANSFER` participant # rollups are independent dbt builds whose drift can move the summed total — # exactly why test_participant.py asserts no-op only for the full-access # employee, never for the scoped ownership profiles. _AGGREGATED_KEYS = ( "all_other_rollup", "all_other_timeseries", "topn_timeseries", "topn_rollup", "total", ) 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 TestParticipationAggregatedStreams: """/participant//aggregated-streams — whole-catalogue ROLLUP, FF-pair.""" @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS) @pytest.mark.parametrize("dimension", ["store", "country", "sos"]) def test_dimension(self, pair, dimension): """Aggregated-streams windowed total is monotonic under the flag.""" 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 assert_ff_monotonic( on_total, off_total, label=f"{pair.key} /participant/aggregated-streams {dimension}", ) class TestParticipationTrackStreams: """/participant//track-streams-all — DAILY whole-catalogue timeseries.""" @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_track_streams(self, pair, probe): """Per-day catalogue streams are monotonic under the flag 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") assert_ff_monotonic( on_total, off_total, label=f"{pair.key} /participant/track-streams-all probe={probe}", ) class TestParticipationTrackStreamsByStore: """/participant//track-streams-by-store — DAILY per-store timeseries.""" @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_track_streams_by_store(self, pair, probe): """Per-store catalogue streams are monotonic under the flag 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) assert_ff_monotonic( _store_streams_total(on), _store_streams_total(off), label=f"{pair.key} /participant/track-streams-by-store probe={probe}", ) class TestParticipationDemographics: """/participant//demographics — DAILY whole-catalogue, FF-pair.""" @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS) def test_default(self, pair): """Demographic counts are monotonic under the flag.""" 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) assert_ff_monotonic( _demographics_total(on), _demographics_total(off), label=f"{pair.key} /participant/demographics", ) class TestParticipationMetrics: """/participant-metrics — ROLLUP listing keyed by participant, FF-pair. The 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 participation ROLLUP path: the participant's all-time streams can only grow under the flag (the originating artist regains its USED_TO slice). """ @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS) def test_participant_row(self, pair): """The participant's all-time streams are monotonic under the flag.""" 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) assert_ff_monotonic( (on_row or {}).get("streams_all_time") or 0, (off_row or {}).get("streams_all_time") or 0, label=f"{pair.key} /participant-metrics streams_all_time", ) def _day_after(date_str: str) -> str: """Return the calendar day after an ISO date string.""" y, m, d = (int(p) for p in date_str.split("-")) return (date(y, m, d) + timedelta(days=1)).isoformat() def _single_day_total(headers: dict, day: str) -> float: """Product 5244974's track streams on a single day, for an artist viewer.""" _, total = _product_timeseries_total(headers, f"start_date={day}&end_date={day}") return total class TestOriginatingArtistBoundary: """The USED_TO time-slice upper bound is inclusive at ``access_until_date``. Mirrors ``test_boundary.py`` for the participation axis. The originating artist sees its move-boundary day 2026-04-30 (~4,007 streams) and nothing the day after — pinning the boundary from both sides. Regression guard against a strict ``<`` (drops the former participant's last day) or a fuzzy ``<= cutoff + 1`` (leaks one post-move day into the frozen window). """ def test_cutoff_day_visible_to_originating(self): """Originating artist sees its access_until_date day — inclusive ``<=``.""" assert _single_day_total(ORIGINATING_ARTIST.ff_on, ORIGINATING_BOUNDARY) > 0 def test_day_after_cutoff_invisible_to_originating(self): """Originating artist sees nothing the day after its cutoff — no leak.""" day_after = _day_after(ORIGINATING_BOUNDARY) streams = _single_day_total(ORIGINATING_ARTIST.ff_on, day_after) assert streams == 0, ( f"originating artist leaked {streams} streams on day-after-cutoff " f"({day_after}) — USED_TO arm upper bound is wrong" ) def test_cutoff_day_visible_to_destination(self): """Control: the cutoff day has streams (destination, current participant).""" assert _single_day_total(DESTINATION_ARTIST.ff_on, ORIGINATING_BOUNDARY) > 0 def test_day_after_cutoff_visible_to_destination(self): """Control: 2026-05-01 (start of W2) is visible to the destination.""" assert ( _single_day_total( DESTINATION_ARTIST.ff_on, _day_after(ORIGINATING_BOUNDARY) ) > 0 )