"""Published max-available-date join tests (GO-4885).

Three templates join DATA_AVAILABILITY_GET_MAX_AVAILABLE_STREAMING_STORES_DATE
at query time. That view sits on the _DBT staging table, so mid-pipeline it
reports the new max date ~40 minutes before the streams/metrics tables swap —
every store looks lagged and the 1D windows NULL out (the GO-4885 "By Sources
shows –" bug). Behind the `insights_published_max_available_date` feature flag
(threaded into the render context as `published_max_available_date_enabled`)
the templates read the view's _PUBLISHED twin instead, which dbt re-points only
after the full upgrade_main_pipeline swap batch. These tests assert:

1. With the flag on, each template joins the _PUBLISHED view.
2. With the flag off, each template renders byte-identically to the flag being
   absent entirely — i.e. the original join, so the flag is a clean no-op.
3. The flag survives the query-class schema load (field on
   QueryWithPermissionsSchema) end to end, defaulting to off.
4. With the flag on, the account template additionally caps its fact scan at
   the anchor date (the SOS templates already carry that cap unconditionally),
   so the mid-swap-batch window can't leak the next day into its 7/28-day sums.
"""

import pytest

from analytics.connectors import snowflake as sf
from analytics.connectors.snowflake import get_template_engine
from analytics.queries.account import AccountProductsByDownloadsFixedPeriod
from analytics.queries.streams import (
    GlobalParticipantAggregatedStreamsBySOS,
    ProductAggregatedStreamsBySOS,
    SoundRecordingAggregatedStreamsBySOS,
    SoundRecordingAggregatedStreamsBySOSV2,
)
from tests.unit.queries.test_macros import strip_sql

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

ORIGINAL_VIEW = "DATA_AVAILABILITY_GET_MAX_AVAILABLE_STREAMING_STORES_DATE"
PUBLISHED_VIEW = "DATA_AVAILABILITY_GET_MAX_AVAILABLE_STREAMING_STORES_DATE_PUBLISHED"

# Feature-flag context variable (insights_published_max_available_date).
FLAG = "published_max_available_date_enabled"

PERMS = dict(
    permission_label_ids=[],
    permission_artist_ids=[],
    permission_subaccount_ids=[],
    permission_label_participant_ids=[],
    permission_feed_ids=[],
    transfer_product_ownership_enabled=False,
)

# Every template that joins the max-available-date view at query time, with a
# minimal-but-complete param set to render it.
TEMPLATES = {
    "streams/aggregated_streams_by_sos.sql": {
        **PERMS,
        "isrc": "ISRC",
        "max_available_date": "2026-06-12",
        "days_back": 28,
        "store_ids": [1, 286],
        "countries": [],
        "is_feed_data_available": False,
    },
    "streams/aggregated_streams_by_sos_v2.sql": {
        **PERMS,
        "isrc": "ISRC",
        "max_available_date": "2026-06-12",
        "store_ids": [1, 286],
        "countries": [],
        "streams_sos_columns": ["streams_passive", "streams_active"],
        "is_feed_data_available": False,
    },
    "account/products_by_downloads_fixed_period.sql": {
        **PERMS,
        "account_id": 123,
        "account_type": "vendor",
        "max_available_date": "2026-06-12",
        "countries": [],
        "store_ids": [],
        "deletions": False,
        "sale_start_date_months_back": 0,
        "order_by": "downloads_7_days",
        "order_dir": "DESC",
        "limit": 5,
    },
}


def _render_file(rel_path, params):
    """Render a template file through the production jinjasql engine."""
    text = SQL_DIR.joinpath(rel_path).read_text(encoding="utf-8")
    query = get_template_engine().prepare_query(text, params)[0]
    return strip_sql(query)


@pytest.mark.parametrize("rel_path,params", list(TEMPLATES.items()))
def test_template_joins_published_view_when_enabled(rel_path, params):
    """Flag on: the availability join reads the _PUBLISHED twin."""
    sql = _render_file(rel_path, {**params, FLAG: True})
    assert PUBLISHED_VIEW in sql, f"_PUBLISHED join missing in {rel_path}"


@pytest.mark.parametrize("rel_path,params", list(TEMPLATES.items()))
def test_template_joins_original_view_when_disabled(rel_path, params):
    """Flag off: the original join, no _PUBLISHED reference anywhere."""
    sql = _render_file(rel_path, {**params, FLAG: False})
    assert PUBLISHED_VIEW not in sql, f"unexpected _PUBLISHED join in {rel_path}"
    assert ORIGINAL_VIEW in sql, f"original view join missing in {rel_path}"


