"""Source-of-Streams reclassification macro tests (IN-17361; originally IN-17362).

The `source_of_streams_reclassification` macros reclassify the entire
"Listener's Collection" bucket as Active, at read time, for every store whose By
Source breakdown has a Collection bucket — Spotify (286), Apple Music (1),
Amazon Music (187), LINE Japan (1503), SoundCloud (1505) and VKontakte (1405) —
mirroring the equivalent source_of_streams UDF change (dbt-analytics#1749)
without a historical backfill. The reclassification is gated by the
`insights_line_soundcloud_collection_as_active` feature flag (name historical),
threaded into the render context as `line_soundcloud_collection_as_active_enabled`
and read by the macros (which every SOS template imports `with context`). These
tests assert:

1. With the flag on, the macros render the exact store-conditional CASE
   expressions, and every SOS SQL template wires both macros in and renders.
2. With the flag off (or absent), the macros emit the raw streams_active /
   streams_collection columns — the original, un-reclassified behaviour — so
   toggling the flag off is a clean no-op.
"""

import pytest

from analytics.connectors import snowflake as sf
from analytics.connectors.snowflake import get_template_engine
from tests.unit.queries.test_macros import strip_sql

SQL_DIR = sf.BASE_DIR.joinpath("analytics/queries/sql")

# Rendered CASE expressions the macros must emit (default `s.` alias prefix).
ACTIVE_CASE = (
    "CASE WHEN s.store_id IN (1, 187, 286, 1405, 1503, 1505) "
    "THEN s.streams_active + s.streams_collection "
    "ELSE s.streams_active END"
)
COLLECTION_CASE = (
    "CASE WHEN s.store_id IN (1, 187, 286, 1405, 1503, 1505) THEN 0 ELSE s.streams_collection END"
)

# Feature-flag context variable (insights_line_soundcloud_collection_as_active)
# that gates the reclassification; True turns it on, False/absent is a no-op.
FLAG = "line_soundcloud_collection_as_active_enabled"


def _render(template_str, params):
    """Render a raw template string through the production jinjasql engine."""
    query = get_template_engine().prepare_query(template_str, params)[0]
    return strip_sql(query)


def _render_file(rel_path, params):
    text = SQL_DIR.joinpath(rel_path).read_text(encoding="utf-8")
    return _render(text, params)


# --- 1. flag ON: macro renders the exact CASE expressions ----------------- #


def test_reclassified_active_default_prefix():
    sql = _render(
        "{% from 'source_of_streams_reclassification.sql' import"
        " reclassified_active with context %}"
        "SELECT {{ reclassified_active() | sqlsafe }}",
        {FLAG: True},
    )
    assert ACTIVE_CASE in sql


def test_reclassified_collection_default_prefix():
    sql = _render(
        "{% from 'source_of_streams_reclassification.sql' import"
        " reclassified_collection with context %}"
        "SELECT {{ reclassified_collection() | sqlsafe }}",
        {FLAG: True},
    )
    assert COLLECTION_CASE in sql


def test_macros_accept_empty_prefix():
    """A call site with no table alias renders bare column references."""
    sql = _render(
        "{% from 'source_of_streams_reclassification.sql' import"
        " reclassified_active, reclassified_collection with context %}"
        "SELECT {{ reclassified_active('') | sqlsafe }},"
        " {{ reclassified_collection('') | sqlsafe }}",
        {FLAG: True},
    )
    assert (
        "CASE WHEN store_id IN (1, 187, 286, 1405, 1503, 1505) "
        "THEN streams_active + streams_collection ELSE streams_active END"
    ) in sql
    assert (
        "CASE WHEN store_id IN (1, 187, 286, 1405, 1503, 1505) THEN 0 ELSE streams_collection END" in sql
    )


# --- 1b. flag OFF/absent: macro emits the raw columns, no reclassification - #


@pytest.mark.parametrize("flag_params", [{FLAG: False}, {}])
def test_reclassified_active_disabled_emits_raw_column(flag_params):
    sql = _render(
        "{% from 'source_of_streams_reclassification.sql' import"
        " reclassified_active with context %}"
        "SELECT {{ reclassified_active() | sqlsafe }}",
        flag_params,
    )
    assert "s.streams_active" in sql
    assert "CASE WHEN" not in sql


