"""Fixtures and helpers for the ows-playlist transfer-ownership suite.

Ported from `ows-analytics/tests/integration/transfer_ownership/conftest.py`,
then narrowed to the ows-playlist scope. Differences:

- Canonical ISRC is ``BX69Y2100018`` — the heaviest ISRC under product
  ``5244974`` (UPC ``198000137366``), with multi-window coverage on the
  canonical playlist.
- ``CANONICAL_STORE_PLAYLIST_ID`` / ``CANONICAL_STORE_ID`` pin a single
  playlist that carries multi-window stream data for the canonical ISRC.
- ``PROBES`` use ``start_date + days`` (ows-playlist date contract), not
  ``start_date + end_date``.
- ``ACCOUNT_WINDOW_PROFILE`` is dropped — ows-playlist has no
  ``/account/<id>/*`` analogues.
- ``assert_isolated_transfer`` / ``assert_ff_grants_transfer`` are dropped —
  the only grant path the suite reaches is on the v2 DAILY macro, which is
  precise-probe territory; the ROLLUP table-swap endpoints assert
  metric-neutrality via ``assert_ff_pair_equal``.

See README.md for the full scenario.
"""

from __future__ import annotations

import json
from typing import NamedTuple

import pytest

# ---------------------------------------------------------------------------
# Profiles  (all carry the insights_transfer_product_ownership flag in QA)
# ---------------------------------------------------------------------------

# (key, InsightsProfile id) — key is the stable string used by the visibility
# matrix and pytest param ids.
PROFILES = (
    ("employee", "474215"),
    ("ala_comunicacao", "9900038"),
    ("sued_nunes", "9900042"),
)


def _insights_headers(profile_id: str) -> dict:
    """Build the request header dict for an InsightsProfile id."""
    return {
        "Orchard-Profile-Id": profile_id,
        "Orchard-Profile-Type": "InsightsProfile",
    }


PROFILE_HEADERS = {key: _insights_headers(pid) for key, pid in PROFILES}

# Module-level aliases for direct use in tests.
EMPLOYEE = PROFILE_HEADERS["employee"]
ALA_COMUNICACAO = PROFILE_HEADERS["ala_comunicacao"]
SUED_NUNES = PROFILE_HEADERS["sued_nunes"]

# Parametrize axis: every profile, with the key as the pytest id.
TRANSFER_PROFILES = [
    pytest.param(headers, id=key) for key, headers in PROFILE_HEADERS.items()
]

# ---------------------------------------------------------------------------
# FF-OFF twins  (same account access, insights_transfer_product_ownership OFF)
# ---------------------------------------------------------------------------
#
# Each profile above has a flag-disabled twin with *identical* account access.
# A request made with an FF-OFF profile runs the production (legacy) code
# path: current-ownership only, via dim_release. Note that ``dim_release``
# for product 5244974 still attributes the product to its *original* vendor
# (label 81790, Alá Comunicação) — the 2026-04-30 transfer to vendor 797716
# is recorded only in ``PRODUCT_OWNERSHIP_ACCESS``. So under FF-OFF:
# ``ala_comunicacao`` sees everything (dim_release agrees they own it);
# ``sued_nunes`` sees nothing (dim_release has not caught up). This is the
# exact data hazard PORT-5 routes around.

PROFILES_FF_OFF = (
    ("employee", "1100"),
    ("ala_comunicacao", "9900039"),
    ("sued_nunes", "9900043"),
)

PROFILE_HEADERS_FF_OFF = {key: _insights_headers(pid) for key, pid in PROFILES_FF_OFF}


class ProfilePair(NamedTuple):
    """An FF-ON / FF-OFF profile twin sharing one account scope."""

    key: str  # stable profile key, e.g. "ala_comunicacao"
    ff_on: dict  # headers for the flag-enabled (v2) profile
    ff_off: dict  # headers for the flag-disabled (legacy) twin


# profile key -> ProfilePair. Both members share account access; only the
# insights_transfer_product_ownership flag differs.
PROFILE_PAIRS = {
    key: ProfilePair(key, PROFILE_HEADERS[key], PROFILE_HEADERS_FF_OFF[key])
    for key in PROFILE_HEADERS
}

# Parametrize axis for the FF-pair tests: one ProfilePair per profile.
TRANSFER_PROFILE_PAIRS = [
    pytest.param(pair, id=key) for key, pair in PROFILE_PAIRS.items()
]

# ---------------------------------------------------------------------------
# The transferred product
# ---------------------------------------------------------------------------

