"""Transfer-ownership tests for /account//* endpoints. Two tiers: * **Precise time-slice probes** — `/account//timeseries` and `/account//summary`. `type=ACCOUNT_STREAMS_BY_PRODUCT` / `type=PRODUCT` with `ids=5244974` isolate the transferred product server-side, so the response is product 5244974's streams *as scoped to the queried account*. account_scope_filter time-slices by the *account's* own ownership window, not the viewer's, so the visibility expectation is keyed by the account (see conftest.ACCOUNT_WINDOW_PROFILE) — employee querying account 81790 sees exactly the W1 slice that the originating profile does. * **FF-pair differential** — `/account//products`, `/top-content` and the video endpoints. These aggregate an account's entire catalogue with no per-product selector, so a single request cannot isolate product 5244974's transfer. Each test instead fires the request with the FF-ON profile and its FF-OFF twin (same account access, flag off — see conftest.PROFILE_PAIRS): `/account//products` returns a per-product row list, so product 5244974's own row is picked out and checked exactly; top-content and video assert the v2 path neither leaks a transferred-away product nor drops an account's data. All three account-scoped endpoints time-slice by the *account's* window, so the differential is keyed by ACCOUNT_WINDOW_PROFILE, not the viewer. """ 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 ( ACCOUNT_WINDOW_PROFILE, CURRENT_VIEW, DESTINATION_VENDOR_ID, FORMER_OWNERS, ORIGINATING_VENDOR_ID, PROBES, PRODUCT_ID, PRODUCT_ISRCS, PROFILE_HEADERS, TRANSFER_PROFILE_PAIRS, TRANSFER_PROFILES, WINDOW_PROBES, assert_ff_monotonic, assert_ff_noop, assert_visibility, find_row, profile_can_reach_account, profile_key_for, rows_for, sum_metric, ) # (account_id, account_type) — the account_type query param is mandatory. ACCOUNTS = [ pytest.param((ORIGINATING_VENDOR_ID, "vendor"), id="ala_81790"), pytest.param((DESTINATION_VENDOR_ID, "vendor"), id="sued_797716"), ] # Video endpoints are vendor-only — both transfer accounts are vendors in # this scenario, so they coincide with ACCOUNTS. VIDEO_ACCOUNTS = [ pytest.param((ORIGINATING_VENDOR_ID, "vendor"), id="ala_81790"), pytest.param((DESTINATION_VENDOR_ID, "vendor"), id="sued_797716"), ] # Wide date range so every (real, populated) account returns rows. _FULL = PROBES["full"] def _account_url(template: str, account_id: int, account_type: str, params: str) -> str: """Build an /account//* URL with the mandatory account_type param.""" base = template.replace("", str(account_id)) return f"{base}?account_type={account_type}&{params}" def _skip_if_unreachable(hdrs, account_id: int) -> str: """Return the profile key, skipping the test if it cannot reach the account.""" profile_key = profile_key_for(hdrs) if not profile_can_reach_account(profile_key, account_id): pytest.skip(f"{profile_key} has no access to account {account_id}") return profile_key def _skip_pair_if_unreachable(pair, account_id: int) -> None: """Skip the test if the profile pair cannot reach the account. Both members of a pair share account access, so one check covers both. """ if not profile_can_reach_account(pair.key, account_id): pytest.skip(f"{pair.key} has no access to account {account_id}") class TestAccountTimeseries: """/account//timeseries — v2 account_scope_filter, DAILY grain. `type=ACCOUNT_STREAMS_BY_PRODUCT&ids=5244974` isolates the transferred product server-side (the SQL applies `id IN (...)` post-aggregation), so the response is product 5244974's daily streams scoped to the queried account. account_scope_filter time-slices by the account's ownership window, so visibility is per account (ACCOUNT_WINDOW_PROFILE), not per viewer — both the employee and the owning profile see the same slice. """ @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES) @pytest.mark.parametrize("account", ACCOUNTS) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_streams(self, hdrs, account, probe): """Product 5244974 stream timeseries per account x window probe.""" account_id, account_type = account _skip_if_unreachable(hdrs, account_id) url = _account_url( config.ACCOUNT_METRICS_TIME_SERIES_URL, account_id, account_type, params=f"type=ACCOUNT_STREAMS_BY_PRODUCT&ids={PRODUCT_ID}&{PROBES[probe]}", ) payload = assert_endpoint(url, headers=hdrs) assert_visibility( payload, profile_key=ACCOUNT_WINDOW_PROFILE[account_id], probe=probe, metric_key="value", ) class TestAccountSummary: """/account//summary — v2 account_scope_filter, DAILY grain. `type=PRODUCT&ids=5244974` isolates the transferred product server-side, so this is a precise time-slice probe rather than a whole-catalogue aggregate. Visibility is keyed by the queried account's ownership window. """ @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES) @pytest.mark.parametrize("account", ACCOUNTS) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_streams(self, hdrs, account, probe): """Product 5244974 summary per account x window probe.""" account_id, account_type = account _skip_if_unreachable(hdrs, account_id) url = _account_url( config.ACCOUNT_METRICS_SUMMARY_URL, account_id, account_type, params=f"type=PRODUCT&ids={PRODUCT_ID}&{PROBES[probe]}&{PAGINATION}", ) payload = assert_endpoint(url, headers=hdrs) assert_visibility( payload, profile_key=ACCOUNT_WINDOW_PROFILE[account_id], probe=probe, ) class TestAccountProducts: """/account//products — v2 account_scope_filter against the rollup table. The endpoint returns one bare `{"product_id": ...}` row per product — it carries no metric column (the streams figure used to order the rollup is dropped in logic.account_products.get_products), so the only transfer signal it exposes is *presence*: does product 5244974 appear in the account's product list? `order_by=streams_all_time` routes to the fixed-period rollup query, which reads the `*_PRODUCT_TRANSFER` table when the flag is on. Visibility is keyed by (account, viewer): * A scoped viewer (the account's own profile) sees what the account's window dictates — a former-owner account has no 5244974 row without the flag (legacy current-ownership hides the transferred-away product) and one with it; a current-owner account lists it both ways. * The employee is governed by README §5: the v2 path reads `is_current = TRUE` rows only, so the employee sees product 5244974 solely under its current-owner account (797716) — both flags. On a former-owner account the employee's `is_current` permission clamp intersects emptily with the account-scope, so 5244974 is absent under both flags. """ @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS) @pytest.mark.parametrize("account", ACCOUNTS) def test_transferred_product_row(self, pair, account): """Product 5244974's presence obeys the (account, viewer) role.""" account_id, account_type = account _skip_pair_if_unreachable(pair, account_id) url = _account_url( config.ACCOUNT_PRODUCTS_URL, account_id, account_type, # Large limit so product 5244974's row is never paginated away. params="order_by=streams_all_time&order_dir=desc&limit=100000", ) on = assert_endpoint(url, headers=pair.ff_on) off = assert_endpoint(url, headers=pair.ff_off) if on["items"]: assert on["items"][0].get("product_id"), "product row missing product_id" on_present = find_row(on["items"], "product_id", PRODUCT_ID) is not None off_present = find_row(off["items"], "product_id", PRODUCT_ID) is not None label = f"{pair.key} /account/{account_id}/products product {PRODUCT_ID}" if pair.key == "employee": # README §5: the employee reads `is_current = TRUE` rows only, so # product 5244974 appears under its current-owner account (797716) # — both flags — and is absent under a former-owner account both # flags (the account scope and the `is_current` clamp intersect # emptily). The account window does *not* override the viewer # clamp for the employee. if account_id == DESTINATION_VENDOR_ID: assert on_present and off_present, ( f"{label}: employee on the current-owner account must " f"list product {PRODUCT_ID} with the flag on and off " f"(FF-ON={on_present}, FF-OFF={off_present})." ) else: assert not on_present and not off_present, ( f"{label}: employee must not see product {PRODUCT_ID} under " f"a former-owner account (per README §5, the employee " f"reads is_current rows only) — FF-ON={on_present}, " f"FF-OFF={off_present}." ) return role_key = ACCOUNT_WINDOW_PROFILE[account_id] if role_key in CURRENT_VIEW: assert on_present == off_present, ( f"{label}: the flag changed whether the current-owner account " f"lists the product (FF-ON={on_present}, FF-OFF={off_present}) " f"— v2 resolves to current ownership here, so it must not." ) assert on_present, f"{label}: current-owner account does not list it" elif role_key in FORMER_OWNERS: assert not off_present, ( f"{label}: legacy listed a transferred-away product for a " f"former-owner account — FF-OFF must not surface product " f"{PRODUCT_ID}." ) assert on_present, ( f"{label}: v2 did not grant the former-owner account its " f"product {PRODUCT_ID} row — FF-ON must surface it." ) class TestAccountTopContent: """/account//top-content — v2 account_scope_filter, ROLLUP grain. top-content is a top-N leaderboard, so product 5244974 need not rank into it — but one fact is unconditional: a *former-owner account* cannot surface the transferred-away product, or its ISRCs, without the flag. The differential asserts that no-leak on the FF-OFF response, and that a current-owner account treats the product identically with the flag on and off. """ TOP_KEYS = ( "topn_artists", "topn_products", "topn_songs", "topn_countries", "topn_stores", ) @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS) @pytest.mark.parametrize("account", ACCOUNTS) def test_transfer_scoping(self, pair, account): """A former-owner account never leaks product 5244974 into top-content.""" account_id, account_type = account _skip_pair_if_unreachable(pair, account_id) url = _account_url( config.ACCOUNT_TOP_CONTENT_URL, account_id, account_type, params="top_size=1000", ) 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 self.TOP_KEYS: assert key in on and key in off, f"missing top-content list: {key}" role = ACCOUNT_WINDOW_PROFILE[account_id] label = f"{pair.key} /account/{account_id}/top-content" on_has = find_row(on["topn_products"], "product_id", PRODUCT_ID) is not None off_has = find_row(off["topn_products"], "product_id", PRODUCT_ID) is not None if role in FORMER_OWNERS: assert not off_has, ( f"{label}: legacy leaked transferred-away product {PRODUCT_ID} " f"into a former-owner account's top-content" ) leaked_songs = rows_for(off["topn_songs"], "isrc", PRODUCT_ISRCS) assert not leaked_songs, ( f"{label}: legacy leaked {len(leaked_songs)} of product " f"{PRODUCT_ID}'s ISRCs into a former-owner account's " f"top-content" ) elif role in CURRENT_VIEW: assert on_has == off_has, ( f"{label}: the flag changed whether product {PRODUCT_ID} " f"ranks into a current-owner account's top-content " f"(FF-ON={on_has}, FF-OFF={off_has})" ) # View metrics whose presence varies by endpoint/type — summed defensively. _VIDEO_METRICS = ("value", "views") def _video_total(items: list[dict]) -> float: """Sum every view-count metric across video summary / timeseries items.""" return sum(sum_metric(items, metric) for metric in _VIDEO_METRICS) class TestAccountVideo: """/account//video/{summary,timeseries} — v2 fact-table swap. Account video swaps to the `_PRODUCT_TRANSFER` fact table via the video-permissions macro (not the three core v2 macros). These accounts are music labels that carry little or no video traffic, so the FF-pair is a regression guard: the table swap must not drop an account's video data (FF-ON >= FF-OFF) and must be exactly neutral for a current-owner account. """ @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS) @pytest.mark.parametrize("account", VIDEO_ACCOUNTS) def test_summary(self, pair, account): """Account video summary is flag-neutral / monotonic per account window.""" account_id, account_type = account _skip_pair_if_unreachable(pair, account_id) url = _account_url( config.ACCOUNT_VIDEO_METRICS_SUMMARY_URL, account_id, account_type, params="type=TOTAL&views_type=ALL", ) on = assert_endpoint(url, headers=pair.ff_on) off = assert_endpoint(url, headers=pair.ff_off) self._assert_neutral(pair, account_id, on["items"], off["items"], "summary") @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS) @pytest.mark.parametrize("account", VIDEO_ACCOUNTS) def test_timeseries(self, pair, account): """Account video timeseries is flag-neutral / monotonic per account window.""" account_id, account_type = account _skip_pair_if_unreachable(pair, account_id) url = _account_url( config.ACCOUNT_VIDEO_METRICS_TIME_SERIES_URL, account_id, account_type, params=f"type=TOTAL&views_type=ALL&{_FULL}", ) on = assert_endpoint(url, headers=pair.ff_on) off = assert_endpoint(url, headers=pair.ff_off) self._assert_neutral(pair, account_id, on["items"], off["items"], "timeseries") @staticmethod def _assert_neutral(pair, account_id, on_items, off_items, kind): on_total = _video_total(on_items) off_total = _video_total(off_items) label = f"{pair.key} /account/{account_id}/video/{kind}" assert_ff_monotonic(on_total, off_total, label=label) if ACCOUNT_WINDOW_PROFILE[account_id] in CURRENT_VIEW: assert_ff_noop(on_total, off_total, label=label) class TestAccountScopeViewerDecoupling: """Account-scope is keyed by the account's window, not the viewer's. account_scope_filter restricts /account//* to the account's catalog and time-slices by the account's ownership window — not the viewer's. Any two viewers who can both reach the same account must see exactly the same slice for the same query: the employee viewing account 81790 sees what ala_vendor sees of 81790, not "all of 81790 plus W2" just because the employee is full-access elsewhere. Catches a regression where account-scope leaks into viewer-scope (or vice versa) — e.g. a JOIN that ORs the two filters instead of ANDing them. The existing per-account tests check shape ("data" vs "empty"); these pin the numbers cross-viewer. """ @pytest.mark.parametrize("account", ACCOUNTS) def test_summary_equal_across_viewers(self, account): """employee vs owning-profile see equal totals on /account//summary.""" account_id, account_type = account owning_key = ACCOUNT_WINDOW_PROFILE[account_id] url = _account_url( config.ACCOUNT_METRICS_SUMMARY_URL, account_id, account_type, params=f"type=PRODUCT&ids={PRODUCT_ID}&{_FULL}&{PAGINATION}", ) emp_payload = assert_endpoint(url, headers=PROFILE_HEADERS["employee"]) own_payload = assert_endpoint(url, headers=PROFILE_HEADERS[owning_key]) emp_total = sum_metric(emp_payload["items"], "streams") own_total = sum_metric(own_payload["items"], "streams") if not (emp_total and own_total): pytest.skip(f"no data for account {account_id} — QA data/POA not ready") assert abs(emp_total - own_total) <= 0.02 * own_total, ( f"employee ({emp_total}) and {owning_key} ({own_total}) disagree " f"on /account/{account_id}/summary?ids={PRODUCT_ID} — account-" f"scope is leaking into viewer-scope" ) @pytest.mark.parametrize("account", ACCOUNTS) def test_timeseries_equal_across_viewers(self, account): """employee vs owning-profile see equal totals on /account//timeseries.""" account_id, account_type = account owning_key = ACCOUNT_WINDOW_PROFILE[account_id] url = _account_url( config.ACCOUNT_METRICS_TIME_SERIES_URL, account_id, account_type, params=f"type=ACCOUNT_STREAMS_BY_PRODUCT&ids={PRODUCT_ID}&{_FULL}", ) emp_payload = assert_endpoint(url, headers=PROFILE_HEADERS["employee"]) own_payload = assert_endpoint(url, headers=PROFILE_HEADERS[owning_key]) emp_total = sum_metric(emp_payload["items"], "value") own_total = sum_metric(own_payload["items"], "value") if not (emp_total and own_total): pytest.skip(f"no data for account {account_id} — QA data/POA not ready") assert abs(emp_total - own_total) <= 0.02 * own_total, ( f"employee ({emp_total}) and {owning_key} ({own_total}) disagree " f"on /account/{account_id}/timeseries?ids={PRODUCT_ID} — account-" f"scope is leaking into viewer-scope" )