| Количество сделок по аналогичном объектам | {_mono(f"{n_deals} шт.")} |
| Период сделок | -- {_mono(f"{period_start.strftime('%m.%Y')} – {today.strftime('%m.%Y')}")} |
| Сделки | +{_mono(deals_as_of)} |
User-agent: *" + ) class _Resp: status_code = 200 - text = '{"html": "
User-agent: *"}' + text = "{}" @staticmethod def json() -> dict[str, str]: - return {"html": "
User-agent: *"} + return {"html": _body} class _Client: def __init__(self, **_kw: Any) -> None: ... @@ -196,7 +206,10 @@ async def test_probe_asks_the_working_host_of_each_source( assert ok is True assert must_contain in seen["payload"]["url"] - assert seen["payload"]["url"].endswith("/robots.txt") # нагрузки на площадку нет + # #2855: путь тоже сторожим — у robots.txt-источников он лёгкий, у Домклика это + # боевой count-эндпоинт (одно число в ответе, без пагинации и выдачи), то есть + # нагрузка на площадку остаётся минимальной, а защита — той же, что у работы. + assert path_suffix in seen["payload"]["url"] # Инстанс сайдкара остаётся 'generic' — проба не отбирает лок у боевой сессии. assert seen["payload"]["source"] == "generic" @@ -233,7 +246,10 @@ async def test_stub_page_with_status_200_is_a_failure(monkeypatch: pytest.Monkey assert ok is False # Тракт узла исправен — виновата ПАРА: площадка не отдала ресурс этому exit-IP. assert fail_kind == "page" - assert "not robots.txt" in detail + # #2855: сообщение называет КОНКРЕТНЫЙ маркер, потому что он теперь зависит от + # площадки ('User-agent' у robots.txt-источников, 'snippetsCount' у Домклика). + # Прежний текст «not robots.txt» стал бы враньём там, где robots.txt и не просили. + assert "no marker 'User-agent'" in detail async def test_stub_page_bans_the_pair(monkeypatch: pytest.MonkeyPatch) -> None: @@ -454,3 +470,72 @@ async def test_flaky_failure_does_not_ban_the_pair(monkeypatch: pytest.MonkeyPat lease = acquire(db, "domclick") # type: ignore[arg-type] assert lease is not None release(db, lease.id) # type: ignore[arg-type] + + +# ── 8. #2855: проба обязана делить с работой ЗАЩИЩЁННЫЙ ПУТЬ, а не только хост ── + + +@pytest.mark.parametrize( + ("api_body", "expect_ok"), + [ + # Блок: площадка отдаёт robots.txt свободно и закрывает /api/offers/* QRATOR'ом. + # На старом коде проба спрашивала robots.txt → 'User-agent' на месте → ok=True, + # то есть «пара здорова» ровно там, где свип получает блок. Красный на main. + ("
User-agent: *") + return _Resp(api_body) + + monkeypatch.setattr(bf.httpx, "AsyncClient", _Client) + ok, fail_kind, _detail = await bf.probe_proxy_via_browser( + "http://tradein-browser:3000", "http://u:p@node:8080", source="domclick" + ) + + assert ok is expect_ok + if not expect_ok: + # Тракт узла исправен (200 пришёл) — негодна ПАРА, значит "page", не "proxy". + assert fail_kind == "page" + # Спрашивали именно боевой путь, а не robots.txt того же хоста. + assert "/api/offers/" in seen["payload"]["url"] diff --git a/tradein-mvp/backend/tests/test_2857_search_mv_placeholder_columns.py b/tradein-mvp/backend/tests/test_2857_search_mv_placeholder_columns.py new file mode 100644 index 00000000..9df5b583 --- /dev/null +++ b/tradein-mvp/backend/tests/test_2857_search_mv_placeholder_columns.py @@ -0,0 +1,145 @@ +"""Витрина поиска не обещает колонок, которых не заполняет (#2857, эпик #2674). + +`listings_search_mv` с 050 несла четыре колонки, заданные литералом `NULL` прямо +в определении: district, distance_to_metro_m, last_price_change, photos_count. +Это не потеря данных и не оборванный писатель — имена зарезервировали, реализацию +не подключили никогда. Три из них не читает НИКТО (ни бэкенд, ни фронт, ни тесты) +и они сняты миграцией 261; district оставлен намеренно — он объявлен в +schemas/search_response.py, то есть API его отдаёт, и его снос это ломающее +изменение контракта (решение владельца, вынесено отдельно в #2857). + +Проверяется ФАКТ, а не текст: тест собирает СПИСОК КОЛОНОК витрины разбором её +актуального определения (самый старший NN среди файлов, создающих витрину) и +смотрит на состав списка. Переформатирование SQL, перестановка строк или смена +`NULL::int` на `NULL::integer` тест не трогают; возврат колонки — краснит. + +Без БД и сети: миграции читаются как текст, разбираются в структуру. +""" + +from __future__ import annotations + +import os +import re +from pathlib import Path + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql" + +MV = "listings_search_mv" +# Сняты 261: ноль читателей во всём репозитории на момент сноса. +DROPPED = ("distance_to_metro_m", "last_price_change", "photos_count") + + +def _strip_sql_comments(sql: str) -> str: + sql = re.sub(r"/\*.*?\*/", " ", sql, flags=re.DOTALL) + return re.sub(r"--[^\n]*", "", sql) + + +def _latest_definition() -> str: + """Текст файла с самым старшим NN, который создаёт витрину = её актуальный вид.""" + creators = [ + p + for p in SQL_DIR.glob("*.sql") + if re.search( + rf"CREATE\s+MATERIALIZED\s+VIEW\s+{MV}\b", + _strip_sql_comments(p.read_text("utf-8")), + re.I, + ) + ] + assert creators, f"ни одна миграция не создаёт {MV} — тест смотрит не туда" + return max(creators, key=lambda p: int(p.name.split("_", 1)[0])).read_text("utf-8") + + +def mv_columns() -> list[str]: + """Имена колонок витрины в порядке объявления. + + Разбор: от `AS SELECT` до `FROM` на нулевой глубине скобок, разрез по запятым + той же глубины, имя колонки — последний идентификатор элемента (алиас после + `AS` либо хвост `l.foo`). + """ + sql = _strip_sql_comments(_latest_definition()) + body = re.split(rf"CREATE\s+MATERIALIZED\s+VIEW\s+{MV}\s+AS\s+SELECT\b", sql, flags=re.I)[1] + + depth, items, cur = 0, [], [] + for token in re.finditer(r"\(|\)|,|\bFROM\b|[^(),]+", body, re.I): + t = token.group(0) + if t == "(": + depth += 1 + elif t == ")": + depth -= 1 + elif depth == 0 and t == ",": + items.append("".join(cur)) + cur = [] + continue + elif depth == 0 and t.upper() == "FROM": + break + cur.append(t) + items.append("".join(cur)) + + return [item.split()[-1].split(".")[-1] for item in items if item.split()] + + +def test_placeholder_columns_are_gone_from_the_matview() -> None: + """Red => витрина снова обещает поля, которых не заполняет (#2857). + + Три колонки были литеральным `NULL` без единого читателя. Если тест покраснел + после возврата колонки — сначала заведи писателя, потом колонку, а не наоборот. + """ + cols = mv_columns() + still_there = [c for c in DROPPED if c in cols] + assert not still_there, ( + f"{MV} снова отдаёт колонки-заглушки {still_there}. Колонка без писателя " + "читается снаружи как «данные есть, просто у этого объекта пусто» — это " + "хуже мёртвого кода, потому что видно в контракте." + ) + + +def test_district_is_deliberately_kept() -> None: + """Red => district снесли заодно, а он в схеме ответа API. + + schemas/search_response.py объявляет `district: str | None`, services/search_query.py + его тянет — снос ломает контракт /api/v1/search. Это решение владельца (#2857), + а не побочный эффект уборки соседних заглушек. Убирать — вместе со схемой ответа. + """ + assert "district" in mv_columns(), ( + f"district пропал из {MV}, а schemas/search_response.py его всё ещё объявляет: " + "ответ поиска начнёт падать/врать. Снимать поле — только вместе со схемой." + ) + + +def test_search_api_selects_only_columns_the_matview_has() -> None: + """Настоящий инвариант: то, что просит API, витрина обязана иметь. + + Именно эта проверка отличает «список колонок» от «поиска подстроки»: она + краснеет на ЛЮБОЙ колонке, снесённой без правки читателя, а не только на трёх + известных именах. + """ + from app.schemas.search import SearchParams + from app.services.search_query import build_search_query + + sql, _ = build_search_query(SearchParams()) + selected = [ + c.strip() for c in sql[len("SELECT ") : sql.index(f" FROM {MV}")].split(",") if c.strip() + ] + missing = [c for c in selected if c not in mv_columns()] + assert not missing, ( + f"services/search_query.py просит у {MV} колонки, которых в её определении нет: " + f"{missing}. Либо верни колонку в витрину, либо убери её из запроса И из " + "schemas/search_response.py." + ) + + +def test_unique_index_for_concurrent_refresh_survives_recreation() -> None: + """Red => ночной REFRESH ... CONCURRENTLY упадёт. + + app/tasks/refresh_search_matview.py рефрешит витрину CONCURRENTLY (расписание + refresh_search_matview, 03:00-04:00 UTC). Без UNIQUE-индекса PostgreSQL отвечает + «cannot refresh materialized view concurrently ... no unique index» — а витрина, + которую пересоздали и забыли проиндексировать, молчит до самой ночи. + """ + sql = _strip_sql_comments(_latest_definition()) + assert re.search(rf"CREATE\s+UNIQUE\s+INDEX[^;]+ON\s+{MV}\s*\(\s*listing_id\s*\)", sql, re.I), ( + f"в актуальном определении {MV} нет UNIQUE-индекса по listing_id — " + "REFRESH MATERIALIZED VIEW CONCURRENTLY без него невозможен." + ) diff --git a/tradein-mvp/backend/tests/test_930_scheduler_resume_checkpoint.py b/tradein-mvp/backend/tests/test_930_scheduler_resume_checkpoint.py new file mode 100644 index 00000000..cc0ba37a --- /dev/null +++ b/tradein-mvp/backend/tests/test_930_scheduler_resume_checkpoint.py @@ -0,0 +1,295 @@ +"""#930 добивка: планировщик не подхватывал чекпоинт оборванного прогона. + +#930 сделал обе половины механизма — запись точки (`counters.done_buckets`, per-bucket +heartbeat) и её чтение (`run_*_full_load(resume_run_id=...)`, skip-set в SERP-слое), — +но единственным входом оставил админку. У avito full-load админского эндпоинта нет +вовсе, а планировщик передавал `resume_run_id=None` ЛИТЕРАЛОМ (scheduler.py 708/728/799 +на origin/main). То есть боевой путь возобновления не существовал ни одного дня. + +Цена на проде (замер 2026-08-12, 90 суток, read-only): 433 корзины в 30 оборванных +прогонах с ЖИВОЙ незабранной точкой — avito_full_load 242, cian_full_load 134, +avito_full_load_exhaustive 57. Прогон 3547 (09.08, убит деплоем на третьем часу, 35 из +84 корзин дерева) лежит до сих пор и будет подхвачен расписанием 139 16.08. + +Красный прогон на origin/main: + 1. `test_scheduler_hands_checkpoint_to_pipeline` — планировщик отдаёт в пайплайн + resume_run_id=None вместо id прошлого прогона (AssertionError на 3 источниках); + 2. `test_partial_bucket_is_not_complete` — бакет с выпавшей страницей приезжает в + on_bucket неотличимым от целого (у колбэка нет аргумента полноты вообще); + 3. `test_pipeline_keeps_partial_bucket_out_of_checkpoint` — TypeError: `_on_bucket` + на main принимает два аргумента, признаку полноты некуда приехать. +Тесты ладдера (`_resume_decision`) на main падают с AttributeError — функции нет. +""" + +from __future__ import annotations + +import os +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + +from scraper_kit.orchestration import scheduler as sched +from scraper_kit.orchestration.pipeline import run_avito_full_load +from scraper_kit.providers.avito.serp import AvitoScraper + +PFX = "scraper_kit.orchestration.pipeline" + +# Прод-слепок расписания 139 (avito_full_load_exhaustive) на 2026-08-12: params прогона +# 3547 совпадают с default_params расписания байт-в-байт — это и есть «то же задание». +_PARAMS = { + "concurrency": 1, + "interval_days": 7, + "secondary_only": True, + "request_delay_sec": 7.0, + "price_cap_per_bucket": 1400, +} + + +def _candidate(**over: Any) -> SimpleNamespace: + """Строка-кандидат из _RESUME_CANDIDATE_SQL: прогон 3547 как он лежит на проде.""" + base = { + "prev_id": 3547, + "prev_status": "cancelled", + "prev_counters": { + "unique_fetched": 5496, + "done_buckets": [f"room_1_komn:{i}:0" for i in range(35)], + }, + "same_params": True, + "age_h": 164.6, # 6.86 суток — столько будет точке 3547 к подхвату 16.08 + "interval_days": "7", + } + base.update(over) + return SimpleNamespace(**base) + + +class _FakeDb: + """Двойник сессии: отдаёт ОДНУ строку-кандидата на любой SELECT, глотает UPDATE.""" + + def __init__(self, row: Any) -> None: + self.row = row + self.written: list[dict[str, Any]] = [] + + def execute(self, _stmt: Any, params: dict[str, Any] | None = None) -> Any: + if params and "counters" in params: # update_heartbeat пишет вердикт + self.written.append(params) + return MagicMock() + return MagicMock(fetchone=lambda: self.row) + + def commit(self) -> None: + pass + + +# ── 1. Главное: планировщик обязан отдать точку в пайплайн ─────────────────── + + +@pytest.mark.parametrize( + ("job", "pipeline_fn"), + [ + (sched._job_avito_full_load, "run_avito_full_load"), + (sched._job_avito_full_load_exhaustive, "run_avito_full_load"), + (sched._job_cian_full_load, "run_cian_full_load"), + ], +) +async def test_scheduler_hands_checkpoint_to_pipeline(job: Any, pipeline_fn: str) -> None: + """Оборванный прогон с валидной точкой → новый прогон продолжает его, а не с нуля. + + Падает на origin/main: планировщик передаёт литеральный None — 433 корзины за 90 + суток перебирались заново, включая 35 корзин прогона 3547. + """ + db = _FakeDb(_candidate()) + captured: dict[str, Any] = {} + + async def _spy(*_a: Any, **kw: Any) -> None: + captured.update(kw) + + with patch.object(sched, pipeline_fn, _spy): + await job(db, 4000, dict(_PARAMS), MagicMock()) + + assert captured["resume_run_id"] == 3547 + + +async def test_verdict_lands_in_counters_of_new_run() -> None: + """Подхватили или нет — видно В СЧЁТЧИКАХ прогона, а не только в docker-логах. + + Логи теряются при редеплое (контейнер tradein-scraper пересоздаётся), поэтому + молчаливый отказ подхватить неотличим от отсутствия правки. + """ + db = _FakeDb(_candidate(prev_status="zombie")) + with patch.object(sched, "run_avito_full_load", AsyncMock()): + await sched._job_avito_full_load(db, 4000, dict(_PARAMS), MagicMock()) + + assert db.written, "вердикт о подхвате не записан в counters нового прогона" + written = db.written[-1]["counters"] + assert '"resume_reason": "status_zombie"' in written + assert '"resume_candidate": 3547' in written + + +# ── 2. Ладдер отказов: у каждого нуля своя причина ─────────────────────────── + + +@pytest.mark.parametrize( + ("row", "reason"), + [ + (None, "no_prev_run"), + (_candidate(prev_status="done"), "status_done"), + (_candidate(prev_status="zombie"), "status_zombie"), + (_candidate(same_params=False), "params_changed"), + (_candidate(prev_counters={"unique_fetched": 2977}), "no_checkpoint"), + (_candidate(age_h=200.0), "checkpoint_stale"), + (_candidate(prev_counters={"done_buckets": ["a"], "resume_chain": 2}), "chain_limit"), + ], +) +def test_resume_refusals_are_named(row: Any, reason: str) -> None: + """«Не подхватили» — это семь РАЗНЫХ фактов, и в counters они различимы.""" + resume_id, verdict = sched._resume_decision(row) + assert resume_id is None + assert verdict["resume_reason"] == reason + assert verdict["resume_from"] is None + + +def test_resume_chain_is_bounded() -> None: + """Цепочка возобновлений считается и упирается в потолок, а не тянется вечно. + + Потолок выведен из STALE_DIGEST_INTERVAL_FACTOR (см. scheduler.py): полный обход + обязан начаться раньше, чем сводка объявит источник просроченным. + """ + assert sched._MAX_RESUME_CHAIN == sched.STALE_DIGEST_INTERVAL_FACTOR - 1 + _id, first = sched._resume_decision(_candidate()) + assert first["resume_chain"] == 1 + _id2, second = sched._resume_decision( + _candidate(prev_counters={"done_buckets": ["a"], "resume_chain": 1}) + ) + assert second["resume_chain"] == sched._MAX_RESUME_CHAIN + third_id, third = sched._resume_decision( + _candidate(prev_counters={"done_buckets": ["a"], "resume_chain": 2}) + ) + assert third_id is None and third["resume_reason"] == "chain_limit" + + +def test_stale_threshold_follows_the_source_tick() -> None: + """Срок годности точки считается от такта ИСТОЧНИКА, а не общей константой. + + cian ходит раз в 3 суток, avito — раз в 7; одна и та же точка возрастом 100 ч для + первого просрочена, для второго свежая. Плюс сутки — сетка запуска (см. + _resume_decision): 164.6 ч прогона 3547 при такте 7 суток обязаны пройти, иначе + точку отвергал бы jitter расписания, а пропущенный цикл (13 суток) — нет. + """ + assert sched._resume_decision(_candidate(age_h=100.0, interval_days="3"))[0] is None + assert sched._resume_decision(_candidate(age_h=100.0, interval_days="7"))[0] == 3547 + assert sched._resume_decision(_candidate(age_h=164.6, interval_days="7"))[0] == 3547 + assert sched._resume_decision(_candidate(age_h=13 * 24.0, interval_days="7"))[0] is None + + +# ── 3. Недособранный бакет не имеет права попасть в чекпоинт ───────────────── + + +def _serp_config() -> SimpleNamespace: + return SimpleNamespace( + scraper_fetch_mode="curl_cffi", + browser_http_endpoint="http://browser.test/fetch", + scraper_proxy_url=None, + avito_proxy_max_rotations=0, + avito_serp_ok_not_banned=True, + avito_proxy_rotate_settle_s=0.0, + proxy_rotate_attempts=1, + proxy_rotate_attempt_timeout_s=1.0, + scraper_skip_seen_today=False, + ) + + +@pytest.mark.parametrize( + ("page2_html", "expected_complete"), + [(None, False), ("