@pytest.mark.parametrize("flag_params", [{FLAG: False}, {}])
def test_reclassified_collection_disabled_emits_raw_column(flag_params):
    sql = _render(
        "{% from 'source_of_streams_reclassification.sql' import"
        " reclassified_collection with context %}"
        "SELECT {{ reclassified_collection() | sqlsafe }}",
        flag_params,
    )
    assert "s.streams_collection" in sql
    assert "CASE WHEN" not in sql


def test_macros_empty_prefix_disabled_emits_bare_columns():
    """Flag off with no table alias renders bare, un-prefixed raw columns."""
    sql = _render(
        "{% from 'source_of_streams_reclassification.sql' import"
        " reclassified_active, reclassified_collection with context %}"
        "SELECT {{ reclassified_active('') | sqlsafe }},"
        " {{ reclassified_collection('') | sqlsafe }}",
        {FLAG: False},
    )
    assert "SELECT streams_active, streams_collection" in sql
    assert "CASE WHEN" not in sql


# --- 2. every SOS template wires the macros in and still renders ---------- #

PERMS = dict(
    permission_label_ids=[],
    permission_artist_ids=[],
    permission_subaccount_ids=[],
    permission_label_participant_ids=[],
    permission_feed_ids=[],
    transfer_product_ownership_enabled=False,
)
DATES = dict(start_date="2026-06-08", end_date="2026-06-12")
STORES = [1, 1503, 1505]
DISTS = ["theorchard"]

SOS_TEMPLATES = {
    "sound_recording/streams_breakdown.sql": {
        **PERMS,
        **DATES,
        "isrc": "ISRC",
        "store_ids": STORES,
        "distributors": DISTS,
        "country_ids": [],
        "all_time": False,
    },
    "sound_recording/streams_time_series.sql": {
        **PERMS,
        **DATES,
        "type": "TRACK_STREAMS_BY_SOS",
        "track_table": "V_STREAMS_BY_TRACK_FEED_DISTRIBUTOR_DAILY",
        "isrc": "ISRC",
        "store_ids": STORES,
        "distributors": DISTS,
        "countries": [],
        "resolution": None,
    },
    "sound_recording/summary.sql": {
        **PERMS,
        **DATES,
        "type": "SOS",
        "aggregation_field": "isrc",
        "downloads_table": "V_DOWNLOADS_BY_TRACK_FEED_DISTRIBUTOR_DAILY",
        "streams_table": "V_STREAMS_BY_TRACK_FEED_DISTRIBUTOR_DAILY",
        "isrc": "ISRC",
        "store_ids": STORES,
        "distributors": DISTS,
        "countries": [],
        "ids": [],
    },
    "participant/summary_by_sos.sql": {
        **PERMS,
        **DATES,
        "is_detailed": False,
        "global_participant_id": "GPID",
        "store_ids": STORES,
        "distributors": DISTS,
        "country_ids": [],
    },
    "participant/timeseries_streams_by_sos.sql": {
        **PERMS,
        **DATES,
        "is_detailed": False,
        "global_participant_id": "GPID",
        "store_ids": STORES,
        "distributors": DISTS,
        "country_ids": [],
        "resolution": None,
        "ids": [],
    },
    "account/streams_time_series.sql": {
        **PERMS,
        **DATES,
        "type": "ACCOUNT_STREAMS_BY_SOS",
        "account_table": "V_STREAMS_BY_LABEL_TRACK_FEED_DISTRIBUTOR_DAILY",
        "account_id": 123,
        "account_type": "vendor",
        "cross_attribution_pairs": [],
        "fin_label_ids": [],
        "upper_profit_center_ids": [],
        "countries": [],
        "store_ids": STORES,
        "distributors": DISTS,
        "resolution": None,
    },
    "account/summary.sql": {
        **PERMS,
        **DATES,
        "type": "SOS",
        "aggregation_field": "isrc",
        "streams_table": "V_STREAMS_BY_LABEL_TRACK_FEED_DISTRIBUTOR_DAILY",
        "downloads_table": "V_DOWNLOADS_BY_LABEL_TRACK_FEED_DISTRIBUTOR_DAILY",
        "account_id": 123,
        "account_type": "vendor",
        "cross_attribution_pairs": [],
        "fin_label_ids": [],
        "upper_profit_center_ids": [],
        "countries": [],
        "store_ids": STORES,
        "distributors": DISTS,
        "ids": [],
    },
    "product/streams_time_series.sql": {
        **PERMS,
        **DATES,
        "type": "PRODUCT_STREAMS_BY_SOS",
        "product_table": "V_STREAMS_BY_PRODUCT_TRACK_FEED_DISTRIBUTOR_DAILY",
        "product_id": 5,
        "store_ids": STORES,
        "distributors": DISTS,
        "countries": [],
        "multi_product": False,
        "resolution": None,
    },
    "product/summary.sql": {
        **PERMS,
        **DATES,
        "type": "SOS",
        "aggregation_field": "product_id",
        "streams_table": "V_STREAMS_BY_PRODUCT_TRACK_FEED_DISTRIBUTOR_DAILY",
        "downloads_table": "V_DOWNLOADS_BY_PRODUCT_TRACK_FEED_DISTRIBUTOR_DAILY",
        "product_id": 5,
        "store_ids": STORES,
        "distributors": DISTS,
        "countries": [],
        "multi_product": False,
        "ids": [],
    },
    "streams/aggregated_streams_by_sos.sql": {
        **PERMS,
        "isrc": "ISRC",
        "max_available_date": "2026-06-12",
        "days_back": 28,
        "store_ids": STORES,
        "countries": [],
        "is_feed_data_available": False,
    },
}


