bot-backend
4d93e5a1aa
asking_to_sold_ratio фильтровал сделки по хардкоду price_per_m2 BETWEEN 30000 AND 600000 (4 CTE) → премиум ЕКБ >600k выпадал → ratio из global-fallback → bias выкупной цены премиума. Все 4 хардкода → BETWEEN :ppm2_min AND :ppm2_max (bind-params, не f-string). settings.asking_ratio_ppm2_max default 1_200_000 (env ASKING_RATIO_PPM2_MAX), _PPM2_MIN=30_000. Docstring обновлён. NB: точный потолок верифицировать QA против SELECT max(price_per_m2) FROM deals WHERE source='rosreestr' (worker без live-DB; 1.2M — запас, env-override без деплоя). pytest -k asking 21 pass, ruff clean.
326 lines
14 KiB
Python
326 lines
14 KiB
Python
"""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
|
|
# 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 _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.
|
|
"""
|
|
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
|
|
|
|
assert _strip_sql(_normalise_ppm2(seed_derivation)) in _strip_sql(
|
|
_normalise_ppm2(_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
|