@pytest.mark.parametrize("rel_path,params", list(TEMPLATES.items()))
def test_flag_off_renders_identically_to_flag_absent(rel_path, params):
    """Flag off must be a clean no-op: byte-identical to the pre-flag render."""
    with_flag_off = _render_file(rel_path, {**params, FLAG: False})
    without_flag = _render_file(rel_path, params)
    assert with_flag_off == without_flag


# --- account template: anchor cap on the fact scan (flag-gated) ------------ #

ACCOUNT_TEMPLATE = "account/products_by_downloads_fixed_period.sql"
# Unlike the SOS templates, the account fact scan had no upper date bound, so
# mid-swap-batch the freshly swapped downloads table (already holding the next
# day) leaked an extra day into the 7/28-day windows. The cap is gated behind
# the same flag to keep the flag-off render byte-identical. The `s.` qualifier
# makes the substring unique: the pre-existing upper bounds in this template
# are all against sdd.* prev-window dates.
ANCHOR_CAP = "s.download_activity_date <="


def test_account_template_caps_fact_scan_at_anchor_when_enabled():
    """Flag on: the downloads scan is capped at the anchor date."""
    sql = _render_file(ACCOUNT_TEMPLATE, {**TEMPLATES[ACCOUNT_TEMPLATE], FLAG: True})
    assert ANCHOR_CAP in sql


def test_account_template_has_no_anchor_cap_when_disabled():
    """Flag off: no cap — the scan matches today's production render."""
    sql = _render_file(ACCOUNT_TEMPLATE, {**TEMPLATES[ACCOUNT_TEMPLATE], FLAG: False})
    assert ANCHOR_CAP not in sql


def test_account_query_class_threads_flag_through_schema():
    """The flag survives AccountProductsFixedPeriodQuerySchema's load: the
    real query class renders both the _PUBLISHED join and the anchor cap."""
    query = AccountProductsByDownloadsFixedPeriod(
        {**TEMPLATES[ACCOUNT_TEMPLATE], FLAG: True}
    )
    sql = strip_sql(query.prepare_query()[0])
    assert PUBLISHED_VIEW in sql
    assert ANCHOR_CAP in sql


def test_account_query_class_defaults_to_uncapped_original_view():
    """Without the param the schema default (False) renders today's SQL."""
    query = AccountProductsByDownloadsFixedPeriod(TEMPLATES[ACCOUNT_TEMPLATE])
    sql = strip_sql(query.prepare_query()[0])
    assert PUBLISHED_VIEW not in sql
    assert ORIGINAL_VIEW in sql
    assert ANCHOR_CAP not in sql


# --- schema threading: the flag survives the query-class param load -------- #
# Every query class that renders one of the three templates, so the proof runs
# through each schema chain (SoundRecording / Product / GlobalParticipant
# AggregatedStreamsQuerySchema and AccountProductsFixedPeriodQuerySchema), not
# just the raw template files.


def _aggregated_streams_params(**overrides):
    """Complete param set accepted by all aggregated-streams query classes."""
    params = {
        **PERMS,
        "days_back": 28,
        "dimension": "SOS",
        "top_size": 5,
        "order_by": "streams7Days",
        "max_available_date": "2026-06-12",
        "store_ids": [],
        "countries": [],
        "is_feed_data_available": False,
        "streams_sos_columns": ["streams_passive", "streams_active"],
    }
    params.update(overrides)
    return params


QUERY_CLASS_CASES = [
    (SoundRecordingAggregatedStreamsBySOS, _aggregated_streams_params(isrc="ISRC")),
    (SoundRecordingAggregatedStreamsBySOSV2, _aggregated_streams_params(isrc="ISRC")),
    (ProductAggregatedStreamsBySOS, _aggregated_streams_params(product_id="123")),
    (
        GlobalParticipantAggregatedStreamsBySOS,
        _aggregated_streams_params(global_participant_id="GP1"),
    ),
    (AccountProductsByDownloadsFixedPeriod, dict(TEMPLATES[ACCOUNT_TEMPLATE])),
]
QUERY_CLASS_IDS = [cls.__name__ for cls, _ in QUERY_CLASS_CASES]


@pytest.mark.parametrize("query_class,params", QUERY_CLASS_CASES, ids=QUERY_CLASS_IDS)
def test_query_class_threads_flag_through_schema(query_class, params):
    """published_max_available_date_enabled=True survives the schema load."""
    query = query_class({**params, FLAG: True})
    sql = strip_sql(query.prepare_query()[0])
    assert PUBLISHED_VIEW in sql


@pytest.mark.parametrize("query_class,params", QUERY_CLASS_CASES, ids=QUERY_CLASS_IDS)
def test_query_class_defaults_to_original_view(query_class, params):
    """Without the param the schema default (False) keeps the original join."""
    query = query_class(params)
    sql = strip_sql(query.prepare_query()[0])
    assert PUBLISHED_VIEW not in sql
    assert ORIGINAL_VIEW in sql