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

"""Tests for the daily asking→sold ratio refresh (#648 Stage 4).

recompute_asking_to_sold_ratios is SQL-heavy, so most assertions are static: we read the
emitted SQL via .text and check the TRUE-MIRROR shape (DELETE WHERE district='' BEFORE the
re-derivation INSERT), that the derivation reuses the exact 080 logic (deals + listings, the
trailing-12mo window, the 30/30 threshold, the global -1 fallback row), and the psycopg-v3
cast discipline (no :param::type). We also assert the migration 082 contents.

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

Static style mirrors tests/test_listing_source_snapshot.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 asking_to_sold_ratio as ratio_mod

_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
_MIGRATION_080 = _SQL_DIR / "080_asking_to_sold_ratios.sql"
_MIGRATION_082 = _SQL_DIR / "082_scrape_schedules_seed_ratio_refresh.sql"

# Emitted SQL text (SQLAlchemy text() clause → .text gives the raw string).
_DELETE_SQL = str(ratio_mod._DELETE_SQL.text)
_REDERIVE_SQL = str(ratio_mod._REDERIVE_SQL.text)
_DELETE_SQL_REGION = str(ratio_mod._DELETE_SQL_REGION.text)
_REDERIVE_SQL_REGION = str(ratio_mod._REDERIVE_SQL_REGION.text)
_COUNTERS_SQL = str(ratio_mod._COUNTERS_SQL.text)
_ALL_SQL = (
    _DELETE_SQL
    + "\n"
    + _REDERIVE_SQL
    + "\n"
    + _DELETE_SQL_REGION
    + "\n"
    + _REDERIVE_SQL_REGION
    + "\n"
    + _COUNTERS_SQL
)
_TASK_SRC = inspect.getsource(ratio_mod.recompute_asking_to_sold_ratios)


# ── TRUE-MIRROR: DELETE-then-INSERT ───────────────────────────────────────────


def test_refresh_deletes_district_rows_first() -> None:
    """True-mirror: DELETE FROM asking_to_sold_ratios WHERE region=66/district='' first.

    The 080 seed used ON CONFLICT DO UPDATE which leaves stale per-rooms rows for buckets
    that drop below 30/30 on a later run. The refresh must DELETE first instead.

    #3512: the EKB delete is now scoped to `region_code = 66` too (region-aware ratios) —
    the other-region loop uses its own _DELETE_SQL_REGION, tested separately below.
    """
    flat_del = re.sub(r"\s+", " ", _DELETE_SQL).strip()
    assert "DELETE FROM asking_to_sold_ratios" in flat_del
    assert "WHERE region_code = 66 AND district = ''" in flat_del
    # The re-derivation is an INSERT (no ON CONFLICT — DELETE precedes it).
    assert "INSERT INTO asking_to_sold_ratios" in _REDERIVE_SQL
    assert "ON CONFLICT" not in _REDERIVE_SQL


def test_task_body_runs_delete_before_insert_in_one_txn() -> None:
    """Task executes DELETE then re-derive INSERT, commits ONCE (atomic refresh)."""
    body = _TASK_SRC
    del_pos = body.index("_DELETE_SQL")
    ins_pos = body.index("_REDERIVE_SQL")
    commit_pos = body.index("db.commit()")
    # DELETE executed before INSERT, both before the single commit (one transaction).
    assert del_pos < ins_pos < commit_pos
    # Exactly one commit — never an intermediate commit that could leave the table empty.
    assert body.count("db.commit()") == 1


# ── Derivation matches 080 (refresh == re-seed) ───────────────────────────────


def test_rederivation_references_deals_and_listings() -> None:
    assert "FROM deals" in _REDERIVE_SQL
    assert "FROM listings" in _REDERIVE_SQL
    assert "source = 'rosreestr'" in _REDERIVE_SQL
    assert "is_active" in _REDERIVE_SQL


def test_rederivation_uses_12_month_window() -> None:
    """Trailing-12mo deal window — same as the 080 seed."""
    assert "deal_date >= CURRENT_DATE - INTERVAL '12 months'" in _REDERIVE_SQL
    # window_months literal 12 also persisted into the rows.
    flat = re.sub(r"\s+", " ", _REDERIVE_SQL)
    assert "12 AS window_months" in flat


def test_rederivation_applies_3030_threshold_and_ppm_band() -> None:
    assert "d.n_deals >= 30" in _REDERIVE_SQL
    assert "a.n_listings >= 30" in _REDERIVE_SQL
    # ppm² bounds are now bind parameters (not hardcoded literals) — prevents SQL injection
    # and makes the ceiling configurable via settings.asking_ratio_ppm2_max (#767).
    assert "price_per_m2 BETWEEN :ppm2_min AND :ppm2_max" in _REDERIVE_SQL
    assert "LEAST(GREATEST(rooms, 0), 4)" in _REDERIVE_SQL
    assert "percentile_cont(0.5)" in _REDERIVE_SQL


def test_rederivation_writes_global_fallback_row() -> None:
    """Global -1 fallback row (basis='global_fallback') always written when ask>0."""
    assert "-1                                        AS rooms_bucket" in _REDERIVE_SQL or (
        "-1" in _REDERIVE_SQL and "rooms_bucket" in _REDERIVE_SQL
    )
    assert "'global_fallback'" in _REDERIVE_SQL
    assert "'per_rooms'" in _REDERIVE_SQL
    assert "deal_global" in _REDERIVE_SQL
    assert "ask_global" in _REDERIVE_SQL


def test_rederivation_cte_blocks_match_080() -> None:
    """The CTE names are byte-for-byte the 080 derivation blocks."""
    for cte in ("deal_side", "ask_side", "per_bucket", "deal_global", "ask_global", "global_row"):
        assert f"{cte} AS" in _REDERIVE_SQL, f"missing CTE {cte!r}"


def test_rederivation_scopes_sold_side_to_asking_city() -> None:
    """#C2: SOLD-сторона (deal_side + deal_global) скоупится на город asking-стороны (ЕКБ).

    Миграция 177 залила ДКП по всей обл.66, а asking (listings) исторически — только ЕКБ.
    Без скоупа sold-медиана смешивала дешёвую область → ratio 0.877→0.62, «выкупная» −29%.
    Оба deal-CTE (per-rooms + global) несут предикат unconditionally (deals.city не имеет
    массовых NULL как listings.city — #2598 их не касается).
    """
    assert ratio_mod._ASKING_CITY_PATTERN == "%Екатеринбург%"
    # Оба deal-CTE (deal_side + deal_global) скоупятся unconditional-предикатом —
    # ровно 2 вхождения формы БЕЗ city IS NULL (ask-сторона использует другую форму,
    # см. test_ask_side_and_ask_global_scoped_to_asking_city).
    assert _REDERIVE_SQL.count("AND city ILIKE :asking_city") == 2


def test_ask_side_and_ask_global_scoped_to_asking_city() -> None:
    """#2583 H2: ask-сторона (ask_side + ask_global) ТЕПЕРЬ ТОЖЕ скоупится на asking_city.

    Oblast-развёртки заработали 12 июля — областные объявления (дешевле ЕКБ) попали в
    знаменатель ask_median БЕЗ городского скоупа, а sold-сторона осталась скоуплена на
    ЕКБ (см. предыдущий тест) → асимметрия занижала ask_median и завышала ratio на
    2.5-5.3% по бакетам комнат 1-4 (замер на проде, аудит #2583 H2). Falsifiable: этот
    assert FALSE на непропатченном коде (ask_side/ask_global без city-предиката вообще)
    и TRUE после того как предикат `(city IS NULL OR city ILIKE :asking_city)` добавлен —
    проверено `git stash` на строках реализации.
    """
    _a = _REDERIVE_SQL.index("ask_side AS")
    _b = _REDERIVE_SQL.index("per_bucket AS")
    ask_side_block = _REDERIVE_SQL[_a:_b]
    assert "AND (city IS NULL OR city ILIKE :asking_city)" in ask_side_block

    _c = _REDERIVE_SQL.index("ask_global AS")
    _d = _REDERIVE_SQL.index("global_row AS")
    ask_global_block = _REDERIVE_SQL[_c:_d]
    assert "AND (city IS NULL OR city ILIKE :asking_city)" in ask_global_block


def test_ask_side_keeps_city_is_null_rows_not_naive_filter() -> None:
    """Guard against the naive (wrong) fix — a plain symmetric `city ILIKE :asking_city`.

    listings.city заполнена пока только у Авито (#2598/#2606) — Циан/Домклик/Яндекс
    строки несут city IS NULL. На проде (2026-08, аудит #2583 H2) это ~8200 из ~11500
    строк, проходящих остальные WHERE-предикаты (~71%). Наивный симметричный
    `city ILIKE :asking_city` (как у deal_side) молча выбросил бы все city IS NULL
    строки, схлопнув ask_median c ~11500 до ~2100 ЕКБ-only объявлений — именно та
    over-correction, от которой предостерегает #2583 H2.
    """
    cte_pairs = (("ask_side AS", "per_bucket AS"), ("ask_global AS", "global_row AS"))
    for cte_name, next_cte in cte_pairs:
        start = _REDERIVE_SQL.index(cte_name)
        end = _REDERIVE_SQL.index(next_cte)
        block = _REDERIVE_SQL[start:end]
        assert "city IS NULL" in block, f"{cte_name}: missing IS NULL tolerance"
        # The naive fix (deal_side-style, no NULL tolerance) must NOT appear standalone.
        naive = re.search(r"AND\s+city\s+ILIKE\s+:asking_city(?!\))", block)
        assert naive is None, f"{cte_name}: found naive filter without IS NULL tolerance"


def _strip_sql(s: str) -> str:
    """Drop -- line comments and collapse whitespace — leaves only the executable SQL.

    Comments aren't load-bearing; the derivation MUST match 080 in its executable form
    (the CTE expressions / filters), which is what 'refresh == re-seed' means.
    """
    no_comments = re.sub(r"--[^\n]*", "", s)
    return re.sub(r"\s+", " ", no_comments).strip()


def test_migration_080_derivation_is_subset_of_refresh_sql() -> None:
    """Refresh derivation == 080 seed derivation (the WITH...SELECT before ON CONFLICT).

    Strip -- comments + normalise whitespace, then confirm the 080 CTE body
    (deal_side … per_bucket UNION ALL) is present verbatim in the refresh SQL — so
    refresh has exact re-seed semantics on the EXECUTABLE derivation.

    ppm² bounds are compared after normalisation: the 080 seed uses literal integers
    (30000 / 600000) while the refresh uses bind params (:ppm2_min / :ppm2_max) —
    both are replaced with the token PPM2_BAND_PLACEHOLDER before the subset check.

    #1186: the refresh now adds the novostroyki guard predicate to each ask_* CTE;
    it is normalised away here so the 080 seed (no guard) still matches.

    #2583 H2: the refresh now also adds the NULL-tolerant city-scope predicate to each
    ask_* CTE (symmetric to the #C2 SOLD-side guard) — normalised away the same way.

    #2620: the refresh now buckets ask_side by AREA (_AREA_ROOMS_BUCKET_SQL) instead of
    listings.rooms — the 080 seed still uses the rooms-based LEAST/GREATEST formula for
    ask_side (pre-#2620, the bug this fixes). Both bucket formulas normalise to the same
    placeholder token so this test keeps proving everything ELSE unchanged (CTE shape,
    threshold, window, deal_side bucketing) — the #2620 divergence itself is asserted by
    test_ask_side_buckets_by_area_not_rooms below.

    #2620 hardening: the refresh also adds `AND area_m2 IS NOT NULL` to ask_side/ask_global
    (NULL area_m2 would fall into the CASE ELSE branch = bucket 4 — a latent trap). Absent
    in the 080 seed; dropped here the same way as the other guards above.

    #2656: the refresh now also adds the freshness predicate (`scraped_at > NOW() -
    LISTINGS_FRESH_DAYS days`) to each ask_* CTE — the same window the estimator applies to
    the NUMERATOR (_COMMON_WHERE). Absent in the 080 seed; normalised away here, asserted
    on its own in tests/test_freshness_filter_2656.py.
    """
    seed_sql = _MIGRATION_080.read_text("utf-8")
    # Extract the WITH … (up to the ON CONFLICT) from the seed.
    with_idx = seed_sql.index("WITH")
    onconflict_idx = seed_sql.index("ON CONFLICT (rooms_bucket, district) DO UPDATE")
    seed_derivation = seed_sql[with_idx:onconflict_idx]

    def _normalise_ppm2(s: str) -> str:
        """Replace ppm² bound tokens with a common placeholder for cross-comparison."""
        # Seed uses literal integers; refresh uses bind params.
        s = re.sub(r"30000 AND 600000", "PPM2_BAND_PLACEHOLDER", s)
        s = re.sub(r":ppm2_min AND :ppm2_max", "PPM2_BAND_PLACEHOLDER", s)
        return s

    def _drop_segment_guard(s: str) -> str:
        """Remove the #1186 novostroyki guard predicate (absent in the 080 seed)."""
        return re.sub(
            r"AND\s*\((?:\w+\.)?listing_segment\s+IS\s+NULL\s+"
            r"OR\s+(?:\w+\.)?listing_segment\s*=\s*'vtorichka'\)",
            "",
            s,
        )

    def _drop_city_guard(s: str) -> str:
        """Remove the #C2 SOLD-side + #2583 H2 ASK-side city-scope predicates.

        Both are absent in the 080 seed: #C2 added the unconditional SOLD-side guard
        (deal_side/deal_global), #2583 H2 later added the NULL-tolerant ASK-side guard
        (ask_side/ask_global).
        """
        s = re.sub(r"AND\s+city\s+ILIKE\s+:asking_city", "", s)
        s = re.sub(r"AND\s*\(\s*city\s+IS\s+NULL\s+OR\s+city\s+ILIKE\s+:asking_city\s*\)", "", s)
        return s

    def _normalise_bucket_expr(s: str) -> str:
        """Collapse the rooms-based and area-based bucket formulas to one placeholder.

        #2620: ask_side buckets by area now (_AREA_ROOMS_BUCKET_SQL), not rooms. Both
        forms appear an equal number of times (2x each: SELECT expr + GROUP BY) once
        deal_side's untouched rooms-formula occurrences are also normalised, so this
        keeps the containment check valid for the parts of the derivation #2620 did NOT
        touch (deal_side stays rooms-bucketed — it has no other choice, deals.rooms IS
        the synthetic area bucket already).
        """
        s = re.sub(r"LEAST\(GREATEST\(rooms,\s*0\),\s*4\)", "BUCKET_PLACEHOLDER", s)
        s = re.sub(
            r"CASE WHEN area_m2 < 30 THEN 0 WHEN area_m2 < 44 THEN 1 "
            r"WHEN area_m2 < 62 THEN 2 WHEN area_m2 < 85 THEN 3 ELSE 4 END",
            "BUCKET_PLACEHOLDER",
            s,
        )
        return s

    def _drop_area_not_null_guard(s: str) -> str:
        """Remove the #2620 hardening guard (absent in the 080 seed).

        NULL area_m2 falls into the CASE ELSE branch (bucket 4) of the area formula — a
        latent trap. ask_side/ask_global added `AND area_m2 IS NOT NULL` explicitly.
        """
        return re.sub(r"AND\s+area_m2\s+IS\s+NOT\s+NULL", "", s)

    def _drop_freshness_guard(s: str) -> str:
        """Remove the #2656 freshness predicate (absent in the 080 seed)."""
        return re.sub(r"AND\s+scraped_at\s*>\s*NOW\(\)\s*-\s*\(:fresh_days[^\n]*?interval", "", s)

    def _drop_region_code_column(s: str) -> str:
        """Remove the #3512 region_code column/literal (absent in the 080 seed).

        Migration 304 adds region_code to the table; the refresh INSERT now writes it
        explicitly (region_code, ... ) / ( ..., 66) — the 080 seed predates the column.
        """
        s = re.sub(r"basis,\s*region_code", "basis", s)
        s = re.sub(r"basis,\s*66\s+FROM", "basis FROM", s)
        return s

    def _norm(s: str) -> str:
        return _strip_sql(
            _normalise_ppm2(
                _normalise_bucket_expr(
                    _drop_freshness_guard(
                        _drop_region_code_column(
                            _drop_area_not_null_guard(_drop_city_guard(_drop_segment_guard(s)))
                        )
                    )
                )
            )
        )

    assert _norm(seed_derivation) in _norm(_REDERIVE_SQL)


def test_ask_side_buckets_by_area_not_rooms() -> None:
    """#2620: ask_side buckets listings by AREA (same CASE as deals), not listings.rooms.

    Root cause of the ratio>1 bug in bucket "4+": deals.rooms is synthetic (derived from
    area_m2 at import time, deploy/import-rosreestr.sh — Rosreestr doesn't report room
    counts), while listings.rooms is a REAL room count. Comparing a synthetic area-bucket
    to a real-rooms-bucket mixed two different classifications — measured on prod
    (2026-08, #2620): 23-55% of listings migrate to a different bucket depending on which
    classification is used, not just in the "4+" bucket (which was ALSO truncated: deals
    caps rooms at 4 via ELSE 4, listings.rooms does not).

    Falsifiable: this assert is FALSE on the pre-#2620 code (ask_side bucketed by
    LEAST(GREATEST(rooms, 0), 4), same as deal_side) and TRUE once ask_side switches to
    _AREA_ROOMS_BUCKET_SQL.
    """
    assert ratio_mod._AREA_ROOMS_BUCKET_SQL.startswith("CASE WHEN area_m2 < 30 THEN 0")
    assert ratio_mod._AREA_ROOMS_BUCKET_SQL in _REDERIVE_SQL

    _a = _REDERIVE_SQL.index("ask_side AS")
    _b = _REDERIVE_SQL.index("per_bucket AS")
    ask_side_block = _REDERIVE_SQL[_a:_b]
    assert ratio_mod._AREA_ROOMS_BUCKET_SQL in ask_side_block
    assert "LEAST(GREATEST(rooms" not in ask_side_block

    # deal_side is UNCHANGED — deals.rooms is already the synthetic area bucket, there is
    # no separate "real rooms" column to prefer instead (see docstring above).
    _c = _REDERIVE_SQL.index("deal_side AS")
    deal_side_block = _REDERIVE_SQL[_c:_a]
    assert "LEAST(GREATEST(rooms, 0), 4)" in deal_side_block


# ── Counters query ────────────────────────────────────────────────────────────


def test_counters_query_returns_three_counters() -> None:
    assert "rows_written" in _COUNTERS_SQL
    assert "per_rooms_rows" in _COUNTERS_SQL
    assert "used_global_fallback" in _COUNTERS_SQL
    assert "FILTER (WHERE basis = 'per_rooms')" in _COUNTERS_SQL
    assert "FILTER (WHERE rooms_bucket = -1)" in _COUNTERS_SQL
    assert "FROM asking_to_sold_ratios" in _COUNTERS_SQL


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


def test_sql_uses_no_double_colon_bind_casts() -> None:
    """psycopg v3: never :param::type. Literal ::-casts on expressions are fine."""
    assert not re.search(r":\w+::", _ALL_SQL)


# ── Task return / finalisation contract ───────────────────────────────────────


def test_task_returns_counters_and_finalises_run() -> None:
    assert "mark_done" in _TASK_SRC
    assert "mark_failed" in _TASK_SRC
    assert "db.rollback()" in _TASK_SRC
    assert "raise" in _TASK_SRC  # re-raise on failure — no silent swallow


# ── Migration 082 ─────────────────────────────────────────────────────────────


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


def test_migration_082_seeds_schedule_enabled_true_window_6_7() -> None:
    sql = _MIGRATION_082.read_text("utf-8")
    assert "'asking_to_sold_ratio_refresh'" in sql
    assert "'{}'::jsonb" in sql
    assert "ON CONFLICT (source) DO NOTHING" in sql
    # enabled=true (SAFE — pure internal DB), window 6..7 UTC (after rosreestr 04:00-06:00).
    assert "true,                                        -- SAFE" in sql
    insert_block = sql.split("INSERT INTO scrape_schedules")[1]
    assert "false" not in insert_block
    # window_start_hour=6, window_end_hour=7 — assert the two literals appear post-INSERT.
    flat = re.sub(r"[ \t]+", " ", insert_block)
    assert "\n 6,\n 7,\n" in flat or ("6," in flat and "7," in flat)
    # next_run_at = tomorrow + 06:00 UTC.
    assert "make_interval(hours => 6)" in sql
    assert "BEGIN;" in sql and "COMMIT;" in sql


def test_migration_082_updates_table_comment() -> None:
    sql = _MIGRATION_082.read_text("utf-8")
    assert "COMMENT ON TABLE scrape_schedules" in sql
    assert "asking_to_sold_ratio_refresh (#648)" in sql


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


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


class _FakeMappingResult:
    def __init__(self, row: dict[str, Any] | None) -> None:
        self._row = row

    def mappings(self) -> "_FakeMappingResult":
        return self

    def first(self) -> dict[str, Any] | None:
        return self._row


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

    #3512: execute call order is now EKB (DELETE + INSERT) followed by a
    DELETE+INSERT pair per _OTHER_REGION_CODES region, then ONE counters SELECT
    at the very end — identified by statement IDENTITY (`stmt is ratio_mod._COUNTERS_SQL`),
    not by a hardcoded call index, so the fake stays correct if the region registry grows.
    """

    def __init__(
        self,
        counters_row: dict[str, Any],
    ) -> None:
        self._counters_row = counters_row
        self.executed: list[Any] = []
        self.committed = False
        self.rolled_back = False

    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeMappingResult:
        self.executed.append((stmt, params))
        if stmt is ratio_mod._COUNTERS_SQL:
            return _FakeMappingResult(self._counters_row)
        return _FakeMappingResult(None)

    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:
    """recompute_asking_to_sold_ratios maps the counters row into its return + marks done.

    #3512: FakeDB now sees 2 statements for EKB + 2 per _OTHER_REGION_CODES region +
    1 final COUNTERS SELECT (counters are still the 3 legacy keys, #2002).
    """
    marked: dict[str, Any] = {}
    monkeypatch.setattr(
        ratio_mod.runs_mod,
        "mark_done",
        lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
    )
    monkeypatch.setattr(ratio_mod.runs_mod, "mark_failed", lambda *a, **k: None)

    db = _FakeDB(
        counters_row={"rows_written": 4, "per_rooms_rows": 3, "used_global_fallback": 1},
    )
    out = ratio_mod.recompute_asking_to_sold_ratios(db, run_id=99)  # type: ignore[arg-type]

    expected = {
        "rows_written": 4,
        "per_rooms_rows": 3,
        "used_global_fallback": 1,
    }
    assert out == expected
    assert db.committed is True
    # EKB (DELETE+INSERT) + 2 per other region + 1 COUNTERS SELECT.
    expected_calls = 2 + 2 * len(ratio_mod._OTHER_REGION_CODES) + 1
    assert len(db.executed) == expected_calls
    assert db.executed[-1][0] is ratio_mod._COUNTERS_SQL
    assert marked["run_id"] == 99
    assert marked["counters"] == expected


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(ratio_mod.runs_mod, "mark_done", lambda *a, **k: None)
    monkeypatch.setattr(
        ratio_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) -> _FakeMappingResult:
            raise RuntimeError("boom")

    db = _BoomDB(counters_row={})
    with pytest.raises(RuntimeError, match="boom"):
        ratio_mod.recompute_asking_to_sold_ratios(db, run_id=7)  # type: ignore[arg-type]
    assert db.rolled_back is True
    assert failed["run_id"] == 7


# ── Migration 098 band / settings-default consistency ────────────────────────


def test_migration_098_band_matches_settings_default() -> None:
    """Migration 098 seed hardcodes ppm² band [30000, 1_200_000].

    These literals MUST stay in sync with:
      - app/tasks/asking_to_sold_ratio._PPM2_MIN  (lower bound)
      - app.core.config.Settings.asking_ratio_ppm2_max default (upper bound)

    If either default changes, the one-time migration seed diverges from the
    daily-refresh output — this test is the regression guard.
    """
    from app.core.config import Settings
    from app.tasks.asking_to_sold_ratio import _PPM2_MIN

    # Lower bound — matches the 30000 literal in migration 098.
    assert _PPM2_MIN == 30_000, (
        f"_PPM2_MIN changed ({_PPM2_MIN}); update migration 098 seed literals to match"
    )
    # Upper bound — matches the 1200000 literal in migration 098.
    assert Settings().asking_ratio_ppm2_max == 1_200_000, (
        "asking_ratio_ppm2_max default changed; update migration 098 seed literals to match"
    )


# ── area_bucket() Python twin matches _AREA_ROOMS_BUCKET_SQL (#2620) ─────────


def test_area_bucket_matches_sql_boundaries() -> None:
    """area_bucket() has IDENTICAL boundaries to _AREA_ROOMS_BUCKET_SQL (#2620).

    Two representations of one truth (see module comment: shell/import-rosreestr.sh → SQL
    _AREA_ROOMS_BUCKET_SQL → Python area_bucket()). If SQL (computes the ratio in ask_side)
    and Python (applies it in estimator.py) drift apart, the #2620 mismatch is silently
    reintroduced. Boundaries are parsed straight out of the SQL string, not hardcoded
    independently, so this fails the moment either one is edited without the other.
    """
    sql_bounds = [int(n) for n in re.findall(r"area_m2 < (\d+)", ratio_mod._AREA_ROOMS_BUCKET_SQL)]
    assert sql_bounds == [30, 44, 62, 85]

    cases = {
        0: 0,
        29: 0,
        29.99: 0,
        30: 1,
        43: 1,
        43.99: 1,
        44: 2,
        61: 2,
        61.99: 2,
        62: 3,
        84: 3,
        84.99: 3,
        85: 4,
        200: 4,
    }
    for area, expected in cases.items():
        assert ratio_mod.area_bucket(area) == expected, f"area={area} -> expected {expected}"

    for i, bound in enumerate(sql_bounds):
        assert ratio_mod.area_bucket(bound - 0.01) == i
        assert ratio_mod.area_bucket(bound) == i + 1


# ── #3512: per-region ratio (migration 304) ───────────────────────────────────

_MIGRATION_304 = _SQL_DIR / "304_asking_to_sold_ratios_region.sql"


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


def test_migration_304_adds_region_code_and_composite_pk() -> None:
    sql = _MIGRATION_304.read_text("utf-8")
    assert "ADD COLUMN IF NOT EXISTS region_code int NOT NULL DEFAULT 66" in sql
    assert "PRIMARY KEY (region_code, rooms_bucket, district)" in sql
    # PK-guard идемпотентен (тот же паттерн, что 298_deal_city_price_bands_region.sql).
    assert "pg_get_constraintdef" in sql
    assert "DROP CONSTRAINT asking_to_sold_ratios_pkey" in sql
    assert "BEGIN;" in sql and "COMMIT;" in sql
    assert "SET LOCAL lock_timeout" in sql


def test_migration_304_uses_psycopg_safe_sql() -> None:
    sql = _MIGRATION_304.read_text("utf-8")
    assert not re.search(r":\w+::", sql)


def test_other_region_codes_matches_registry_minus_ekb() -> None:
    """_OTHER_REGION_CODES — весь реестр регионов (regions_mod.REGIONS) минус 66.

    Новый регион в реестре автоматически подхватывается пересчётом ratio (#3512) —
    без правки этого файла.
    """
    from app.services import regions as regions_mod

    expected = tuple(sorted(code for code in regions_mod.REGIONS if code != 66))
    assert ratio_mod._OTHER_REGION_CODES == expected
    assert 66 not in ratio_mod._OTHER_REGION_CODES
    assert 77 in ratio_mod._OTHER_REGION_CODES  # Москва — уже в реестре (#3051)


def test_rederive_sql_region_scopes_by_region_code_not_city() -> None:
    """_REDERIVE_SQL_REGION скоупит ОБЕ стороны по region_code, БЕЗ городской квоты ЕКБ.

    В отличие от _REDERIVE_SQL (byte-identical historical EKB derivation), генерик-путь
    для остальных регионов не несёт `city ILIKE` — эта квота была костылём конкретно
    ЕКБ-исторического asking-покрытия (#C2), не общим правилом (#3512).
    """
    assert "city ILIKE" not in _REDERIVE_SQL_REGION
    assert _REDERIVE_SQL_REGION.count("region_code = CAST(:region_code AS int)") >= 4
    assert "FROM deals" in _REDERIVE_SQL_REGION
    assert "FROM listings" in _REDERIVE_SQL_REGION
    # Тот же порог 30/30 и то же 12-мес окно, что и у ЕКБ-деривации (переиспользуется,
    # не изобретается заново — требование задачи).
    assert "d.n_deals >= 30" in _REDERIVE_SQL_REGION
    assert "a.n_listings >= 30" in _REDERIVE_SQL_REGION
    assert "deal_date >= CURRENT_DATE - INTERVAL '12 months'" in _REDERIVE_SQL_REGION
    # INSERT пишет параметризованный регион, а не литерал 66.
    assert "CAST(:region_code AS int) FROM global_row" in _REDERIVE_SQL_REGION
    assert "CAST(:region_code AS int) FROM per_bucket" in _REDERIVE_SQL_REGION


def test_delete_sql_region_scopes_by_region_code() -> None:
    flat = re.sub(r"\s+", " ", _DELETE_SQL_REGION).strip()
    assert "DELETE FROM asking_to_sold_ratios" in flat
    assert "WHERE region_code = CAST(:region_code AS int) AND district = ''" in flat


def test_rederive_sql_writes_region_code_66_literal() -> None:
    """_REDERIVE_SQL (ЕКБ-путь) теперь пишет region_code=66 явно в новую колонку."""
    assert "region_code" in _REDERIVE_SQL
    assert "basis, 66 FROM global_row" in _REDERIVE_SQL
    assert "basis, 66 FROM per_bucket" in _REDERIVE_SQL
    # Числа для 66 не должны были поменяться: городская квота ЕКБ (_ASKING_CITY_PATTERN)
    # осталась нетронутой в обоих deal-CTE и обоих ask-CTE.
    assert _REDERIVE_SQL.count("AND city ILIKE :asking_city") == 2
    assert _REDERIVE_SQL.count("AND (city IS NULL OR city ILIKE :asking_city)") == 2


def test_task_loops_other_regions_between_ekb_and_counters() -> None:
    """recompute_asking_to_sold_ratios: EKB derive -> other-regions loop -> counters."""
    ekb_pos = _TASK_SRC.index("_REDERIVE_SQL,")
    loop_pos = _TASK_SRC.index("_OTHER_REGION_CODES")
    region_delete_pos = _TASK_SRC.index("_DELETE_SQL_REGION")
    region_insert_pos = _TASK_SRC.index("_REDERIVE_SQL_REGION,")
    counters_pos = _TASK_SRC.index("_COUNTERS_SQL")
    assert ekb_pos < loop_pos < region_delete_pos < region_insert_pos < counters_pos