gendesign/tradein-mvp/backend/tests/test_listing_source_snapshot.py

"""Tests for the per-source listing price-history snapshot writer (#570).

Writer (snapshot_listing_sources) is SQL-heavy, so most assertions are static: we read
the emitted SQL via inspect/text-attr and check the upsert + event-diff CTE shape, table/
column names, and the psycopg-v3 cast discipline (no :param::type). We also assert the
migration 079 contents.

Plus one cheap behavioural test: a fake db (monkeypatched .execute) drives the counter
logic without a real Postgres.

Static style mirrors tests/test_rosreestr_dedup_key.py / test_yandex_city_sweep.py.
"""

import inspect
import os
import re
from pathlib import Path
from typing import Any

import pytest

# Importing app.services.scheduler / app.tasks pulls app.core.config.Settings → needs
# DATABASE_URL. Stub it BEFORE app imports (as in test_scheduler.py) — these tests are
# static / fake-db; no live database is touched.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from app.tasks import listing_source_snapshot as snap_mod

_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
_MIGRATION_079 = _SQL_DIR / "079_listing_source_history.sql"

# Emitted SQL text (SQLAlchemy text() clause → .text gives the raw string).
_SNAPSHOT_SQL = str(snap_mod._SNAPSHOT_SQL.text)
_EVENT_DIFF_SQL = str(snap_mod._EVENT_DIFF_SQL.text)
_ALL_SQL = _SNAPSHOT_SQL + "\n" + _EVENT_DIFF_SQL
_WRITER_SRC = inspect.getsource(snap_mod.snapshot_listing_sources)


# ── Snapshot upsert SQL ───────────────────────────────────────────────────────


def test_snapshot_targets_correct_table_and_columns() -> None:
    assert "INSERT INTO listing_source_snapshots" in _SNAPSHOT_SQL
    for col in (
        "listing_source_id",
        "snapshot_date",
        "price_rub",
        "is_active",
        "last_seen_at",
        "payload_hash",
        "observed_at",
        "run_id",
    ):
        assert col in _SNAPSHOT_SQL, f"snapshot SQL missing column {col!r}"
    assert "FROM listing_sources" in _SNAPSHOT_SQL
    assert "CURRENT_DATE" in _SNAPSHOT_SQL


def test_snapshot_is_upsert_last_write_wins() -> None:
    """ON CONFLICT (listing_source_id, snapshot_date) DO UPDATE — last-write-wins for the day."""
    assert "ON CONFLICT (listing_source_id, snapshot_date) DO UPDATE" in _SNAPSHOT_SQL
    # Collapse runs of whitespace so column-alignment spacing doesn't matter.
    flat = re.sub(r"\s+", " ", _SNAPSHOT_SQL)
    for col in ("price_rub", "is_active", "last_seen_at", "payload_hash", "observed_at", "run_id"):
        assert f"{col} = EXCLUDED.{col}" in flat, f"upsert SET missing {col}"


def test_snapshot_derives_is_active_and_payload_hash() -> None:
    # is_active derived from freshness window; payload_hash = md5(raw_payload::text).
    assert "make_interval(days => :freshness_days)" in _SNAPSHOT_SQL
    assert "AS is_active" in _SNAPSHOT_SQL
    assert "md5(raw_payload::text)" in _SNAPSHOT_SQL


# ── Event-diff CTE SQL ────────────────────────────────────────────────────────


