tests/test_quality_gate_semantics.py

"""
Semantic (non-fatal) checks in the post-load quality gate.

The original gate only caught measures that loaded 100% zero/null — a
guaranteed blank tile. These checks catch data that *renders* but reads wrong,
which is worse in a demo because nobody notices until a prospect does:

  - Tixr shipped TICKET_SELL_THROUGH_RATE = 1.96 (a 196% sell-through).
  - Triumph's *_MONTH bucket columns held random days of the month.
  - 500 fact rows spread over 24 months left most periods nearly empty.

These are warnings, not failures, by design — see _run_semantic_checks.

Run with: pytest tests/test_quality_gate_semantics.py -v
"""

import pytest

from demoprep_app.integrations.snowflake.quality_gate import (
    MIN_ROWS_PER_MONTH_WARN,
    bounded_range_for_column,
    is_date_column,
    is_month_bucket_column,
    run_measure_quality_gate,
)


class TestClassifiers:
    def test_rate_and_ratio_are_zero_to_one(self):
        assert bounded_range_for_column("TICKET_SELL_THROUGH_RATE") == (0.0, 1.0)
        assert bounded_range_for_column("WIN_RATIO") == (0.0, 1.0)

    def test_percent_columns_are_zero_to_one_hundred(self):
        assert bounded_range_for_column("BROKER_MARGIN_PCT") == (0.0, 100.0)
        assert bounded_range_for_column("DISCOUNT_PERCENT") == (0.0, 100.0)
        assert bounded_range_for_column("UPTIME_PERCENTAGE") == (0.0, 100.0)

    def test_unbounded_columns_return_none(self):
        assert bounded_range_for_column("TOTAL_REVENUE_USD") is None
        assert bounded_range_for_column("ORDER_COUNT") is None
        # "RATE" must be a suffix, not a substring — RATED_CAPACITY is not a rate.
        assert bounded_range_for_column("RATED_CAPACITY") is None

    def test_month_bucket_requires_date_type(self):
        assert is_month_bucket_column("ORDER_MONTH", "DATE")
        assert is_month_bucket_column("ORDER_MONTH", "TIMESTAMP_NTZ")
        # A numeric *_MONTH is a month number, not a bucket date.
        assert not is_month_bucket_column("FISCAL_MONTH", "NUMBER(2,0)")
        assert not is_month_bucket_column("ORDER_DATE", "DATE")

    def test_is_date_column(self):
        assert is_date_column("DATE")
        assert is_date_column("TIMESTAMP_LTZ")
        assert not is_date_column("VARCHAR(10)")
        assert not is_date_column("NUMBER")


class SemanticCursor:
    """Routes each query shape to a canned answer.

    Distinguishes the four query shapes the gate issues: the column catalogue,
    the per-table non-zero aggregate, bounds violations, month-start
    violations, and the density probe.
    """

    def __init__(self, columns_rows, aggregate, *, out_of_range=(0, None, None),
                 bad_month_rows=0, density=(1000, 12)):
        self._columns_rows = columns_rows
        self._aggregate = aggregate
        self._out_of_range = out_of_range
        self._bad_month_rows = bad_month_rows
        self._density = density
        self._result = None

    def execute(self, sql, params=None):
        if "INFORMATION_SCHEMA.COLUMNS" in sql:
            self._result = self._columns_rows
        elif "DATE_TRUNC" in sql:
            self._result = self._density
        elif "DAY(" in sql:
            self._result = (self._bad_month_rows,)
        elif "MIN(" in sql and "MAX(" in sql:
            self._result = self._out_of_range
        else:
            self._result = self._aggregate

    def fetchall(self):
        return self._result

    def fetchone(self):
        return self._result

    def close(self):
        pass


class SemanticConnection:
    def __init__(self, cursor):
        self._cursor = cursor

    def cursor(self):
        return self._cursor


COLUMNS = [
    ("FACT_TICKET_SALE", "TICKET_ID", "NUMBER"),
    ("FACT_TICKET_SALE", "TICKETS_SOLD", "NUMBER"),
    ("FACT_TICKET_SALE", "TICKET_SELL_THROUGH_RATE", "FLOAT"),
    ("FACT_TICKET_SALE", "SALE_MONTH", "DATE"),
]
# (row_count, nonzero TICKETS_SOLD, nonzero RATE)
CLEAN_AGG = (1000, 1000, 1000)


def _run(cursor):
    return run_measure_quality_gate(SemanticConnection(cursor), "DEMO_SCHEMA")


def test_out_of_range_rate_warns_but_does_not_fail():
    cursor = SemanticCursor(COLUMNS, CLEAN_AGG, out_of_range=(7, 0.0, 1.96))
    profile = _run(cursor)

    warning = " ".join(profile["semantic_warnings"])
    assert "TICKET_SELL_THROUGH_RATE" in warning
    assert "7 row(s) outside expected range" in warning
    assert "1.96" in warning
    assert profile["failures"] == []


def test_non_month_start_dates_warn():
    cursor = SemanticCursor(COLUMNS, CLEAN_AGG, bad_month_rows=412)
    profile = _run(cursor)

    warning = " ".join(profile["semantic_warnings"])
    assert "SALE_MONTH" in warning
    assert "412 row(s) are not month-start" in warning


def test_sparse_panel_warns():
    """500 rows over 24 months = ~21/month, below the density threshold."""
    cursor = SemanticCursor(COLUMNS, CLEAN_AGG, density=(500, 24))
    profile = _run(cursor)

    warning = " ".join(profile["semantic_warnings"])
    assert "FACT_TICKET_SALE is sparse over time" in warning
    assert "20.8/month" in warning


def test_dense_panel_is_silent():
    cursor = SemanticCursor(COLUMNS, CLEAN_AGG, density=(12000, 24))
    profile = _run(cursor)
    assert not any("sparse" in w for w in profile["semantic_warnings"])


def test_clean_data_produces_no_semantic_warnings():
    cursor = SemanticCursor(COLUMNS, CLEAN_AGG)
    profile = _run(cursor)
    assert profile["semantic_warnings"] == []


def test_semantic_failure_cannot_break_the_run():
    """A broken semantic probe degrades to a note, never an exception."""

    class ExplodingCursor(SemanticCursor):
        def execute(self, sql, params=None):
            if "INFORMATION_SCHEMA.COLUMNS" in sql:
                self._result = self._columns_rows
            elif "SUM(IFF" in sql:
                self._result = self._aggregate
            else:
                raise RuntimeError("probe blew up")

    profile = _run(ExplodingCursor(COLUMNS, CLEAN_AGG))
    assert any("semantic checks skipped" in w for w in profile["warnings"])
    assert profile["failures"] == []


def test_density_threshold_is_named_not_magic():
    assert MIN_ROWS_PER_MONTH_WARN == 30