bot-backend
c927b77777
Some checks failed
Deploy Trade-In / changes (push) Successful in 12s
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / test (push) Failing after 3m29s
Deploy Trade-In / build-backend (push) Has been skipped
Deploy Trade-In / deploy (push) Has been skipped
267 lines
13 KiB
Python
267 lines
13 KiB
Python
"""#2674: «временная» ошибка домовой IMV-оценки снова означает временную.
|
||
|
||
Что было. Очередь backfill'а выбирает РОВНО ОДИН imv_status за прогон
|
||
(`only_status`, дефолт 'pending'). Возможность повтора существовала и была
|
||
задокументирована в admin-API, но за всю историю (41 прогон, 26.06–11.08.2026)
|
||
её не передал никто: все прогоны ушли с 'pending'. Дома, упавшие по причине,
|
||
которой больше нет (503/500 от браузерного сайдкара, #2698), лежали в
|
||
'transient_error' без единого шанса вернуться — 1390 штук на 12.08.2026.
|
||
|
||
Что проверяем — по существу, а не по настройке:
|
||
* дом в 'transient_error' ПОПАДАЕТ в пакет прогона с дефолтными параметрами
|
||
(тот же вызов, что делает расписание) — на старом коде он не попадает;
|
||
* pending при этом из пакета НЕ вытесняется;
|
||
* дом, исчерпавший лимит попыток, в автоматический повтор не берётся, а
|
||
ЯВНЫЙ only_status='transient_error' (ручной запрос оператора) берёт и его;
|
||
* дом без пригодных параметров помечается no_params БЕЗ запроса к площадке.
|
||
|
||
Живой Postgres обязателен: проверяется поведение SQL-выборки, а не то, какие
|
||
строки вернул мок. Без БД тест самоскипается (в CI БД есть — ci-tradein.yml
|
||
поднимает Postgres и накатывает схему).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from typing import Any
|
||
from unittest.mock import AsyncMock, patch
|
||
|
||
import pytest
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
from app.services import house_imv_backfill as hib
|
||
|
||
# Фикстуры живут в собственном диапазоне id (ср. tests/test_house_dedup_merge.py).
|
||
_H_STUCK = 900601 # transient_error, попыток 0 → обязан вернуться в очередь
|
||
_H_EXHAUSTED = 900602 # transient_error, попытки исчерпаны → в автоповтор не берём
|
||
_H_PENDING = 900603 # pending → не должен быть вытеснен retry-очередью
|
||
_H_NO_PARAMS = 900604 # pending без объявлений с rooms+area → no_params без запроса
|
||
_IDS = (_H_STUCK, _H_EXHAUSTED, _H_PENDING, _H_NO_PARAMS)
|
||
|
||
|
||
def _live_session() -> Any | None:
|
||
"""Session к достижимому не-заглушечному Postgres, иначе None."""
|
||
try:
|
||
from sqlalchemy import create_engine, text
|
||
from sqlalchemy.orm import sessionmaker
|
||
|
||
dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
|
||
if not dsn or "localhost:5432/test" in dsn:
|
||
return None
|
||
engine = create_engine(dsn, future=True)
|
||
conn = engine.connect()
|
||
conn.execute(text("SELECT 1"))
|
||
conn.close()
|
||
return sessionmaker(bind=engine, future=True)()
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _seed(db: Any) -> None:
|
||
from sqlalchemy import text as _t
|
||
|
||
# url/source/ext_house_id — NOT NULL в houses (002_core_tables).
|
||
db.execute(
|
||
_t(
|
||
"INSERT INTO houses (id, source, ext_house_id, url, address, lat, lon, "
|
||
" imv_status, imv_transient_attempts, last_imv_attempt_at) VALUES "
|
||
"(:s, 'avito', 'EXT-2674-S', 'u', 'тестдом 2674, 1', 56.84, 60.60, "
|
||
" 'transient_error', 0, NOW() - INTERVAL '10 days'),"
|
||
"(:x, 'avito', 'EXT-2674-X', 'u', 'тестдом 2674, 2', 56.84, 60.60, "
|
||
" 'transient_error', :cap, NOW() - INTERVAL '10 days'),"
|
||
"(:p, 'avito', 'EXT-2674-P', 'u', 'тестдом 2674, 3', 56.84, 60.60, "
|
||
" 'pending', 0, NULL),"
|
||
"(:n, 'avito', 'EXT-2674-N', 'u', 'тестдом 2674, 4', 56.84, 60.60, "
|
||
" 'pending', 0, NULL)"
|
||
),
|
||
{
|
||
"s": _H_STUCK,
|
||
"x": _H_EXHAUSTED,
|
||
"p": _H_PENDING,
|
||
"n": _H_NO_PARAMS,
|
||
# Литерал, а не hib._MAX_TRANSIENT_ATTEMPTS: подготовка теста обязана
|
||
# работать и на СТАРОМ коде, иначе красный прогон получится про
|
||
# отсутствующий атрибут, а не про застрявший дом. Синхронность
|
||
# литерала и константы стережёт test_transient_attempts_counter_*.
|
||
"cap": 3,
|
||
},
|
||
)
|
||
# Объявления с rooms+area+house_type — у всех, КРОМЕ _H_NO_PARAMS.
|
||
# source_url / dedup_hash / price_rub — NOT NULL, dedup_hash UNIQUE.
|
||
db.execute(
|
||
_t(
|
||
"INSERT INTO listings (id, source, source_url, source_id, dedup_hash, price_rub, "
|
||
" house_id_fk, rooms, area_m2, floor, total_floors, house_type) "
|
||
"VALUES "
|
||
"(:l1, 'avito', 'http://t/2674/1', 'L-2674-1', 'dh-2674-1', 5000000, :s, "
|
||
" 2, 52.0, 3, 9, 'brick'),"
|
||
"(:l2, 'avito', 'http://t/2674/2', 'L-2674-2', 'dh-2674-2', 5000000, :x, "
|
||
" 2, 52.0, 3, 9, 'brick'),"
|
||
"(:l3, 'avito', 'http://t/2674/3', 'L-2674-3', 'dh-2674-3', 5000000, :p, "
|
||
" 2, 52.0, 3, 9, 'brick')"
|
||
),
|
||
{
|
||
"l1": 910601,
|
||
"l2": 910602,
|
||
"l3": 910603,
|
||
"s": _H_STUCK,
|
||
"x": _H_EXHAUSTED,
|
||
"p": _H_PENDING,
|
||
},
|
||
)
|
||
db.commit()
|
||
|
||
|
||
def _cleanup(db: Any) -> None:
|
||
from sqlalchemy import text as _t
|
||
|
||
db.rollback()
|
||
db.execute(_t("DELETE FROM listings WHERE id IN (910601, 910602, 910603)"))
|
||
db.execute(_t("DELETE FROM houses WHERE id = ANY(CAST(:ids AS bigint[]))"), {"ids": list(_IDS)})
|
||
db.commit()
|
||
db.close()
|
||
|
||
|
||
async def _run(db: Any, **kwargs: Any) -> Any:
|
||
"""Прогнать backfill с замоканным вызовом площадки.
|
||
|
||
Мокаем ровно сетевую границу: evaluate_via_imv поднимает «адрес не найден»,
|
||
поэтому дом уходит в not_found и путь сохранения (три таблицы) не нужен —
|
||
проверяем именно ВЫБОРКУ. Флаг браузерного фетчера гасим, чтобы не поднимать
|
||
сайдкар.
|
||
"""
|
||
err = hib.IMVAddressNotFoundError("address not found (test)")
|
||
with (
|
||
patch.object(hib.settings, "avito_imv_use_browser_fetcher", False),
|
||
patch.object(hib, "evaluate_via_imv", AsyncMock(side_effect=err)),
|
||
):
|
||
return await hib.backfill_house_imv(db, request_delay_sec=0.0, **kwargs)
|
||
|
||
|
||
@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
|
||
async def test_stuck_transient_house_returns_to_the_queue_by_itself() -> None:
|
||
"""Прогон с дефолтными параметрами (как у расписания) забирает застрявший дом.
|
||
|
||
На старом коде выборка — `WHERE imv_status = :status` с единственным
|
||
значением 'pending', поэтому дом остаётся в 'transient_error' навсегда.
|
||
"""
|
||
from sqlalchemy import text as _t
|
||
|
||
db = _live_session()
|
||
assert db is not None
|
||
if db.execute(_t("SELECT count(*) FROM houses")).scalar():
|
||
pytest.skip("тестовая БД не пуста — фикстуры не попадут в голову очереди")
|
||
try:
|
||
_seed(db)
|
||
|
||
# Ровно то, что вызывает расписание: batch_size из default_params, статус по умолчанию.
|
||
result = await _run(db, batch_size=50)
|
||
|
||
picked = {
|
||
r.id: r.imv_status
|
||
for r in db.execute(
|
||
_t("SELECT id, imv_status FROM houses WHERE id = ANY(CAST(:ids AS bigint[]))"),
|
||
{"ids": list(_IDS)},
|
||
)
|
||
}
|
||
assert picked[_H_STUCK] != "transient_error", (
|
||
"дом с временной ошибкой и непотраченным лимитом обязан попасть в пакет "
|
||
f"обычного прогона; фактически статус {picked[_H_STUCK]!r} "
|
||
f"(в пакет прогона попало домов: {result.checked})"
|
||
)
|
||
# pending не вытеснен повтором.
|
||
assert picked[_H_PENDING] != "pending", "pending-дом обязан остаться в том же пакете"
|
||
# Исчерпавший лимит в автоповтор не берётся — иначе вечный цикл.
|
||
assert picked[_H_EXHAUSTED] == "transient_error"
|
||
assert result.retried == 1
|
||
# Дом без объявлений с rooms+area помечен ДО пакета, слот не потрачен.
|
||
assert picked[_H_NO_PARAMS] == "no_params"
|
||
assert result.premarked == 1
|
||
finally:
|
||
_cleanup(db)
|
||
|
||
|
||
@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
|
||
async def test_explicit_only_status_still_takes_exhausted_houses() -> None:
|
||
"""Явный only_status — ручной запрос оператора: лимит попыток его не режет."""
|
||
from sqlalchemy import text as _t
|
||
|
||
db = _live_session()
|
||
assert db is not None
|
||
if db.execute(_t("SELECT count(*) FROM houses")).scalar():
|
||
pytest.skip("тестовая БД не пуста — фикстуры не попадут в голову очереди")
|
||
try:
|
||
_seed(db)
|
||
|
||
result = await _run(db, batch_size=50, only_status="transient_error")
|
||
|
||
statuses = {
|
||
r.id: r.imv_status
|
||
for r in db.execute(
|
||
_t("SELECT id, imv_status FROM houses WHERE id = ANY(CAST(:ids AS bigint[]))"),
|
||
{"ids": list(_IDS)},
|
||
)
|
||
}
|
||
assert (
|
||
statuses[_H_EXHAUSTED] != "transient_error"
|
||
), "явно запрошенный статус обрабатывается целиком, включая исчерпавшие лимит"
|
||
# Автоповтора поверх явного запроса нет: pending не тронут.
|
||
assert statuses[_H_PENDING] == "pending"
|
||
assert result.retried == 0
|
||
finally:
|
||
_cleanup(db)
|
||
|
||
|
||
@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
|
||
async def test_transient_attempts_counter_only_counts_transient() -> None:
|
||
"""Счётчик — это «сколько раз подряд упал по временной причине», и он обнуляем.
|
||
|
||
Без верхней границы у повтора не было бы выхода; без обнуления успехом дом,
|
||
однажды поймавший 3 сетевых сбоя, навсегда терял бы право на повтор.
|
||
"""
|
||
from sqlalchemy import text as _t
|
||
|
||
db = _live_session()
|
||
assert db is not None
|
||
if db.execute(_t("SELECT count(*) FROM houses")).scalar():
|
||
pytest.skip("тестовая БД не пуста — фикстуры не попадут в голову очереди")
|
||
try:
|
||
_seed(db)
|
||
assert hib._MAX_TRANSIENT_ATTEMPTS == 3, "литерал 'cap' в _seed рассинхронизирован"
|
||
|
||
def _attempts() -> int:
|
||
return int(
|
||
db.execute(
|
||
_t("SELECT imv_transient_attempts FROM houses WHERE id = :i"),
|
||
{"i": _H_STUCK},
|
||
).scalar()
|
||
)
|
||
|
||
hib._mark_status(db, _H_STUCK, "transient_error", "boom")
|
||
assert _attempts() == 1
|
||
hib._mark_status(db, _H_STUCK, "not_found", "адрес не найден")
|
||
assert _attempts() == 1, "нетранзиентные исходы счётчик не двигают"
|
||
db.execute(
|
||
_t("UPDATE houses SET imv_status='ok', imv_transient_attempts=0 WHERE id = :i"),
|
||
{"i": _H_STUCK},
|
||
)
|
||
db.commit()
|
||
assert _attempts() == 0
|
||
finally:
|
||
_cleanup(db)
|
||
|
||
|
||
def test_retry_queue_is_not_bound_to_only_status() -> None:
|
||
"""Статика (бежит и без БД): retry-выборка не зависит от параметра only_status.
|
||
|
||
Ровно этот дефект и застрял на проде: единственный путь к 'transient_error'
|
||
шёл через значение, которое никто не передавал.
|
||
"""
|
||
retry_sql = str(hib._RETRY_QUEUE_SQL.text)
|
||
assert "imv_status = 'transient_error'" in retry_sql
|
||
assert ":status" not in retry_sql, "retry-очередь обязана быть независимой от only_status"
|
||
assert "imv_transient_attempts < :max_attempts" in retry_sql, "у повтора должен быть выход"
|
||
# psycopg v3: только CAST(:x AS type), никогда :x::type.
|
||
for sql in (retry_sql, str(hib._QUEUE_SQL.text), str(hib._PREMARK_UNUSABLE_SQL.text)):
|
||
assert "::" not in sql
|