def test_event_diff_is_set_based_lateral_not_python_loop() -> None:
    """Event diff is one set-based INSERT … SELECT with a LATERAL join — no Python loop.

    #2607: prior used to be a `DISTINCT ON (listing_source_id) ... FROM
    listing_source_snapshots` CTE joined via plain JOIN — the planner's Nested Loop
    (no Materialize, misestimated `today` row count) re-executed the DISTINCT ON over
    the whole table once per today-row, hanging for days. Rewritten as `JOIN LATERAL
    (... ORDER BY snapshot_date DESC LIMIT 1) ON true` — forces a per-row indexed
    point-lookup via idx_lss_source_date instead of a full-table DISTINCT ON.
    """
    assert "WITH today AS" in _EVENT_DIFF_SQL
    assert "JOIN LATERAL" in _EVENT_DIFF_SQL
    assert "prior AS" not in _EVENT_DIFF_SQL, "prior CTE removed — replaced by LATERAL (#2607)"
    assert "DISTINCT ON" not in _EVENT_DIFF_SQL, "DISTINCT ON over full table removed (#2607)"
    assert "INSERT INTO listing_source_events" in _EVENT_DIFF_SQL
    # LATERAL subquery: most-recent snapshot strictly before today, per listing_source_id.
    assert "s.listing_source_id = t.listing_source_id" in _EVENT_DIFF_SQL
    assert "s.snapshot_date < CURRENT_DATE" in _EVENT_DIFF_SQL
    assert "snapshot_date = CURRENT_DATE" in _EVENT_DIFF_SQL
    assert "ORDER BY s.snapshot_date DESC" in _EVENT_DIFF_SQL
    assert "LIMIT 1" in _EVENT_DIFF_SQL
    # No Python iteration over rows in the writer body (set-based only — LATERAL is a
    # Postgres execution-plan construct, not a Python loop).
    body = _WRITER_SRC.split('"""', 2)[-1]
    assert "for " not in body, "writer must be set-based — no Python row loop"


def test_event_diff_emits_price_change_with_diff_percent() -> None:
    assert "'price_change'" in _EVENT_DIFF_SQL
    # diff_percent = (new-old)/old*100. NULLIF на знаменателе (#2674): выражения
    # VALUES вычисляются ДО фильтра e.fires, поэтому предикат "p.price_rub <> 0"
    # больше не защищает само деление.
    assert "(t.price_rub - p.price_rub)" in _EVENT_DIFF_SQL
    assert "/ NULLIF(p.price_rub, 0) * 100" in _EVENT_DIFF_SQL
    # Only when the price actually changed and old is a usable denominator.
    assert "t.price_rub <> p.price_rub" in _EVENT_DIFF_SQL
    assert "p.price_rub <> 0" in _EVENT_DIFF_SQL
    assert "p.price_rub IS NOT NULL" in _EVENT_DIFF_SQL


def test_event_diff_is_idempotent_on_conflict() -> None:
    assert "ON CONFLICT (listing_source_id, change_time, event_type) DO NOTHING" in _EVENT_DIFF_SQL


# ── psycopg v3 cast discipline ────────────────────────────────────────────────


def test_sql_uses_psycopg_v3_casts_not_double_colon() -> None:
    """psycopg v3: bind params via CAST(:x AS type), never :x::type.

    Literal ::-casts on columns (raw_payload::text, ::numeric) are fine — the ban is
    only on :param::type, which psycopg v3 breaks.
    """
    # No bind-param (:name) followed by ::type anywhere in the emitted SQL.
    assert not re.search(r":\w+::", _ALL_SQL)
    # run_id bound via CAST(:run_id AS bigint).
    assert "CAST(:run_id AS bigint)" in _SNAPSHOT_SQL
    # freshness window bound via named param inside make_interval (no cast needed).
    assert ":freshness_days" in _SNAPSHOT_SQL


# ── Writer return / finalisation contract ─────────────────────────────────────


def test_writer_returns_counters_and_finalises_run() -> None:
    assert "snapshotted" in _WRITER_SRC
    assert "price_change_events" in _WRITER_SRC
    assert "mark_done" in _WRITER_SRC
    assert "mark_failed" in _WRITER_SRC


# ── Migration 079 ─────────────────────────────────────────────────────────────


def test_migration_079_exists() -> None:
    assert _MIGRATION_079.is_file(), f"missing migration: {_MIGRATION_079}"


