"""Transfer-ownership FF on/off rendering tests.

Locks down the byte-level rendered SQL for the four §3 query classes
touched by the transfer-ownership port:

- §3a DAILY: placement_streams_time_series + total_vs_playlist_time_series
- §3b ROLLUP: bulk_playlist_analytics + sound_recording_top_placements

FF off (default `transfer_product_ownership_enabled=False`) must keep
the legacy `dim_track_clean_mv → dim_release` subquery shape and the
original FROM table name. FF on must emit the v2 EXISTS-on-POA
predicate aliased to `s.product_id` and (for ROLLUP queries) swap the
FROM to `_PRODUCT_TRANSFER`.
"""

import pytest

from playlist.queries.fetch_queries import (
    fetch_bulk_playlist_analytics,
    fetch_placement_streams_time_series,
    fetch_sound_recording_top_placements,
    fetch_total_vs_playlist_streams_time_series,
)

PERMS_LABEL = {"permission_label_ids": ["1", "2", "3"]}
PERMS_LP = {"permission_label_participant_ids": ["100"]}
PERMS_FULL = {
    "permission_label_ids": [],
    "permission_subaccount_ids": [],
    "permission_label_participant_ids": [],
}

PLAYLIST_PARAMS = {
    "isrc": "USACQ0500001",
    "store_id": "286",
    "store_playlist_id": "37i9dQZF1DX2apWzyECwyZ",
    "start_date": "2024-01-01",
    "end_date": "2024-03-01",
}


def _rendered(mock_execute_orm):
    return mock_execute_orm.call_args[0][0]


# ---------------------------------------------------------------------------
# §3a placement_streams_time_series
# ---------------------------------------------------------------------------


@pytest.mark.disable_mock_execute
def test_placement_streams_ff_off_legacy_shape(mock_execute_orm):
    fetch_placement_streams_time_series(
        {**PLAYLIST_PARAMS, "transfer_product_ownership_enabled": False},
        PERMS_LABEL,
    )
    sql = _rendered(mock_execute_orm)
    assert "dim_track_clean_mv" in sql
    assert "dtc.upc" in sql
    assert "product_ownership_access" not in sql


@pytest.mark.disable_mock_execute
def test_placement_streams_ff_on_v2_shape(mock_execute_orm):
    fetch_placement_streams_time_series(
        {**PLAYLIST_PARAMS, "transfer_product_ownership_enabled": True},
        PERMS_LABEL,
    )
    sql = _rendered(mock_execute_orm)
    assert "product_ownership_access" in sql
    assert "s.product_id" in sql
    assert "s.download_activity_date <= poa.access_until_date" in sql
    # legacy join shape dropped on the v2 path
    assert "dim_track_clean_mv" not in sql


@pytest.mark.disable_mock_execute
def test_placement_streams_ff_on_label_participant_uses_product_lookup(
    mock_execute_orm,
):
    fetch_placement_streams_time_series(
        {**PLAYLIST_PARAMS, "transfer_product_ownership_enabled": True},
        PERMS_LP,
    )
    sql = _rendered(mock_execute_orm)
    assert "label_participant_participated_in_orchard_product" in sql
    assert "lp.product_id = s.product_id" in sql


# ---------------------------------------------------------------------------
# §3a total_vs_playlist_streams_time_series
# ---------------------------------------------------------------------------


@pytest.mark.disable_mock_execute
def test_total_vs_playlist_ff_off_legacy_shape(mock_execute_orm):
    fetch_total_vs_playlist_streams_time_series(
        {
            "isrc": "USACQ0500001",
            "start_date": "2024-01-01",
            "end_date": "2024-03-01",
            "transfer_product_ownership_enabled": False,
        },
        PERMS_LABEL,
    )
    sql = _rendered(mock_execute_orm)
    assert "dim_track_clean_mv" in sql
    assert "product_ownership_access" not in sql


