diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 4945ef04..8a3cfd2e 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -14,6 +14,10 @@ Sources: - rosreestr_dkp_import → import_rosreestr_dkp (this module, #563) - listing_source_snapshot → snapshot_listing_sources (tasks/listing_source_snapshot.py, #570; pure internal DB snapshot — no external calls, enabled by default) + - asking_to_sold_ratio_refresh → recompute_asking_to_sold_ratios + (tasks/asking_to_sold_ratio.py, #648 Stage 4; pure internal DB + re-derivation of the asking→sold ratios — no external calls, + enabled by default; window after rosreestr_dkp_import) """ from __future__ import annotations @@ -405,6 +409,45 @@ async def trigger_listing_source_snapshot_run( return run_id +async def trigger_asking_to_sold_ratio_run( + db: Session, schedule_row: dict[str, Any] +) -> int | None: + """Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task). + + Daily TRUE-MIRROR refresh асking→sold коэффициентов (#648 Stage 4): DELETE WHERE + district='' + re-derive INSERT (та же 080-derivation) в одной транзакции, чтобы ratio + не устаревал по мере ночного импорта ДКП. Mirror trigger_listing_source_snapshot_run: + задача синхронная (чистый internal DB re-derive, БЕЗ внешних HTTP-вызовов / анти-бота), + поэтому гоняем её в run_in_executor. SAFE to enable — schedule seed enabled=true (082), + в отличие от avito/yandex sweep (dormant). + + Returns run_id (или None если skip — есть running run). + """ + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + async def _run() -> None: + run_db = SessionLocal() + try: + # recompute_asking_to_sold_ratios is synchronous (DB-only, no async HTTP) + from app.tasks.asking_to_sold_ratio import recompute_asking_to_sold_ratios + + loop = asyncio.get_event_loop() + await loop.run_in_executor(None, recompute_asking_to_sold_ratios, run_db, run_id) + except Exception: + logger.exception( + "scheduler: recompute_asking_to_sold_ratios crashed run_id=%d", run_id + ) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info("scheduler: triggered asking_to_sold_ratio_refresh run_id=%d", run_id) + return run_id + + def import_rosreestr_dkp( db: Session, run_id: int, @@ -676,6 +719,8 @@ async def scheduler_loop() -> None: await trigger_rosreestr_dkp_run(db, sch) elif source == "listing_source_snapshot": await trigger_listing_source_snapshot_run(db, sch) + elif source == "asking_to_sold_ratio_refresh": + await trigger_asking_to_sold_ratio_run(db, sch) else: logger.warning("scheduler: unknown source=%s, skip", source) finally: diff --git a/tradein-mvp/backend/app/tasks/asking_to_sold_ratio.py b/tradein-mvp/backend/app/tasks/asking_to_sold_ratio.py new file mode 100644 index 00000000..aec1e8c3 --- /dev/null +++ b/tradein-mvp/backend/app/tasks/asking_to_sold_ratio.py @@ -0,0 +1,223 @@ +"""Daily recompute of the asking→sold correction ratios (#648 Stage 4). + +ПРОБЛЕМА: asking_to_sold_ratios (migration 080) засеяна один раз derivation-CTE +(медиана SOLD ДКП за 12 мес vs медиана ASKING активных listings, per-rooms + global -1 +fallback). Оценщик (estimator.py, Stage 3) читает её per-estimate (cache 300s) и домножает +asking-median прогноз на sold/asking. По мере ночного импорта новых ДКП-сделок +(rosreestr_dkp_import) ratio устаревает — нужен периодический пересчёт по тому же derivation. + +TRUE-MIRROR REFRESH (флаг database-expert): 080-seed использовал ON CONFLICT DO UPDATE, +который при повторном прогоне ОБНОВЛЯЕТ существующие строки, но НЕ удаляет per-rooms строки +бакетов, упавших ниже порога 30/30 на более позднем прогоне (StaLE rows). Поэтому refresh +сначала DELETE FROM asking_to_sold_ratios WHERE district = '' (все строки #648), затем +заново гоняет ту же 080-derivation INSERT...SELECT. DELETE+INSERT в ОДНОЙ транзакции +(атомарно — таблица никогда не пуста mid-refresh). После commit таблица == свежий re-seed. + +Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается in-app +scheduler'ом через trigger_asking_to_sold_ratio_run() (scheduler.py), по образцу +snapshot_listing_sources / import_rosreestr_dkp (sync task в run_in_executor). + +Окно расписания 06:00-07:00 UTC — ПОСЛЕ rosreestr_dkp_import (04:00-06:00 UTC), чтобы +refresh потреблял свежие ДКП-сделки того же дня. + +SQL derivation ниже — БАЙТ-В-БАЙТ та же логика, что seed в data/sql/080_asking_to_sold_ratios.sql +(deal_side / ask_side / per_bucket + deal_global / ask_global / global_row: трейлинг-12мес +окно, ppm²-полоса [30000,600000], бакет LEAST(GREATEST(rooms,0),4), порог n_deals>=30 AND +n_listings>=30 для per_rooms, global -1 строка всегда). ON CONFLICT убран — DELETE идёт первым, +конфликтов нет (повторный прогон в одной tx невозможен, refresh = re-seed по семантике). +""" + +from __future__ import annotations + +import logging + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.services import scrape_runs as runs_mod + +logger = logging.getLogger(__name__) + +# ── True-mirror cleanup: drop all #648 rows before re-derivation ────────────── +# district = '' — все строки #648 (district зарезервирован под #647, пока всегда ''). +# Удаляем ПЕРЕД re-derive, чтобы бакеты, упавшие ниже порога 30/30, не оставались +# stale (ON CONFLICT DO UPDATE такие строки бы не тронул). В одной транзакции с INSERT. +_DELETE_SQL = text( + """ + DELETE FROM asking_to_sold_ratios + WHERE district = '' + """ +) + +# ── Derivation + re-seed (БАЙТ-В-БАЙТ из 080, ON CONFLICT убран — DELETE идёт первым) ── +# deal_side / ask_side / per_bucket + deal_global / ask_global / global_row: +# sold_median = percentile_cont(0.5) по deals.price_per_m2 (source='rosreestr', +# ppm² ∈ [30000,600000], deal_date >= CURRENT_DATE − 12 months), +# бакет LEAST(GREATEST(rooms,0),4). +# ask_median = percentile_cont(0.5) по listings.price_per_m2 (is_active, та же ppm²-полоса). +# per_rooms строки — только при n_deals>=30 AND n_listings>=30 AND ask>0 AND sold>0. +# global -1 строка (basis='global_fallback') — всегда (если ask>0 AND sold>0). window_months=12. +# Литералы (порог/окно/полоса) — без bind-параметров, поэтому CAST(:x AS type) не нужен. +_REDERIVE_SQL = text( + """ + WITH + -- SOLD медианы по бакетам комнат за трейлинг-12мес (ДКП Росреестра). + deal_side AS ( + SELECT + LEAST(GREATEST(rooms, 0), 4) AS rooms_bucket, + percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS sold_median, + COUNT(*) AS n_deals + FROM deals + WHERE source = 'rosreestr' + AND rooms IS NOT NULL + AND price_per_m2 BETWEEN 30000 AND 600000 + AND deal_date >= CURRENT_DATE - INTERVAL '12 months' + GROUP BY LEAST(GREATEST(rooms, 0), 4) + ), + -- ASKING медианы по бакетам комнат среди ТЕКУЩИХ активных объявлений. + ask_side AS ( + SELECT + LEAST(GREATEST(rooms, 0), 4) AS rooms_bucket, + percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS ask_median, + COUNT(*) AS n_listings + FROM listings + WHERE is_active + AND rooms IS NOT NULL + AND price_per_m2 BETWEEN 30000 AND 600000 + GROUP BY LEAST(GREATEST(rooms, 0), 4) + ), + -- Per-rooms строки: только бакеты с обеими сторонами, прошедшие порог 30/30 и ask>0. + -- Тонкие бакеты (n<30) сюда НЕ попадают → estimator делает fallback на -1. + per_bucket AS ( + SELECT + d.rooms_bucket, + ''::text AS district, + (d.sold_median / a.ask_median)::numeric AS ratio, + round(d.sold_median)::bigint AS sold_median, + round(a.ask_median)::bigint AS ask_median, + d.n_deals::int AS n_deals, + a.n_listings::int AS n_listings, + 12 AS window_months, + 'per_rooms'::text AS basis + FROM deal_side d + JOIN ask_side a USING (rooms_bucket) + WHERE d.n_deals >= 30 + AND a.n_listings >= 30 + AND a.ask_median IS NOT NULL + AND a.ask_median > 0 + AND d.sold_median IS NOT NULL -- divide-safety (порог n_deals>=30 гарантирует) + AND d.sold_median > 0 + ), + -- SOLD медиана по ВСЕМ комнатам (без бакет-фильтра) за трейлинг-12мес — для global row. + deal_global AS ( + SELECT + percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS sold_median, + COUNT(*) AS n_deals + FROM deals + WHERE source = 'rosreestr' + AND rooms IS NOT NULL + AND price_per_m2 BETWEEN 30000 AND 600000 + AND deal_date >= CURRENT_DATE - INTERVAL '12 months' + ), + -- ASKING медиана по ВСЕМ активным listings (без бакет-фильтра) — для global row. + ask_global AS ( + SELECT + percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS ask_median, + COUNT(*) AS n_listings + FROM listings + WHERE is_active + AND rooms IS NOT NULL + AND price_per_m2 BETWEEN 30000 AND 600000 + ), + -- Global fallback строка rooms_bucket=-1 (пишется всегда, если ask>0). + global_row AS ( + SELECT + -1 AS rooms_bucket, + ''::text AS district, + (d.sold_median / a.ask_median)::numeric AS ratio, + round(d.sold_median)::bigint AS sold_median, + round(a.ask_median)::bigint AS ask_median, + d.n_deals::int AS n_deals, + a.n_listings::int AS n_listings, + 12 AS window_months, + 'global_fallback'::text AS basis + FROM deal_global d + CROSS JOIN ask_global a + WHERE a.ask_median IS NOT NULL + AND a.ask_median > 0 + AND d.sold_median IS NOT NULL -- защита от пустого окна сделок (иначе ratio=NULL) + AND d.sold_median > 0 + ) + INSERT INTO asking_to_sold_ratios ( + rooms_bucket, district, ratio, sold_median, ask_median, + n_deals, n_listings, window_months, basis + ) + SELECT rooms_bucket, district, ratio, sold_median, ask_median, + n_deals, n_listings, window_months, basis FROM global_row + UNION ALL + SELECT rooms_bucket, district, ratio, sold_median, ask_median, + n_deals, n_listings, window_months, basis FROM per_bucket + """ +) + +# ── Post-insert counters ────────────────────────────────────────────────────── +# Считываем итог из таблицы (всё ещё в той же транзакции — до commit): сколько строк +# записано всего, сколько per_rooms, был ли использован global -1 fallback. +_COUNTERS_SQL = text( + """ + SELECT + COUNT(*) AS rows_written, + COUNT(*) FILTER (WHERE basis = 'per_rooms') AS per_rooms_rows, + COUNT(*) FILTER (WHERE rooms_bucket = -1) AS used_global_fallback + FROM asking_to_sold_ratios + WHERE district = '' + """ +) + + +def recompute_asking_to_sold_ratios(db: Session, run_id: int) -> dict[str, int]: + """Пересчитать asking_to_sold_ratios (TRUE-MIRROR refresh, #648 Stage 4). + + Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources). + В ОДНОЙ транзакции (атомарно — таблица никогда не пуста mid-refresh): + 1. DELETE FROM asking_to_sold_ratios WHERE district = '' — снести stale-строки. + 2. Заново прогнать 080-derivation INSERT...SELECT (per_rooms при 30/30 + global -1). + Затем counters из таблицы, commit, mark_done. Семантика == re-seed миграции 080. + + Финализирует scrape_runs (mark_done / mark_failed) и пишет counters. + + Returns {"rows_written": N, "per_rooms_rows": M, "used_global_fallback": 0|1}. + """ + counters: dict[str, int] = { + "rows_written": 0, + "per_rooms_rows": 0, + "used_global_fallback": 0, + } + try: + # DELETE + re-derive INSERT в одной транзакции (НЕ коммитим между ними — + # таблица не должна остаться пустой, если INSERT упадёт). + db.execute(_DELETE_SQL) + db.execute(_REDERIVE_SQL) + + row = db.execute(_COUNTERS_SQL).mappings().first() + if row is not None: + counters["rows_written"] = int(row["rows_written"] or 0) + counters["per_rooms_rows"] = int(row["per_rooms_rows"] or 0) + counters["used_global_fallback"] = int(row["used_global_fallback"] or 0) + + db.commit() + runs_mod.mark_done(db, run_id, counters) + logger.info( + "recompute_asking_to_sold_ratios run_id=%d done: " + "rows_written=%d per_rooms_rows=%d used_global_fallback=%d", + run_id, + counters["rows_written"], + counters["per_rooms_rows"], + counters["used_global_fallback"], + ) + return counters + except Exception as exc: + logger.exception("recompute_asking_to_sold_ratios run_id=%d failed", run_id) + db.rollback() + runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters) + raise diff --git a/tradein-mvp/backend/data/sql/082_scrape_schedules_seed_ratio_refresh.sql b/tradein-mvp/backend/data/sql/082_scrape_schedules_seed_ratio_refresh.sql new file mode 100644 index 00000000..f701589f --- /dev/null +++ b/tradein-mvp/backend/data/sql/082_scrape_schedules_seed_ratio_refresh.sql @@ -0,0 +1,53 @@ +-- 082_scrape_schedules_seed_ratio_refresh.sql +-- #648 Stage 4 — seed scrape_schedules row для daily asking→sold ratio refresh. +-- +-- ПРОБЛЕМА: asking_to_sold_ratios (migration 080) засеяна один раз; оценщик (estimator.py, +-- Stage 3) читает её per-estimate (cache 300s) и домножает asking-median прогноз на +-- sold/asking. По мере ночного импорта новых ДКП-сделок (rosreestr_dkp_import) ratio +-- устаревает. Stage 4 = TRUE-MIRROR refresh-задача recompute_asking_to_sold_ratios() +-- (app/tasks/asking_to_sold_ratio.py): DELETE WHERE district='' + re-derive INSERT +-- (та же 080-derivation) в одной транзакции, по расписанию. Запускается in-app +-- scheduler'ом (source='asking_to_sold_ratio_refresh', trigger_asking_to_sold_ratio_run). +-- +-- enabled = true — БЕЗОПАСНО включать сразу: pure-internal DB re-derivation, никаких +-- внешних HTTP-вызовов / анти-бота (в отличие от avito/yandex sweep, которые seed'ятся +-- DORMANT). DELETE+INSERT атомарны (одна транзакция) — таблица не остаётся пустой. +-- Окно 06:00-07:00 UTC — ПОСЛЕ rosreestr_dkp_import (04:00-06:00 UTC, см. 072), чтобы +-- refresh потреблял свежие ДКП-сделки того же дня (а не отставал на сутки). +-- next_run_at = завтрашнее наступление окна (tomorrow + 06:00 UTC), иначе при NULL +-- get_due_schedules() выстрелит сразу после деплоя. AT TIME ZONE 'UTC' фиксирует +-- wall-clock час в UTC независимо от session TimeZone GUC (образец 078/079). +-- +-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)), +-- 080_asking_to_sold_ratios.sql (таблица, derivation которую переиспользует refresh). +-- Idempotent: ON CONFLICT (source) DO NOTHING — безопасно запускать повторно. +-- Apply after: 081_trade_in_estimates_expected_sold.sql + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'asking_to_sold_ratio_refresh', + true, -- SAFE: pure internal DB, no external calls + 6, + 7, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 6)) AT TIME ZONE 'UTC', + '{}'::jsonb +) +ON CONFLICT (source) DO NOTHING; + +COMMENT ON TABLE scrape_schedules IS + 'In-app scheduler config (заменяет cron-script setup). ' + 'Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), ' + 'cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), ' + 'asking_to_sold_ratio_refresh (#648).'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_asking_to_sold_ratio.py b/tradein-mvp/backend/tests/test_asking_to_sold_ratio.py new file mode 100644 index 00000000..4e68adf8 --- /dev/null +++ b/tradein-mvp/backend/tests/test_asking_to_sold_ratio.py @@ -0,0 +1,313 @@ +"""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 scheduler wiring (trigger fn + dispatch +branch) and 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.services import scheduler +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 + assert "price_per_m2 BETWEEN 30000 AND 600000" 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 _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. + """ + 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] + + assert _strip_sql(seed_derivation) in _strip_sql(_REDERIVE_SQL) + + +# ── 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 + + +# ── Scheduler wiring ────────────────────────────────────────────────────────── + + +def test_trigger_fn_exists_and_is_async() -> None: + fn = getattr(scheduler, "trigger_asking_to_sold_ratio_run", None) + assert fn is not None, "trigger_asking_to_sold_ratio_run missing from scheduler" + assert inspect.iscoroutinefunction(fn) + + +def test_dispatch_branch_wired() -> None: + """scheduler_loop dispatches source='asking_to_sold_ratio_refresh' to the trigger.""" + loop_src = inspect.getsource(scheduler.scheduler_loop) + assert 'source == "asking_to_sold_ratio_refresh"' in loop_src + assert "trigger_asking_to_sold_ratio_run(db, sch)" in loop_src + + +def test_trigger_runs_sync_task_in_executor() -> None: + """Sync DB-only task → run_in_executor (mirror snapshot), not a bare await.""" + trig_src = inspect.getsource(scheduler.trigger_asking_to_sold_ratio_run) + assert "run_in_executor(None, recompute_asking_to_sold_ratios" in trig_src + assert "_claim_run(db, schedule_row)" in trig_src + + +# ── 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. + + The third execute (counters query) returns the seeded counter row; the first two + (DELETE, INSERT) return an empty result. + """ + + 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. + 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.""" + 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] + + assert out == {"rows_written": 4, "per_rooms_rows": 3, "used_global_fallback": 1} + assert db.committed is True + # Three statements: DELETE, re-derive INSERT, counters SELECT. + assert len(db.executed) == 3 + assert marked["run_id"] == 99 + assert marked["counters"] == {"rows_written": 4, "per_rooms_rows": 3, "used_global_fallback": 1} + + +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