def test_migration_079_creates_three_schema_objects_and_view() -> None:
    sql = _MIGRATION_079.read_text("utf-8")
    # 1. snapshots table (plain — explicitly NOT partitioned, noted in header).
    assert "CREATE TABLE IF NOT EXISTS listing_source_snapshots" in sql
    assert "PRIMARY KEY (listing_source_id, snapshot_date)" in sql
    assert "REFERENCES listing_sources(id) ON DELETE CASCADE" in sql
    assert "REFERENCES scrape_runs(id) ON DELETE SET NULL" in sql
    assert "CREATE INDEX IF NOT EXISTS idx_lss_snapshot_date" in sql
    assert "CREATE INDEX IF NOT EXISTS idx_lss_source_date" in sql
    # 2. events table + CHECK + UNIQUE.
    assert "CREATE TABLE IF NOT EXISTS listing_source_events" in sql
    assert "id                bigserial   PRIMARY KEY" in sql.replace("\t", " ") or (
        "id" in sql and "bigserial" in sql and "PRIMARY KEY" in sql
    )
    assert "'price_change', 'delisted', 'relisted', 'edited', 'first_seen'" in sql
    assert "UNIQUE (listing_source_id, change_time, event_type)" in sql
    # 3. offer_price_history backlink column + index.
    assert "ALTER TABLE offer_price_history" in sql
    assert "ADD COLUMN IF NOT EXISTS listing_source_id bigint" in sql
    assert "idx_oph_source_change_time" in sql
    # 4. view.
    assert "CREATE OR REPLACE VIEW v_listing_source_price_on_date" in sql
    assert "JOIN listing_sources ls ON ls.id = lss.listing_source_id" in sql
    # Transactional + idempotent + partitioning note.
    assert "BEGIN;" in sql and "COMMIT;" in sql
    assert "НЕТ" in sql and "артицион" in sql  # partitioning explicitly deferred


def test_migration_079_seeds_schedule_enabled_true() -> None:
    sql = _MIGRATION_079.read_text("utf-8")
    assert "'listing_source_snapshot'" in sql
    assert "'{}'::jsonb" in sql
    assert "ON CONFLICT (source) DO NOTHING" in sql
    # enabled=true (SAFE — pure internal DB). Assert the seeded VALUES row has true,
    # not false (unlike the dormant yandex/avito sweeps).
    assert "true,                                        -- SAFE" in sql
    assert "false" not in sql.split("INSERT INTO scrape_schedules")[1]


def test_migration_079_uses_psycopg_safe_sql() -> None:
    """Migration is plain DDL (no bind params), but guard against accidental :x::type."""
    sql = _MIGRATION_079.read_text("utf-8")
    assert not re.search(r":\w+::", sql)


# ── Cheap behavioural test: counter logic via fake db ─────────────────────────


class _FakeResult:
    def __init__(self, rowcount: int, rows: list[tuple[str, int]] | None = None) -> None:
        self.rowcount = rowcount
        self._rows = rows or []

    def fetchall(self) -> list[tuple[str, int]]:
        """Event-diff statement возвращает пары (counter_key, n) — см. #2674."""
        return self._rows


class _FakeDB:
    """Minimal stand-in for a SQLAlchemy Session — records execute() calls, returns rowcounts."""

    def __init__(
        self, rowcounts: list[int], event_rows: list[tuple[str, int]] | None = None
    ) -> None:
        self._rowcounts = list(rowcounts)
        self._event_rows = event_rows or []
        self.executed: list[Any] = []
        self.committed = False
        self.rolled_back = False

    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
        self.executed.append((stmt, params))
        return _FakeResult(self._rowcounts.pop(0), self._event_rows)

    def commit(self) -> None:
        self.committed = True

    def rollback(self) -> None:
        self.rolled_back = True


def test_counter_logic_with_fake_db(monkeypatch: pytest.MonkeyPatch) -> None:
    """snapshot_listing_sources maps the snapshot rowcount + per-type event rows to counters.

    #2674: event-diff больше не отдаёт один rowcount — внешний SELECT над
    data-modifying CTE возвращает пары (counter_key, n) по типам событий, а
    counters инициализированы всеми пятью ключами (не сработавший тип = честный 0).
    """
    marked: dict[str, Any] = {}
    monkeypatch.setattr(
        snap_mod.runs_mod,
        "mark_done",
        lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
    )
    monkeypatch.setattr(snap_mod.runs_mod, "mark_failed", lambda *a, **k: None)

    # rowcounts: SET LOCAL statement_timeout (ignored), snapshot upsert, event-diff.
    db = _FakeDB(rowcounts=[0, 18355, 0], event_rows=[("price_change_events", 42)])
    out = snap_mod.snapshot_listing_sources(db, run_id=99)  # type: ignore[arg-type]

    assert out == {
        "snapshotted": 18355,
        "price_change_events": 42,
        "edited_events": 0,
        "first_seen_events": 0,
    }
    assert db.committed is True
    assert len(db.executed) == 3
    # First statement sets the per-transaction wall-clock budget (#2607).
    stmt0, _params0 = db.executed[0]
    assert "SET LOCAL statement_timeout" in str(stmt0)
    # run_id threaded into the snapshot statement's bind params (now executed[1]).
    _stmt, params = db.executed[1]
    assert params is not None and params["run_id"] == 99
    # Run finalised via mark_done with the same counters.
    assert marked["run_id"] == 99
    assert marked["counters"] == out