@pytest.mark.disable_mock_execute
def test_total_vs_playlist_ff_on_v2_shape_both_ctes(mock_execute_orm):
    fetch_total_vs_playlist_streams_time_series(
        {
            "isrc": "USACQ0500001",
            "start_date": "2024-01-01",
            "end_date": "2024-03-01",
            "transfer_product_ownership_enabled": True,
        },
        PERMS_LABEL,
    )
    sql = _rendered(mock_execute_orm)
    # both CTEs (playlist_streams_timeseries + streams_timeseries) must
    # carry the EXISTS POA predicate — the macro emits the shape twice.
    assert sql.count("product_ownership_access poa") >= 2
    assert "s.product_id" in sql
    assert "dim_track_clean_mv" not in sql


# ---------------------------------------------------------------------------
# §3b bulk_playlist_analytics (table swap only, no permissions_filter)
# ---------------------------------------------------------------------------

BULK_PARAMS = {
    "playlists": [
        {
            "store_playlist_id": "37i9dQZF1DX2apWzyECwyZ",
            "store_id": "286",
            "storefront": None,
            "row_number": 0,
        }
    ],
    "storefront_defined": False,
}


@pytest.mark.disable_mock_execute
def test_bulk_playlist_analytics_ff_off_keeps_legacy_table(mock_execute_orm):
    fetch_bulk_playlist_analytics(
        {**BULK_PARAMS, "transfer_product_ownership_enabled": False},
        PERMS_LABEL,
    )
    sql = _rendered(mock_execute_orm)
    assert "streams_by_playlist_feed_rollup" in sql.lower()
    assert "_product_transfer" not in sql.lower()


@pytest.mark.disable_mock_execute
def test_bulk_playlist_analytics_ff_on_swaps_to_product_transfer(mock_execute_orm):
    fetch_bulk_playlist_analytics(
        {**BULK_PARAMS, "transfer_product_ownership_enabled": True},
        PERMS_LABEL,
    )
    sql = _rendered(mock_execute_orm)
    assert "streams_by_playlist_feed_rollup_product_transfer" in sql.lower()


# ---------------------------------------------------------------------------
# §3b sound_recording_top_placements (table swap on playlist_rollup CTE only)
# ---------------------------------------------------------------------------

TOP_PARAMS = {
    "isrc": "USACQ0500001",
    "limit": 12,
    "offset": 0,
    "sort_key": "streams_all_time",
    "sort_direction": "DESC",
}


@pytest.mark.disable_mock_execute
def test_sound_recording_top_placements_ff_off_keeps_legacy_table(mock_execute_orm):
    fetch_sound_recording_top_placements(
        {**TOP_PARAMS, "transfer_product_ownership_enabled": False},
        PERMS_LABEL,
    )
    sql = _rendered(mock_execute_orm)
    assert "streams_by_playlist_rollup" in sql.lower()
    assert "_product_transfer" not in sql.lower()


@pytest.mark.disable_mock_execute
def test_sound_recording_top_placements_ff_on_swaps_to_product_transfer(
    mock_execute_orm,
):
    fetch_sound_recording_top_placements(
        {**TOP_PARAMS, "transfer_product_ownership_enabled": True},
        PERMS_LABEL,
    )
    sql = _rendered(mock_execute_orm)
    assert "streams_by_playlist_rollup_product_transfer" in sql.lower()


# ---------------------------------------------------------------------------
# Macro regression: EXISTS predicates always alias-qualified with `s`
# ---------------------------------------------------------------------------


@pytest.mark.disable_mock_execute
@pytest.mark.parametrize(
    "perms",
    [
        {"permission_label_ids": ["1"]},
        {"permission_subaccount_ids": ["7123"]},
        {"permission_label_participant_ids": ["42"]},
        PERMS_FULL,
    ],
)
def test_v2_macro_alias_qualifies_product_id(mock_execute_orm, perms):
    """The v2 EXISTS predicate must reference `s.product_id` — bare
    `product_id` silently resolves to the inner POA column and leaks
    access. Regression guard: every `product_id` mention must be
    alias-qualified (`s.`, `poa.`, or `lp.`)."""
    import re

    fetch_placement_streams_time_series(
        {**PLAYLIST_PARAMS, "transfer_product_ownership_enabled": True},
        perms,
    )
    sql = _rendered(mock_execute_orm)
    # find all `product_id` mentions without a `.` directly preceding —
    # those are bare and leak access.
    bare = re.findall(r"(?<![\.\w])product_id\b", sql)
    assert not bare, f"unqualified product_id leaked in v2 output:\n{sql}"