PRODUCT_ID = "5244974"
PRODUCT_NAME = "Travessia"
UPC = "198000137366"
ARTIST_ID = 3125999
GLOBAL_PARTICIPANT_ID = "611a71d6-b9a9-4598-a83f-6aed1bf73904"

# Canonical ISRC for the heavy parametrize axes. BX69Y2100018 is by far the
# heaviest ISRC under product 5244974 — 72,265 streams across 576 days on
# the canonical playlist, with usable coverage in both transfer windows.
ISRC = "BX69Y2100018"

# All 13 ISRCs under product 5244974. Every one of them has W1 coverage on
# (mix, 286), so the W1 sweep is a true 13-element axis. W2 coverage is
# uneven; the precise probes use the canonical ISRC.
PRODUCT_ISRCS = [f"BX69Y21000{n:02d}" for n in range(12, 25)]

# ---------------------------------------------------------------------------
# Canonical placement  (data probe: ows-playlist 2026-05-27)
# ---------------------------------------------------------------------------
#
# `mix` on store 286 is a Spotify algorithmic playlist that streams
# BX69Y2100018 across both transfer windows in FACTS.QA:
#
#   early probe (2024-10-01 → 2026-04-30): 69,549 streams  (Alá tenure)
#   late  probe (2026-05-01 → 2026-05-26):  2,716 streams  (Sued tenure)
#
# It is the playlist with the strongest multi-window coverage on the
# canonical ISRC, so it anchors every precise DAILY probe.

CANONICAL_STORE_PLAYLIST_ID = "mix"
CANONICAL_STORE_ID = "286"

# ---------------------------------------------------------------------------
# Ownership timeline  (FACTS.QA.PRODUCT_OWNERSHIP_ACCESS, product 5244974)
# ---------------------------------------------------------------------------
#
#   ... 2026-04-30 | 2026-05-01 ... now
#   W1 Alá/81790   | W2 Sued/797716 (current)
#
# One transfer, two windows. A former-owner cell is therefore only
# reachable for ``ala_comunicacao`` on the ``late`` probe.

# W1 — original (former) owner: Alá Comunicação e Cultura, vendor 81790.
ALA_COMUNICACAO_VENDOR_ID = 81790
ALA_COMUNICACAO_ACCESS_UNTIL = "2026-04-30"  # inclusive last day of tenure

# W2 — current owner: Sued Nunes Produções LTDA, vendor 797716.
SUED_NUNES_VENDOR_ID = 797716

# ---------------------------------------------------------------------------
# Date-range probes
# ---------------------------------------------------------------------------
#
# ows-playlist DAILY handlers take `start_date + days`, not `start_date +
# end_date`. Each probe sits *inside* one window so the visibility matrix is
# unambiguous.

PROBES = {
    "early": "start_date=2024-10-01&days=577",  # 2024-10-01 → 2026-04-30 (W1)
    "late": "start_date=2026-05-01&days=26",  # 2026-05-01 → 2026-05-26 (W2)
    "full": "start_date=2014-01-01&days=4530",  # spans both windows
}

# Single-window probes — the visibility matrix is unambiguous for these.
WINDOW_PROBES = [pytest.param(p, id=p) for p in ("early", "late")]
# Every probe, including the all-spanning `full`.
ALL_PROBES = [pytest.param(p, id=p) for p in PROBES]

# ---------------------------------------------------------------------------
# Visibility matrix
# ---------------------------------------------------------------------------
#
# "data"  -> endpoint returns rows whose metric sums > 0
# "empty" -> endpoint returns no rows, or rows whose metric sums to 0
#
# Each owner sees [beginning-of-time, their cutoff]; windows are nested. The
# matrix is *viewer-scoped* (no account-scope variant — ows-playlist has no
# /account/<id>/* analogue of the ows-analytics matrix).

_VISIBILITY: dict[tuple[str, str], str] = {
    # employee — full access, sees every window
    ("employee", "early"): "data",
    ("employee", "late"): "data",
    ("employee", "full"): "data",
    # sued_nunes — current owner, access_until_date IS NULL, sees everything
    ("sued_nunes", "early"): "data",
    ("sued_nunes", "late"): "data",
    ("sued_nunes", "full"): "data",
    # ala_comunicacao — former owner, cutoff 2026-04-30: sees W1, not W2
    ("ala_comunicacao", "early"): "data",
    ("ala_comunicacao", "late"): "empty",
    ("ala_comunicacao", "full"): "data",
}

# Slack absorbing run-to-run DIM_RELEASE CDC drift between paired requests
# (catalogue-wide aggregates can shift by single-digit percent across a
# rebuild — see ows-analytics project_full_set_aggregation_dim_release_cdc).
FF_DRIFT = 0.02

