"""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) _COUNTERS_SQL = str(ratio_mod._COUNTERS_SQL.text) _ALL_SQL = _DELETE_SQL + "\n" + _REDERIVE_SQL + "\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 district='' before the INSERT. 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. """ flat_del = re.sub(r"\s+", " ", _DELETE_SQL).strip() assert "DELETE FROM asking_to_sold_ratios" in flat_del assert "WHERE 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. """ 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 _norm(s: str) -> str: return _strip_sql( _normalise_ppm2( _normalise_bucket_expr( _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. Execute call order: 1. DELETE asking_to_sold_ratios 2. REDERIVE INSERT asking_to_sold_ratios 3. COUNTERS SELECT (returns counters_row) """ 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)) # 3rd call is the counters SELECT (asking_to_sold_ratios). if len(self.executed) == 3: 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. #2002: tiered refresh removed — counters are the 3 legacy keys, FakeDB has 3 execute() calls (DELETE + REDERIVE INSERT + COUNTERS SELECT). """ 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 # Three statements: DELETE + REDERIVE INSERT + COUNTERS SELECT (legacy per-rooms). assert len(db.executed) == 3 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