From e925a0ba6aeef2d94b67dc031f371c0ca18dbb1b Mon Sep 17 00:00:00 2001 From: Light1YT Date: Fri, 29 May 2026 12:18:04 +0000 Subject: [PATCH] =?UTF-8?q?feat(tradein):=20per-source=20listing=20price?= =?UTF-8?q?=20history=20=E2=80=94=20schema=20+=20daily=20snapshot=20(#570)?= =?UTF-8?q?=20(#646)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tradein-mvp/backend/app/services/scheduler.py | 39 +++ .../app/tasks/listing_source_snapshot.py | 141 +++++++++ .../data/sql/079_listing_source_history.sql | 140 +++++++++ .../tests/test_listing_source_snapshot.py | 281 ++++++++++++++++++ 4 files changed, 601 insertions(+) create mode 100644 tradein-mvp/backend/app/tasks/listing_source_snapshot.py create mode 100644 tradein-mvp/backend/data/sql/079_listing_source_history.sql create mode 100644 tradein-mvp/backend/tests/test_listing_source_snapshot.py diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 01c4759b..4945ef04 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -12,6 +12,8 @@ Sources: - yandex_city_sweep → run_yandex_city_sweep (scrape_pipeline.py, #561; shipped DORMANT) - cian_history_backfill → run_cian_backfill (this module, #560) - 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) """ from __future__ import annotations @@ -368,6 +370,41 @@ async def trigger_rosreestr_dkp_run(db: Session, schedule_row: dict[str, Any]) - return run_id +async def trigger_listing_source_snapshot_run( + db: Session, schedule_row: dict[str, Any] +) -> int | None: + """Создать scrape_runs + launch snapshot_listing_sources в executor (sync DB-only task). + + Per-source price-history daily snapshot (#570). Mirror trigger_rosreestr_dkp_run: + задача синхронная (чистый internal DB-snapshot, БЕЗ внешних HTTP-вызовов / анти-бота), + поэтому гоняем её в run_in_executor. SAFE to enable — schedule seed enabled=true (079), + в отличие от 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: + # snapshot_listing_sources is synchronous (DB-only, no async HTTP) + from app.tasks.listing_source_snapshot import snapshot_listing_sources + + loop = asyncio.get_event_loop() + await loop.run_in_executor(None, snapshot_listing_sources, run_db, run_id) + except Exception: + logger.exception("scheduler: snapshot_listing_sources 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 listing_source_snapshot run_id=%d", run_id) + return run_id + + def import_rosreestr_dkp( db: Session, run_id: int, @@ -637,6 +674,8 @@ async def scheduler_loop() -> None: await trigger_cian_backfill_run(db, sch) elif source == "rosreestr_dkp_import": await trigger_rosreestr_dkp_run(db, sch) + elif source == "listing_source_snapshot": + await trigger_listing_source_snapshot_run(db, sch) else: logger.warning("scheduler: unknown source=%s, skip", source) finally: diff --git a/tradein-mvp/backend/app/tasks/listing_source_snapshot.py b/tradein-mvp/backend/app/tasks/listing_source_snapshot.py new file mode 100644 index 00000000..2516fd02 --- /dev/null +++ b/tradein-mvp/backend/app/tasks/listing_source_snapshot.py @@ -0,0 +1,141 @@ +"""Daily per-source snapshot writer (#570). + +Берёт текущее состояние listing_sources (последний снимок на canonical listing × source) +и пишет ежедневный ряд в listing_source_snapshots + change-log price_change в +listing_source_events. Так история per-source цены копится СРАЗУ, независимо от +(сейчас DORMANT) скраперов — см. шапку data/sql/079_listing_source_history.sql. + +Задача синхронная (DB-only, никаких внешних HTTP-вызовов) — запускается in-app +scheduler'ом через trigger_listing_source_snapshot_run() (scheduler.py), +по образцу import_rosreestr_dkp (sync task в run_in_executor). + +Вся работа — два set-based SQL statement'а (snapshot upsert + event-diff CTE), +никакого row-by-row Python: 18 355 строк обслуживаются одним INSERT … SELECT каждый. +""" + +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__) + +# Окно свежести: источник считается активным, если last_seen_at не старше N дней. +FRESHNESS_WINDOW_DAYS = 7 + +# ── Daily snapshot upsert ───────────────────────────────────────────────────── +# Снимок на (listing_source_id, CURRENT_DATE). ON CONFLICT → last-write-wins за день +# (повторный прогон в те же сутки перезаписывает снимок свежими значениями). +# is_active derived: last_seen_at в пределах окна свежести на момент снимка. +# payload_hash = md5(raw_payload::text) — ::text на колонке допустим (это не bind-param). +# run_id через CAST(:run_id AS bigint) — psycopg v3 (никогда :run_id::bigint). +_SNAPSHOT_SQL = text( + """ + INSERT INTO listing_source_snapshots ( + listing_source_id, snapshot_date, price_rub, is_active, + last_seen_at, payload_hash, observed_at, run_id + ) + SELECT + id, + CURRENT_DATE, + price_rub, + (last_seen_at > now() - make_interval(days => :freshness_days)) AS is_active, + last_seen_at, + md5(raw_payload::text), + now(), + CAST(:run_id AS bigint) + FROM listing_sources + ON CONFLICT (listing_source_id, snapshot_date) DO UPDATE SET + price_rub = EXCLUDED.price_rub, + is_active = EXCLUDED.is_active, + last_seen_at = EXCLUDED.last_seen_at, + payload_hash = EXCLUDED.payload_hash, + observed_at = EXCLUDED.observed_at, + run_id = EXCLUDED.run_id + """ +) + +# ── Event diff: price_change ────────────────────────────────────────────────── +# Для каждого источника сравниваем сегодняшнюю цену (snapshot_date = CURRENT_DATE) с +# самым свежим ПРЕДЫДУЩИМ снимком (snapshot_date < CURRENT_DATE). Если цена изменилась +# (обе NOT NULL, old <> 0) — пишем price_change. +# today — снимок за сегодня (только что записан _SNAPSHOT_SQL). +# prior — последний снимок строго ДО сегодня (DISTINCT ON … ORDER BY date DESC). +# Полностью set-based: один INSERT … SELECT по всем источникам, без Python-цикла. +# change_time = now() детерминирует UNIQUE(listing_source_id, change_time, event_type) +# в пределах прогона → ON CONFLICT DO NOTHING делает писатель идемпотентным. +_EVENT_DIFF_SQL = text( + """ + WITH today AS ( + SELECT listing_source_id, price_rub + FROM listing_source_snapshots + WHERE snapshot_date = CURRENT_DATE + ), + prior AS ( + SELECT DISTINCT ON (listing_source_id) + listing_source_id, price_rub + FROM listing_source_snapshots + WHERE snapshot_date < CURRENT_DATE + ORDER BY listing_source_id, snapshot_date DESC + ) + INSERT INTO listing_source_events ( + listing_source_id, change_time, event_type, price_rub, diff_percent + ) + SELECT + t.listing_source_id, + now(), + 'price_change', + t.price_rub, + round((t.price_rub - p.price_rub)::numeric / p.price_rub * 100, 4) + FROM today t + JOIN prior p ON p.listing_source_id = t.listing_source_id + WHERE t.price_rub IS NOT NULL + AND p.price_rub IS NOT NULL + AND p.price_rub <> 0 + AND t.price_rub <> p.price_rub + ON CONFLICT (listing_source_id, change_time, event_type) DO NOTHING + """ +) + + +def snapshot_listing_sources(db: Session, run_id: int) -> dict[str, int]: + """Записать дневной снимок listing_sources + price_change-события. + + Sync (вызывается scheduler-триггером в executor, как import_rosreestr_dkp). + Два set-based statement'а в одной транзакции: + 1. upsert снимка на (listing_source_id, CURRENT_DATE) — last-write-wins. + 2. diff сегодняшней цены против последнего предыдущего снимка → price_change-события. + + Финализирует scrape_runs (mark_done / mark_failed) и пишет counters. + + Returns {"snapshotted": N, "price_change_events": M}. + """ + counters: dict[str, int] = {"snapshotted": 0, "price_change_events": 0} + try: + snap_result = db.execute( + _SNAPSHOT_SQL, + {"freshness_days": FRESHNESS_WINDOW_DAYS, "run_id": run_id}, + ) + counters["snapshotted"] = snap_result.rowcount or 0 + + event_result = db.execute(_EVENT_DIFF_SQL) + counters["price_change_events"] = event_result.rowcount or 0 + + db.commit() + runs_mod.mark_done(db, run_id, counters) + logger.info( + "snapshot_listing_sources run_id=%d done: snapshotted=%d price_change_events=%d", + run_id, + counters["snapshotted"], + counters["price_change_events"], + ) + return counters + except Exception as exc: + logger.exception("snapshot_listing_sources 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/079_listing_source_history.sql b/tradein-mvp/backend/data/sql/079_listing_source_history.sql new file mode 100644 index 00000000..3155b1db --- /dev/null +++ b/tradein-mvp/backend/data/sql/079_listing_source_history.sql @@ -0,0 +1,140 @@ +-- 079_listing_source_history.sql +-- #570 — per-source listing price history. +-- +-- ПРОБЛЕМА: listing_sources хранит ТОЛЬКО последний снимок на (canonical listing × source) +-- (price_rub / last_seen_at / raw_payload перезаписываются при каждом скрейпе) — истории нет. +-- Скраперы сейчас DORMANT (avito/yandex sweep enabled=false), так что ждать накопления +-- истории «органически» нельзя. Решение: ежедневный internal snapshot существующих 18 355 +-- строк listing_sources (4 источника) — история начинает копиться СРАЗУ, независимо от +-- скраперов. Снимок пишет snapshot_listing_sources() (app/tasks/listing_source_snapshot.py), +-- запускаемый in-app scheduler'ом (source='listing_source_snapshot', seed ниже, enabled=true). +-- +-- СОЗДАЁТ: +-- listing_source_snapshots — daily snapshot на источник (PK (listing_source_id, snapshot_date)). +-- listing_source_events — change-log (price_change/delisted/relisted/edited/first_seen). +-- offer_price_history.listing_source_id — backlink на конкретный источник (был только source text). +-- v_listing_source_price_on_date — view «цена источника на дату» (snapshots ⋈ listing_sources). +-- +-- ПАРТИЦИОНИРОВАНИЕ — НАМЕРЕННО НЕТ (отложено). +-- ADR предлагал партиционировать listing_source_snapshots по snapshot_date (RANGE), +-- но на текущем объёме это преждевременно: ~18 355 строк/день × ~30 ≈ 540k строк/мес — +-- plain-таблица с индексом (snapshot_date DESC) обслуживает это без проблем. Партиционирование +-- вводим позже отдельной миграцией, когда retention/объём этого реально потребуют. +-- +-- ЗАВИСИМОСТИ: listing_sources (existing), scrape_runs (015_scrape_runs.sql), +-- offer_price_history (existing), scrape_schedules (052_scrape_schedules.sql, UNIQUE(source)). +-- Idempotent: CREATE TABLE/INDEX IF NOT EXISTS, ADD COLUMN IF NOT EXISTS, CREATE OR REPLACE VIEW, +-- ON CONFLICT (source) DO NOTHING — безопасно запускать повторно. +-- Apply after: 078_scrape_schedules_seed_yandex_sweep.sql + +BEGIN; + +-- ── listing_source_snapshots — ежедневный снимок на источник ────────────────── +-- is_active — derived: last_seen_at в пределах окна свежести (7 дней) на момент снимка. +-- payload_hash — md5(raw_payload::text), дешёвый детектор «изменился ли payload» между днями. +-- run_id — провенанс: какой scrape_runs создал снимок (ON DELETE SET NULL — run могут чистить). +CREATE TABLE IF NOT EXISTS listing_source_snapshots ( + listing_source_id bigint NOT NULL REFERENCES listing_sources(id) ON DELETE CASCADE, + snapshot_date date NOT NULL DEFAULT CURRENT_DATE, + price_rub bigint, + is_active boolean NOT NULL, + last_seen_at timestamptz, + payload_hash text, + observed_at timestamptz NOT NULL DEFAULT now(), + run_id bigint REFERENCES scrape_runs(id) ON DELETE SET NULL, + PRIMARY KEY (listing_source_id, snapshot_date) +); + +COMMENT ON TABLE listing_source_snapshots IS + 'Ежедневный снимок listing_sources на источник (#570). Plain-таблица (не партиционирована — ' + 'см. шапку 079). PK (listing_source_id, snapshot_date), last-write-wins за день.'; + +CREATE INDEX IF NOT EXISTS idx_lss_snapshot_date + ON listing_source_snapshots (snapshot_date DESC); +CREATE INDEX IF NOT EXISTS idx_lss_source_date + ON listing_source_snapshots (listing_source_id, snapshot_date DESC); + +-- ── listing_source_events — change-log по источнику ─────────────────────────── +-- diff_percent — для price_change: (new-old)/old*100. payload_diff — опциональный JSON-дифф. +-- UNIQUE (listing_source_id, change_time, event_type) — идемпотентность писателя (ON CONFLICT). +CREATE TABLE IF NOT EXISTS listing_source_events ( + id bigserial PRIMARY KEY, + listing_source_id bigint NOT NULL REFERENCES listing_sources(id) ON DELETE CASCADE, + change_time timestamptz NOT NULL, + event_type text NOT NULL CHECK ( + event_type IN ('price_change', 'delisted', 'relisted', 'edited', 'first_seen') + ), + price_rub bigint, + diff_percent numeric, + payload_diff jsonb, + recorded_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (listing_source_id, change_time, event_type) +); + +COMMENT ON TABLE listing_source_events IS + 'Change-log на источник (#570): price_change/delisted/relisted/edited/first_seen. ' + 'Писатель — snapshot_listing_sources() (set-based, ON CONFLICT DO NOTHING).'; + +CREATE INDEX IF NOT EXISTS idx_lse_source_change_time + ON listing_source_events (listing_source_id, change_time DESC); +CREATE INDEX IF NOT EXISTS idx_lse_type_change_time + ON listing_source_events (event_type, change_time DESC); + +-- ── offer_price_history — backlink на конкретный источник ───────────────────── +-- Раньше была только колонка source (text) — без ссылки на listing_sources.id. +-- ON DELETE SET NULL: история переживает удаление источника (canonical price-history сохраняется). +ALTER TABLE offer_price_history + ADD COLUMN IF NOT EXISTS listing_source_id bigint + REFERENCES listing_sources(id) ON DELETE SET NULL; + +CREATE INDEX IF NOT EXISTS idx_oph_source_change_time + ON offer_price_history (listing_source_id, change_time DESC); + +-- ── v_listing_source_price_on_date — «цена источника на дату» ───────────────── +-- Удобный join snapshots ⋈ listing_sources: (listing_id, ext_source, дата, цена, активность). +CREATE OR REPLACE VIEW v_listing_source_price_on_date AS +SELECT + ls.listing_id, + ls.ext_source, + lss.snapshot_date, + lss.price_rub, + lss.is_active +FROM listing_source_snapshots lss +JOIN listing_sources ls ON ls.id = lss.listing_source_id; + +COMMENT ON VIEW v_listing_source_price_on_date IS + 'Цена/активность источника на дату снимка (#570): listing_source_snapshots ⋈ listing_sources.'; + +-- ── Seed scrape_schedules row для daily snapshot ────────────────────────────── +-- enabled = true — БЕЗОПАСНО включать сразу: pure-internal DB snapshot, никаких внешних +-- HTTP-вызовов / анти-бота (в отличие от avito/yandex sweep, которые seed'ятся DORMANT). +-- Окно 01:00-02:00 UTC — РАНЬШЕ остальных job'ов (cian/rosreestr 02:00+), чтобы снять +-- снимок до того, как backfill-задачи начнут менять listing_sources в тот же день. +-- next_run_at = завтрашнее наступление окна (tomorrow + 01:00 UTC), иначе при NULL +-- get_due_schedules() выстрелит сразу после деплоя. AT TIME ZONE 'UTC' фиксирует +-- wall-clock час в UTC независимо от session TimeZone GUC. +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'listing_source_snapshot', + true, -- SAFE: pure internal DB, no external calls + 1, + 2, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 1)) 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).'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_listing_source_snapshot.py b/tradein-mvp/backend/tests/test_listing_source_snapshot.py new file mode 100644 index 00000000..8b20acf8 --- /dev/null +++ b/tradein-mvp/backend/tests/test_listing_source_snapshot.py @@ -0,0 +1,281 @@ +"""Tests for the per-source listing price-history snapshot writer (#570). + +Writer (snapshot_listing_sources) is SQL-heavy, so most assertions are static: we read +the emitted SQL via inspect/text-attr and check the upsert + event-diff CTE shape, table/ +column names, and the psycopg-v3 cast discipline (no :param::type). We also assert the +scheduler wiring (trigger fn + dispatch branch) and the migration 079 contents. + +Plus one cheap behavioural test: a fake db (monkeypatched .execute) drives the counter +logic without a real Postgres. + +Static style mirrors tests/test_rosreestr_dedup_key.py / test_yandex_city_sweep.py. +""" + +import inspect +import os +import re +from pathlib import Path +from typing import Any + +import pytest + +# Importing app.services.scheduler / app.tasks pulls app.core.config.Settings → needs +# DATABASE_URL. Stub it BEFORE app imports (as in test_scheduler.py) — these tests are +# static / fake-db; no live database is touched. +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from app.services import scheduler +from app.tasks import listing_source_snapshot as snap_mod + +_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql" +_MIGRATION_079 = _SQL_DIR / "079_listing_source_history.sql" + +# Emitted SQL text (SQLAlchemy text() clause → .text gives the raw string). +_SNAPSHOT_SQL = str(snap_mod._SNAPSHOT_SQL.text) +_EVENT_DIFF_SQL = str(snap_mod._EVENT_DIFF_SQL.text) +_ALL_SQL = _SNAPSHOT_SQL + "\n" + _EVENT_DIFF_SQL +_WRITER_SRC = inspect.getsource(snap_mod.snapshot_listing_sources) + + +# ── Snapshot upsert SQL ─────────────────────────────────────────────────────── + + +def test_snapshot_targets_correct_table_and_columns() -> None: + assert "INSERT INTO listing_source_snapshots" in _SNAPSHOT_SQL + for col in ( + "listing_source_id", + "snapshot_date", + "price_rub", + "is_active", + "last_seen_at", + "payload_hash", + "observed_at", + "run_id", + ): + assert col in _SNAPSHOT_SQL, f"snapshot SQL missing column {col!r}" + assert "FROM listing_sources" in _SNAPSHOT_SQL + assert "CURRENT_DATE" in _SNAPSHOT_SQL + + +def test_snapshot_is_upsert_last_write_wins() -> None: + """ON CONFLICT (listing_source_id, snapshot_date) DO UPDATE — last-write-wins for the day.""" + assert "ON CONFLICT (listing_source_id, snapshot_date) DO UPDATE" in _SNAPSHOT_SQL + # Collapse runs of whitespace so column-alignment spacing doesn't matter. + flat = re.sub(r"\s+", " ", _SNAPSHOT_SQL) + for col in ("price_rub", "is_active", "last_seen_at", "payload_hash", "observed_at", "run_id"): + assert f"{col} = EXCLUDED.{col}" in flat, f"upsert SET missing {col}" + + +def test_snapshot_derives_is_active_and_payload_hash() -> None: + # is_active derived from freshness window; payload_hash = md5(raw_payload::text). + assert "make_interval(days => :freshness_days)" in _SNAPSHOT_SQL + assert "AS is_active" in _SNAPSHOT_SQL + assert "md5(raw_payload::text)" in _SNAPSHOT_SQL + + +# ── Event-diff CTE SQL ──────────────────────────────────────────────────────── + + +def test_event_diff_is_set_based_cte_not_python_loop() -> None: + """Event diff is one set-based INSERT … SELECT over a CTE — never a row-by-row loop.""" + assert "WITH today AS" in _EVENT_DIFF_SQL + assert "prior AS" in _EVENT_DIFF_SQL + assert "DISTINCT ON (listing_source_id)" in _EVENT_DIFF_SQL + assert "INSERT INTO listing_source_events" in _EVENT_DIFF_SQL + # Prior = most-recent snapshot strictly before today. + assert "snapshot_date < CURRENT_DATE" in _EVENT_DIFF_SQL + assert "snapshot_date = CURRENT_DATE" in _EVENT_DIFF_SQL + assert "ORDER BY listing_source_id, snapshot_date DESC" in _EVENT_DIFF_SQL + # No Python iteration over rows in the writer body (set-based only). + body = _WRITER_SRC.split('"""', 2)[-1] + assert "for " not in body, "writer must be set-based — no Python row loop" + + +def test_event_diff_emits_price_change_with_diff_percent() -> None: + assert "'price_change'" in _EVENT_DIFF_SQL + # diff_percent = (new-old)/old*100. + assert "(t.price_rub - p.price_rub)" in _EVENT_DIFF_SQL + assert "/ p.price_rub * 100" in _EVENT_DIFF_SQL + # Only when the price actually changed and old is a usable denominator. + assert "t.price_rub <> p.price_rub" in _EVENT_DIFF_SQL + assert "p.price_rub <> 0" in _EVENT_DIFF_SQL + assert "p.price_rub IS NOT NULL" in _EVENT_DIFF_SQL + + +def test_event_diff_is_idempotent_on_conflict() -> None: + assert "ON CONFLICT (listing_source_id, change_time, event_type) DO NOTHING" in _EVENT_DIFF_SQL + + +# ── psycopg v3 cast discipline ──────────────────────────────────────────────── + + +def test_sql_uses_psycopg_v3_casts_not_double_colon() -> None: + """psycopg v3: bind params via CAST(:x AS type), never :x::type. + + Literal ::-casts on columns (raw_payload::text, ::numeric) are fine — the ban is + only on :param::type, which psycopg v3 breaks. + """ + # No bind-param (:name) followed by ::type anywhere in the emitted SQL. + assert not re.search(r":\w+::", _ALL_SQL) + # run_id bound via CAST(:run_id AS bigint). + assert "CAST(:run_id AS bigint)" in _SNAPSHOT_SQL + # freshness window bound via named param inside make_interval (no cast needed). + assert ":freshness_days" in _SNAPSHOT_SQL + + +# ── Writer return / finalisation contract ───────────────────────────────────── + + +def test_writer_returns_counters_and_finalises_run() -> None: + assert "snapshotted" in _WRITER_SRC + assert "price_change_events" in _WRITER_SRC + assert "mark_done" in _WRITER_SRC + assert "mark_failed" in _WRITER_SRC + + +# ── Scheduler wiring ────────────────────────────────────────────────────────── + + +def test_trigger_fn_exists_and_is_async() -> None: + fn = getattr(scheduler, "trigger_listing_source_snapshot_run", None) + assert fn is not None, "trigger_listing_source_snapshot_run missing from scheduler" + assert inspect.iscoroutinefunction(fn) + + +def test_dispatch_branch_wired() -> None: + """scheduler_loop dispatches source='listing_source_snapshot' to the trigger.""" + loop_src = inspect.getsource(scheduler.scheduler_loop) + assert 'source == "listing_source_snapshot"' in loop_src + assert "trigger_listing_source_snapshot_run(db, sch)" in loop_src + + +def test_trigger_runs_sync_task_in_executor() -> None: + """Sync DB-only task → run_in_executor (mirror rosreestr), not a bare await.""" + trig_src = inspect.getsource(scheduler.trigger_listing_source_snapshot_run) + assert "run_in_executor(None, snapshot_listing_sources" in trig_src + assert "_claim_run(db, schedule_row)" in trig_src + + +# ── Migration 079 ───────────────────────────────────────────────────────────── + + +def test_migration_079_exists() -> None: + assert _MIGRATION_079.is_file(), f"missing migration: {_MIGRATION_079}" + + +def test_migration_079_creates_three_schema_objects_and_view() -> None: + sql = _MIGRATION_079.read_text("utf-8") + # 1. snapshots table (plain — explicitly NOT partitioned, noted in header). + assert "CREATE TABLE IF NOT EXISTS listing_source_snapshots" in sql + assert "PRIMARY KEY (listing_source_id, snapshot_date)" in sql + assert "REFERENCES listing_sources(id) ON DELETE CASCADE" in sql + assert "REFERENCES scrape_runs(id) ON DELETE SET NULL" in sql + assert "CREATE INDEX IF NOT EXISTS idx_lss_snapshot_date" in sql + assert "CREATE INDEX IF NOT EXISTS idx_lss_source_date" in sql + # 2. events table + CHECK + UNIQUE. + assert "CREATE TABLE IF NOT EXISTS listing_source_events" in sql + assert "id bigserial PRIMARY KEY" in sql.replace("\t", " ") or ( + "id" in sql and "bigserial" in sql and "PRIMARY KEY" in sql + ) + assert "'price_change', 'delisted', 'relisted', 'edited', 'first_seen'" in sql + assert "UNIQUE (listing_source_id, change_time, event_type)" in sql + # 3. offer_price_history backlink column + index. + assert "ALTER TABLE offer_price_history" in sql + assert "ADD COLUMN IF NOT EXISTS listing_source_id bigint" in sql + assert "idx_oph_source_change_time" in sql + # 4. view. + assert "CREATE OR REPLACE VIEW v_listing_source_price_on_date" in sql + assert "JOIN listing_sources ls ON ls.id = lss.listing_source_id" in sql + # Transactional + idempotent + partitioning note. + assert "BEGIN;" in sql and "COMMIT;" in sql + assert "НЕТ" in sql and "артицион" in sql # partitioning explicitly deferred + + +def test_migration_079_seeds_schedule_enabled_true() -> None: + sql = _MIGRATION_079.read_text("utf-8") + assert "'listing_source_snapshot'" in sql + assert "'{}'::jsonb" in sql + assert "ON CONFLICT (source) DO NOTHING" in sql + # enabled=true (SAFE — pure internal DB). Assert the seeded VALUES row has true, + # not false (unlike the dormant yandex/avito sweeps). + assert "true, -- SAFE" in sql + assert "false" not in sql.split("INSERT INTO scrape_schedules")[1] + + +def test_migration_079_uses_psycopg_safe_sql() -> None: + """Migration is plain DDL (no bind params), but guard against accidental :x::type.""" + sql = _MIGRATION_079.read_text("utf-8") + assert not re.search(r":\w+::", sql) + + +# ── Cheap behavioural test: counter logic via fake db ───────────────────────── + + +class _FakeResult: + def __init__(self, rowcount: int) -> None: + self.rowcount = rowcount + + +class _FakeDB: + """Minimal stand-in for a SQLAlchemy Session — records execute() calls, returns rowcounts.""" + + def __init__(self, rowcounts: list[int]) -> None: + self._rowcounts = list(rowcounts) + self.executed: list[Any] = [] + self.committed = False + self.rolled_back = False + + def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult: + self.executed.append((stmt, params)) + return _FakeResult(self._rowcounts.pop(0)) + + def commit(self) -> None: + self.committed = True + + def rollback(self) -> None: + self.rolled_back = True + + +def test_counter_logic_with_fake_db(monkeypatch: pytest.MonkeyPatch) -> None: + """snapshot_listing_sources maps the two execute() rowcounts to its counters + marks done.""" + marked: dict[str, Any] = {} + monkeypatch.setattr( + snap_mod.runs_mod, + "mark_done", + lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)), + ) + monkeypatch.setattr(snap_mod.runs_mod, "mark_failed", lambda *a, **k: None) + + db = _FakeDB(rowcounts=[18355, 42]) # snapshot rowcount, then event rowcount + out = snap_mod.snapshot_listing_sources(db, run_id=99) # type: ignore[arg-type] + + assert out == {"snapshotted": 18355, "price_change_events": 42} + assert db.committed is True + assert len(db.executed) == 2 + # run_id threaded into the snapshot statement's bind params. + _stmt, params = db.executed[0] + assert params is not None and params["run_id"] == 99 + # Run finalised via mark_done with the same counters. + assert marked["run_id"] == 99 + assert marked["counters"] == {"snapshotted": 18355, "price_change_events": 42} + + +def test_counter_logic_failure_path_marks_failed(monkeypatch: pytest.MonkeyPatch) -> None: + """On execute error: rollback + mark_failed + re-raise (no silent swallow).""" + failed: dict[str, Any] = {} + monkeypatch.setattr(snap_mod.runs_mod, "mark_done", lambda *a, **k: None) + monkeypatch.setattr( + snap_mod.runs_mod, + "mark_failed", + lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err), + ) + + class _BoomDB(_FakeDB): + def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult: + raise RuntimeError("boom") + + db = _BoomDB(rowcounts=[]) + with pytest.raises(RuntimeError, match="boom"): + snap_mod.snapshot_listing_sources(db, run_id=7) # type: ignore[arg-type] + assert db.rolled_back is True + assert failed["run_id"] == 7