# ---------------------------------------------------------------------------
# Lookups
# ---------------------------------------------------------------------------


def profile_key_for(headers: dict) -> str:
    """Reverse-lookup the short profile key from a header dict.

    Resolves both the FF-ON profiles and their FF-OFF twins to the same key.
    """
    target = headers["Orchard-Profile-Id"]
    for table in (PROFILE_HEADERS, PROFILE_HEADERS_FF_OFF):
        for key, hdrs in table.items():
            if hdrs["Orchard-Profile-Id"] == target:
                return key
    raise KeyError(f"Unknown profile id in headers: {target}")


def expected_visibility(profile_key: str, probe: str) -> str:
    """Return 'data' or 'empty' for a (profile, probe) pair."""
    return _VISIBILITY[(profile_key, probe)]


# ---------------------------------------------------------------------------
# Assertion helpers
# ---------------------------------------------------------------------------


def _sum_metric(items: list, metric_key: str) -> float:
    """Sum `metric_key` across items, treating missing/None as 0."""
    return sum((item.get(metric_key) or 0) for item in items)


def assert_visibility(
    payload: dict,
    *,
    profile_key: str,
    probe: str,
    items_key: str | None = "items",
    metric_key: str = "streams",
) -> None:
    """Assert the payload matches the visibility matrix for (profile, probe).

    - "data"  -> items non-empty AND the metric sums > 0
    - "empty" -> items empty OR the metric sums to 0

    For payloads that are a single object rather than an items list, pass
    `items_key=None`; the metric is then read from the top-level payload.
    """
    expectation = expected_visibility(profile_key, probe)
    label = f"profile={profile_key} probe={probe}"
    if items_key:
        items = payload.get(items_key) or []
        total = _sum_metric(items, metric_key)
        if expectation == "empty":
            assert total == 0, (
                f"{label}: expected empty but {len(items)} item(s) "
                f"totalled {total} on '{metric_key}'"
            )
        else:
            assert items, f"{label}: expected data but items were empty"
            assert (
                total > 0
            ), f"{label}: expected data but items totalled 0 on '{metric_key}'"
    else:
        value = payload.get(metric_key) or 0
        if expectation == "empty":
            assert value == 0, f"{label}: expected empty but {metric_key}={value}"
        else:
            assert value > 0, f"{label}: expected data but {metric_key}={value}"


# ---------------------------------------------------------------------------
# FF-pair differential helpers (used by the ROLLUP table-swap tests)
# ---------------------------------------------------------------------------


def assert_ff_monotonic(ff_on: float, ff_off: float, *, label: str) -> None:
    """FF-ON must see at least as much as FF-OFF.

    v2 only ever grants a former owner back its historical slice — it never
    removes data. A violation means the flag dropped a viewer's data.
    """
    assert ff_on >= ff_off * (1 - FF_DRIFT), (
        f"{label}: the flag REMOVED data — FF-ON={ff_on} < FF-OFF={ff_off}. "
        f"v2 must never show a viewer less than legacy does."
    )


def assert_ff_noop(
    ff_on: float, ff_off: float, *, label: str, tolerance: float = FF_DRIFT
) -> None:
    """FF-ON and FF-OFF must agree — the flag is a no-op for this figure."""
    margin = max(abs(ff_off), 1.0) * tolerance
    assert abs(ff_on - ff_off) <= margin, (
        f"{label}: the flag changed a figure it must not — "
        f"FF-ON={ff_on}, FF-OFF={ff_off} (allowed ±{margin})."
    )


def _normalize(obj):
    """Recursively make a JSON payload order-insensitive for comparison."""
    if isinstance(obj, dict):
        return {key: _normalize(value) for key, value in obj.items()}
    if isinstance(obj, list):
        normalized = [_normalize(item) for item in obj]
        return sorted(
            normalized, key=lambda x: json.dumps(x, sort_keys=True, default=str)
        )
    return obj


def assert_ff_pair_equal(ff_on: dict, ff_off: dict, *, label: str) -> None:
    """Assert the FF-ON and FF-OFF payloads are identical (order-insensitive).

    For endpoints where the flag is, by design, not an ownership-scoping
    change — it only swaps a fact table to its ``_PRODUCT_TRANSFER`` Pattern B
    drop-in — every viewer must get a byte-equal response with the flag on
    and off. Use for the ROLLUP table-swap tests where dbt has already
    guaranteed equivalence.
    """
    assert _normalize(ff_on) == _normalize(ff_off), (
        f"{label}: the flag changed the response — FF-ON and FF-OFF differ, "
        f"but this endpoint's v2 path must be metric-neutral."
    )