# ── budget_sec / statement_timeout (#2607) ─────────────────────────────────────


def test_clamp_budget_sec_defaults_and_bounds() -> None:
    assert snap_mod._clamp_budget_sec(snap_mod.DEFAULT_BUDGET_SEC) == snap_mod.DEFAULT_BUDGET_SEC
    # Below floor / garbage / zero (the historical bug: 0 == "no timeout") clamp to the floor.
    assert snap_mod._clamp_budget_sec(0) == snap_mod._MIN_BUDGET_SEC
    assert snap_mod._clamp_budget_sec(-5) == snap_mod._MIN_BUDGET_SEC
    assert snap_mod._clamp_budget_sec(None) == snap_mod.DEFAULT_BUDGET_SEC
    assert snap_mod._clamp_budget_sec("garbage") == snap_mod.DEFAULT_BUDGET_SEC
    # Above ceiling clamps down — never lets a fat-fingered value re-create "hangs forever".
    assert snap_mod._clamp_budget_sec(999_999) == snap_mod._MAX_BUDGET_SEC
    # Sane custom value passes through unclamped.
    assert snap_mod._clamp_budget_sec(120) == 120.0


def test_snapshot_listing_sources_sets_statement_timeout_from_params(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """budget_sec from default_params is applied via SET LOCAL statement_timeout (ms)."""
    monkeypatch.setattr(snap_mod.runs_mod, "mark_done", lambda *a, **k: None)
    monkeypatch.setattr(snap_mod.runs_mod, "mark_failed", lambda *a, **k: None)

    db = _FakeDB(rowcounts=[0, 10, 1])
    snap_mod.snapshot_listing_sources(db, run_id=1, params={"budget_sec": 120})  # type: ignore[arg-type]

    stmt0, _params0 = db.executed[0]
    assert "SET LOCAL statement_timeout = 120000" in str(stmt0)


def test_snapshot_listing_sources_defaults_budget_sec_when_params_missing(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """No params / no budget_sec key → DEFAULT_BUDGET_SEC applied (never unlimited/0)."""
    monkeypatch.setattr(snap_mod.runs_mod, "mark_done", lambda *a, **k: None)
    monkeypatch.setattr(snap_mod.runs_mod, "mark_failed", lambda *a, **k: None)

    db = _FakeDB(rowcounts=[0, 10, 1])
    snap_mod.snapshot_listing_sources(db, run_id=1)  # type: ignore[arg-type]

    stmt0, _params0 = db.executed[0]
    expected_ms = int(snap_mod.DEFAULT_BUDGET_SEC * 1000)
    assert f"SET LOCAL statement_timeout = {expected_ms}" in str(stmt0)


def test_counter_logic_failure_path_marks_failed(monkeypatch: pytest.MonkeyPatch) -> None:
    """On execute error: rollback + mark_failed + re-raise (no silent swallow)."""
    failed: dict[str, Any] = {}
    monkeypatch.setattr(snap_mod.runs_mod, "mark_done", lambda *a, **k: None)
    monkeypatch.setattr(
        snap_mod.runs_mod,
        "mark_failed",
        lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err),
    )

    class _BoomDB(_FakeDB):
        def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
            raise RuntimeError("boom")

    db = _BoomDB(rowcounts=[])
    with pytest.raises(RuntimeError, match="boom"):
        snap_mod.snapshot_listing_sources(db, run_id=7)  # type: ignore[arg-type]
    assert db.rolled_back is True
    assert failed["run_id"] == 7