@pytest.mark.parametrize("rel_path,params", list(SOS_TEMPLATES.items()))
def test_template_applies_reclassification_when_enabled(rel_path, params):
    """Flag on: each SOS template renders and reclassifies active + collection."""
    sql = _render_file(rel_path, {**params, FLAG: True})
    assert ACTIVE_CASE in sql, f"active reclassification missing in {rel_path}"
    assert COLLECTION_CASE in sql, f"collection reclassification missing in {rel_path}"


@pytest.mark.parametrize("rel_path,params", list(SOS_TEMPLATES.items()))
def test_template_no_reclassification_when_disabled(rel_path, params):
    """Flag off: each SOS template still renders but emits the raw SOS columns,
    i.e. the original pre-IN-17362 behaviour (no store-conditional CASE)."""
    sql = _render_file(rel_path, {**params, FLAG: False})
    assert ACTIVE_CASE not in sql, f"unexpected active reclassification in {rel_path}"
    assert (
        COLLECTION_CASE not in sql
    ), f"unexpected collection reclassification in {rel_path}"


# --- 3. All Stores (empty store_ids) and multi-store combinations ---------- #
#
# The Product page "By Source" PoT is requested as
#   GET /product/<id>/timeseries?type=PRODUCT_STREAMS_BY_SOS&ids=<bucket>
# where "All Stores" omits store_ids entirely (empty list) and a store selection
# passes repeated store_ids params (e.g. store_ids=286&store_ids=1&store_ids=453).
# The reclassification is a per-row CASE on store_id applied inside every SUM,
# and each SOS template's store filter is conditional, so the all-stores
# aggregate and any combination reclassify with no separate code path.

ALL_STORES = []
COMBO_WITH_YOUTUBE = [286, 1, 453]  # Spotify + Apple + YouTube (453 has no collection)
PRIMARY_TRIO = [286, 1, 187]  # Spotify + Apple + Amazon


@pytest.mark.parametrize("store_ids", [ALL_STORES, COMBO_WITH_YOUTUBE, PRIMARY_TRIO])
@pytest.mark.parametrize(
    "rel_path",
    [
        "product/streams_time_series.sql",  # the real Product "By Source" PoT endpoint
        "product/summary.sql",
        "streams/aggregated_streams_by_sos.sql",
    ],
)
def test_reclassification_covers_all_stores_and_combinations(rel_path, store_ids):
    """Flag on: the aggregate reclassifies whether store_ids is empty (All
    Stores) or an explicit multi-store combination — the per-row CASE is summed
    across whatever stores match, so no separate all-stores path is needed."""
    params = {**SOS_TEMPLATES[rel_path], "store_ids": store_ids}
    sql = _render_file(rel_path, {**params, FLAG: True})
    assert ACTIVE_CASE in sql, f"active reclassification missing: {rel_path} {store_ids}"
    assert (
        COLLECTION_CASE in sql
    ), f"collection reclassification missing: {rel_path} {store_ids}"