bot-backend
848e6d3bdd
Деплой (docker recreate tradein-scraper) шлёт SIGTERM с stop_grace_period=120s (Phase 1). Раньше scheduler_main отвечал hard task.cancel() → бегущий scrape-unit получал CancelledError посреди браузер-карточки → карточка гибла без COMMIT. Теперь SIGTERM/SIGINT лишь выставляют кооперативный флаг; tick-loop и длинные задачи опрашивают его на СВОИХ существующих between-unit checkpoint'ах, докоммичивают текущий unit и выходят сами. - app/core/shutdown.py: новый standalone-модуль (asyncio.Event + request_shutdown / shutdown_requested / wait_for_shutdown), без app-зависимостей → нет циклов. - scheduler_main._run: SIGTERM → request_shutdown() вместо task.cancel(); ждём добровольного drain'а, safety-net wait_for(100s < docker grace) с fallback на hard-cancel для некооперирующей задачи. - scheduler_loop: проверка флага в начале тика и после каждого dispatch — не reap/claim новые run'ы во время shutdown, выходим из loop'а. - avito_detail_backfill: break на границе карточки рядом с budget-guard; snapshot pending-query идемпотентен → следующий run сам резюмит остаток. - import_rosreestr_dkp: расширен is_cancelled checkpoint — drain делает mark_done (partial), не mark_cancelled; user-cancel семантика без изменений. Без SIGTERM поведение идентично прежнему. reap_zombies не трогает drained-run'ы (они mark_done, не 'running'). Tests: tests/core/test_shutdown.py, scheduler_main drain+timeout-fallback, avito_detail_backfill partial-drain. 26 + 29 scheduler passed.
512 lines
19 KiB
Python
512 lines
19 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import os
|
||
import sys
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
_wp_mock = MagicMock()
|
||
sys.modules.setdefault("weasyprint", _wp_mock)
|
||
|
||
import pytest # noqa: E402
|
||
|
||
from app.core import shutdown as _sd # noqa: E402
|
||
from app.tasks.avito_detail_backfill import ( # noqa: E402
|
||
AvitoDetailBackfillResult,
|
||
run_avito_detail_backfill,
|
||
)
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _reset_shutdown() -> None:
|
||
"""shutdown — module-global Event: чистим вокруг каждого теста (изоляция #1182)."""
|
||
_sd.reset_shutdown()
|
||
yield
|
||
_sd.reset_shutdown()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _make_snapshot(n: int) -> list[dict]:
|
||
return [{"id": i + 1, "source_url": f"/items/{i + 1}"} for i in range(n)]
|
||
|
||
|
||
def _mock_db(snapshot: list[dict]) -> MagicMock:
|
||
"""Fake Session: first execute() returns snapshot via .mappings().all()."""
|
||
db = MagicMock()
|
||
sel = MagicMock()
|
||
sel.mappings.return_value.all.return_value = snapshot
|
||
db.execute.return_value = sel
|
||
return db
|
||
|
||
|
||
_FETCH = "app.tasks.avito_detail_backfill.fetch_detail"
|
||
_SAVE = "app.tasks.avito_detail_backfill.save_detail_enrichment"
|
||
_RUNS = "app.tasks.avito_detail_backfill.runs_mod"
|
||
_SLEEP = "app.tasks.avito_detail_backfill.asyncio.sleep"
|
||
_SESSION = "app.tasks.avito_detail_backfill.AsyncSession"
|
||
_SCRAPER = "app.tasks.avito_detail_backfill.AvitoScraper"
|
||
_SETTINGS = "app.tasks.avito_detail_backfill.settings"
|
||
_SHUTDOWN = "app.tasks.avito_detail_backfill.shutdown_requested"
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_empty_snapshot_marks_done() -> None:
|
||
"""Empty snapshot -> mark_done immediately, no fetch calls."""
|
||
db = _mock_db([])
|
||
runs = MagicMock()
|
||
fake_session = AsyncMock()
|
||
fake_settings = MagicMock(scraper_fetch_mode="cffi")
|
||
with (
|
||
patch(_SETTINGS, fake_settings),
|
||
patch(_SESSION, return_value=fake_session),
|
||
patch(_SCRAPER),
|
||
patch(_RUNS, runs),
|
||
patch(_FETCH) as mock_fetch,
|
||
):
|
||
result = await run_avito_detail_backfill(
|
||
db, run_id=1, params={"batch_size": 10, "budget_sec": 60}
|
||
)
|
||
|
||
assert isinstance(result, AvitoDetailBackfillResult)
|
||
assert result.attempted == 0
|
||
assert result.enriched == 0
|
||
mock_fetch.assert_not_called()
|
||
runs.mark_done.assert_called_once()
|
||
runs.mark_failed.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_processes_snapshot_to_completion() -> None:
|
||
"""3 listings -> all fetched and enriched, mark_done called."""
|
||
snapshot = _make_snapshot(3)
|
||
db = _mock_db(snapshot)
|
||
runs = MagicMock()
|
||
mock_enrichment = MagicMock()
|
||
mock_fetch = AsyncMock(return_value=mock_enrichment)
|
||
mock_save = MagicMock(return_value=True)
|
||
fake_settings = MagicMock(scraper_fetch_mode="cffi")
|
||
with (
|
||
patch(_SETTINGS, fake_settings),
|
||
patch(_SESSION, return_value=AsyncMock()),
|
||
patch(_SCRAPER),
|
||
patch(_RUNS, runs),
|
||
patch(_FETCH, mock_fetch),
|
||
patch(_SAVE, mock_save),
|
||
patch(_SLEEP, new_callable=AsyncMock),
|
||
):
|
||
result = await run_avito_detail_backfill(
|
||
db, run_id=2, params={"batch_size": 10, "budget_sec": 3600}
|
||
)
|
||
|
||
assert result.attempted == 3
|
||
assert result.enriched == 3
|
||
assert result.blocked == 0
|
||
assert result.failed == 0
|
||
assert mock_fetch.call_count == 3
|
||
runs.mark_done.assert_called_once()
|
||
runs.mark_failed.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_blocked_abort_after_max_consecutive() -> None:
|
||
"""5 consecutive AvitoBlockedError -> abort, mark_done (NOT mark_failed), rotate_ip x5."""
|
||
from app.services.scrapers.avito_exceptions import AvitoBlockedError
|
||
|
||
snapshot = _make_snapshot(10)
|
||
db = _mock_db(snapshot)
|
||
runs = MagicMock()
|
||
blocked_exc = AvitoBlockedError("ip blocked")
|
||
mock_fetch = AsyncMock(side_effect=blocked_exc)
|
||
mock_scraper = MagicMock()
|
||
mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
|
||
fake_settings = MagicMock(scraper_fetch_mode="cffi")
|
||
with (
|
||
patch(_SETTINGS, fake_settings),
|
||
patch(_SESSION, return_value=AsyncMock()),
|
||
patch(_SCRAPER, mock_scraper),
|
||
patch(_RUNS, runs),
|
||
patch(_FETCH, mock_fetch),
|
||
patch(_SLEEP, new_callable=AsyncMock),
|
||
):
|
||
result = await run_avito_detail_backfill(
|
||
db, run_id=3, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}
|
||
)
|
||
|
||
assert result.blocked == 5
|
||
assert result.attempted == 5
|
||
assert result.enriched == 0
|
||
runs.mark_done.assert_called_once()
|
||
runs.mark_failed.assert_not_called()
|
||
assert mock_scraper.return_value._rotate_ip.call_count == 5
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_sigterm_drain_breaks_and_marks_done_partial() -> None:
|
||
"""#1182 Phase 2: shutdown_requested() True → loop выходит на границе карточки,
|
||
mark_done вызывается с ЧАСТИЧНЫМИ счётчиками (не mark_failed, не mark_cancelled).
|
||
|
||
side_effect [False, True]: 1-я карточка обрабатывается (attempted=1, enriched=1),
|
||
перед 2-й приходит SIGTERM-drain → break. Snapshot — pending-query, остаток
|
||
до-резюмит следующий run (resume_cursor не нужен).
|
||
"""
|
||
snapshot = _make_snapshot(3)
|
||
db = _mock_db(snapshot)
|
||
runs = MagicMock()
|
||
mock_enrichment = MagicMock()
|
||
mock_fetch = AsyncMock(return_value=mock_enrichment)
|
||
mock_save = MagicMock(return_value=True)
|
||
fake_settings = MagicMock(scraper_fetch_mode="cffi")
|
||
with (
|
||
patch(_SETTINGS, fake_settings),
|
||
patch(_SESSION, return_value=AsyncMock()),
|
||
patch(_SCRAPER),
|
||
patch(_RUNS, runs),
|
||
patch(_FETCH, mock_fetch),
|
||
patch(_SAVE, mock_save),
|
||
patch(_SLEEP, new_callable=AsyncMock),
|
||
patch(_SHUTDOWN, side_effect=[False, True]),
|
||
):
|
||
result = await run_avito_detail_backfill(
|
||
db, run_id=13, params={"batch_size": 10, "budget_sec": 3600}
|
||
)
|
||
|
||
# Обработана только 1-я карточка, на 2-й — drain-break.
|
||
assert result.attempted == 1
|
||
assert result.enriched == 1
|
||
assert mock_fetch.call_count == 1
|
||
runs.mark_done.assert_called_once()
|
||
runs.mark_failed.assert_not_called()
|
||
runs.mark_cancelled.assert_not_called()
|
||
# mark_done получил ЧАСТИЧНЫЕ счётчики (attempted=1, а не весь snapshot=3).
|
||
done_counters = runs.mark_done.call_args.args[2]
|
||
assert done_counters["attempted"] == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_budget_guard_stops_loop() -> None:
|
||
"""Budget expired before first listing -> fetch_detail not called."""
|
||
snapshot = _make_snapshot(5)
|
||
db = _mock_db(snapshot)
|
||
runs = MagicMock()
|
||
mono_values = iter([0.0, 999.0, 999.0])
|
||
fake_settings = MagicMock(scraper_fetch_mode="cffi")
|
||
with (
|
||
patch(_SETTINGS, fake_settings),
|
||
patch(_SESSION, return_value=AsyncMock()),
|
||
patch(_SCRAPER),
|
||
patch(_RUNS, runs),
|
||
patch("app.tasks.avito_detail_backfill.time.monotonic", side_effect=mono_values),
|
||
patch(_FETCH) as mock_fetch,
|
||
):
|
||
await run_avito_detail_backfill(db, run_id=4, params={"batch_size": 5, "budget_sec": 1})
|
||
|
||
mock_fetch.assert_not_called()
|
||
runs.mark_done.assert_called_once()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_top_level_exception_marks_failed() -> None:
|
||
"""db.execute raises -> mark_failed called, exception re-raised."""
|
||
db = MagicMock()
|
||
db.execute.side_effect = RuntimeError("DB connection lost")
|
||
runs = MagicMock()
|
||
fake_settings = MagicMock(scraper_fetch_mode="cffi")
|
||
with (
|
||
patch(_SETTINGS, fake_settings),
|
||
patch(_SESSION, return_value=AsyncMock()),
|
||
patch(_SCRAPER),
|
||
patch(_RUNS, runs),
|
||
):
|
||
with pytest.raises(RuntimeError, match="DB connection lost"):
|
||
await run_avito_detail_backfill(
|
||
db, run_id=5, params={"batch_size": 5, "budget_sec": 60}
|
||
)
|
||
|
||
runs.mark_failed.assert_called_once()
|
||
runs.mark_done.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_rotate_ip_called_on_each_block() -> None:
|
||
"""1 block + 1 success -> rotate_ip called once, enriched=1."""
|
||
from app.services.scrapers.avito_exceptions import AvitoBlockedError
|
||
|
||
snapshot = _make_snapshot(2)
|
||
db = _mock_db(snapshot)
|
||
runs = MagicMock()
|
||
mock_enrichment = MagicMock()
|
||
blocked_exc = AvitoBlockedError("x")
|
||
mock_fetch = AsyncMock(side_effect=[blocked_exc, mock_enrichment])
|
||
mock_scraper = MagicMock()
|
||
mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
|
||
fake_settings = MagicMock(scraper_fetch_mode="cffi")
|
||
with (
|
||
patch(_SETTINGS, fake_settings),
|
||
patch(_SESSION, return_value=AsyncMock()),
|
||
patch(_SCRAPER, mock_scraper),
|
||
patch(_RUNS, runs),
|
||
patch(_FETCH, mock_fetch),
|
||
patch(_SAVE, return_value=True),
|
||
patch(_SLEEP, new_callable=AsyncMock),
|
||
):
|
||
result = await run_avito_detail_backfill(
|
||
db, run_id=6, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}
|
||
)
|
||
|
||
assert result.enriched == 1
|
||
assert result.blocked == 1
|
||
assert mock_scraper.return_value._rotate_ip.call_count == 1
|
||
runs.mark_done.assert_called_once()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_snapshot_filters_ekb_active_only() -> None:
|
||
"""Снапшот-SELECT (#1814) фильтрует только активные ЕКБ-листинги.
|
||
|
||
Проверяем, что текст запроса содержит `is_active = TRUE` и
|
||
`LIKE '%/ekaterinburg/%'` — legacy не-ЕКБ (moskva/spb/tyumen) и мёртвые
|
||
листинги не попадают в фетч, иначе browser спотыкается → curl-бан 429.
|
||
"""
|
||
db = _mock_db([])
|
||
runs = MagicMock()
|
||
fake_settings = MagicMock(scraper_fetch_mode="cffi")
|
||
with (
|
||
patch(_SETTINGS, fake_settings),
|
||
patch(_SESSION, return_value=AsyncMock()),
|
||
patch(_SCRAPER),
|
||
patch(_RUNS, runs),
|
||
patch(_FETCH),
|
||
):
|
||
await run_avito_detail_backfill(db, run_id=8, params={"batch_size": 10, "budget_sec": 60})
|
||
|
||
# Первый (и единственный при пустом снапшоте) execute — это SELECT-снапшот.
|
||
snapshot_call = db.execute.call_args_list[0]
|
||
sql_text = str(snapshot_call.args[0])
|
||
assert "is_active = TRUE" in sql_text
|
||
assert "/ekaterinburg/" in sql_text
|
||
assert "detail_enriched_at IS NULL" in sql_text
|
||
assert "(lat IS NULL) DESC" in sql_text
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_fetch_exception_continues() -> None:
|
||
"""RuntimeError on one listing -> failed++, db.rollback(), loop continues for next."""
|
||
snapshot = _make_snapshot(2)
|
||
db = _mock_db(snapshot)
|
||
runs = MagicMock()
|
||
mock_enrichment = MagicMock()
|
||
mock_fetch = AsyncMock(side_effect=[RuntimeError("parse error"), mock_enrichment])
|
||
fake_settings = MagicMock(scraper_fetch_mode="cffi")
|
||
with (
|
||
patch(_SETTINGS, fake_settings),
|
||
patch(_SESSION, return_value=AsyncMock()),
|
||
patch(_SCRAPER),
|
||
patch(_RUNS, runs),
|
||
patch(_FETCH, mock_fetch),
|
||
patch(_SAVE, return_value=True),
|
||
patch(_SLEEP, new_callable=AsyncMock),
|
||
):
|
||
result = await run_avito_detail_backfill(
|
||
db, run_id=7, params={"batch_size": 10, "budget_sec": 3600}
|
||
)
|
||
|
||
assert result.failed == 1
|
||
assert result.enriched == 1
|
||
assert result.attempted == 2
|
||
db.rollback.assert_called()
|
||
runs.mark_done.assert_called_once()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_fetch_timeout_skips_and_continues() -> None:
|
||
"""#1950: один fetch_detail зависает дольше hard-timeout → этот листинг failed,
|
||
loop НЕ зависает, переходит к следующему, run завершается mark_done (не zombie).
|
||
|
||
Регрессия run 423 (завис 7.7ч → reaped): fetch_detail без timeout блокировал
|
||
loop навсегда. asyncio.wait_for(timeout) отменяет зависший fetch → TimeoutError.
|
||
"""
|
||
snapshot = _make_snapshot(2)
|
||
db = _mock_db(snapshot)
|
||
runs = MagicMock()
|
||
mock_enrichment = MagicMock()
|
||
|
||
call_urls: list[str] = []
|
||
|
||
async def _fetch(url: str, *, cffi_session: object = None, browser_fetcher: object = None):
|
||
call_urls.append(url)
|
||
if len(call_urls) == 1:
|
||
await asyncio.Event().wait() # висит вечно → wait_for отменит по timeout
|
||
return mock_enrichment
|
||
|
||
# Короткий timeout (50ms) — тест не ждёт реальные 90s; rotate-settle тоже 0.
|
||
fake_settings = MagicMock(
|
||
scraper_fetch_mode="cffi",
|
||
avito_detail_fetch_timeout_s=0.05,
|
||
avito_proxy_rotate_settle_s=0.0,
|
||
)
|
||
with (
|
||
patch(_SETTINGS, fake_settings),
|
||
patch(_SESSION, return_value=AsyncMock()),
|
||
patch(_SCRAPER),
|
||
patch(_RUNS, runs),
|
||
patch(_FETCH, _fetch),
|
||
patch(_SAVE, return_value=True),
|
||
patch(_SLEEP, new_callable=AsyncMock),
|
||
):
|
||
result = await run_avito_detail_backfill(
|
||
db, run_id=11, params={"batch_size": 10, "budget_sec": 3600}
|
||
)
|
||
|
||
# Первый листинг — failed (timeout), второй — enriched. Run завершён mark_done.
|
||
assert result.failed == 1
|
||
assert result.enriched == 1
|
||
assert result.attempted == 2
|
||
assert len(call_urls) == 2, "loop должен дойти до второго листинга, а не зависнуть"
|
||
db.rollback.assert_called()
|
||
runs.mark_done.assert_called_once()
|
||
runs.mark_failed.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_listing_gone_marks_inactive_no_breaker() -> None:
|
||
"""#2034: AvitoListingGoneError (мёртвый 404-листинг) → counters.gone++, blocked
|
||
НЕ растёт, consecutive-block breaker НЕ абортит, листинг помечается is_active=FALSE.
|
||
|
||
Регрессия run 458 (attempted=5 enriched=0 blocked=5 → abort): lat-null очередь
|
||
состоит из dead-листингов; раньше 404 ловился как soft-block → breaker абортил run
|
||
до live-листингов. Теперь 404 нейтрален к breaker'у и метит листинг inactive.
|
||
"""
|
||
from app.services.scrapers.avito_exceptions import AvitoListingGoneError
|
||
|
||
snapshot = _make_snapshot(1)
|
||
db = _mock_db(snapshot)
|
||
runs = MagicMock()
|
||
mock_fetch = AsyncMock(side_effect=AvitoListingGoneError("404 gone"))
|
||
mock_scraper = MagicMock()
|
||
mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
|
||
fake_settings = MagicMock(scraper_fetch_mode="cffi")
|
||
with (
|
||
patch(_SETTINGS, fake_settings),
|
||
patch(_SESSION, return_value=AsyncMock()),
|
||
patch(_SCRAPER, mock_scraper),
|
||
patch(_RUNS, runs),
|
||
patch(_FETCH, mock_fetch),
|
||
patch(_SLEEP, new_callable=AsyncMock),
|
||
):
|
||
result = await run_avito_detail_backfill(
|
||
db,
|
||
run_id=12,
|
||
params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5},
|
||
)
|
||
|
||
assert result.gone == 1
|
||
assert result.blocked == 0
|
||
assert result.attempted == 1
|
||
assert result.enriched == 0
|
||
# breaker НЕ абортил: rotate_ip НЕ дёргался (gone ≠ block), run завершён mark_done.
|
||
mock_scraper.return_value._rotate_ip.assert_not_called()
|
||
runs.mark_done.assert_called_once()
|
||
runs.mark_failed.assert_not_called()
|
||
# UPDATE listings SET is_active = FALSE по row id=1 выполнен (мок db.execute).
|
||
update_calls = [
|
||
c
|
||
for c in db.execute.call_args_list
|
||
if "UPDATE listings SET is_active = FALSE" in str(c.args[0])
|
||
]
|
||
assert len(update_calls) == 1
|
||
assert update_calls[0].args[1] == {"id": 1}
|
||
|
||
|
||
_BROWSER_FETCHER = "app.tasks.avito_detail_backfill.BrowserFetcher"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_use_curl_flag_skips_browser_fetcher() -> None:
|
||
"""avito_detail_backfill_use_curl=True: BrowserFetcher не создаётся,
|
||
fetch_detail вызывается с browser_fetcher=None (curl/backconnect путь).
|
||
"""
|
||
snapshot = _make_snapshot(1)
|
||
db = _mock_db(snapshot)
|
||
runs = MagicMock()
|
||
mock_enrichment = MagicMock()
|
||
mock_fetch = AsyncMock(return_value=mock_enrichment)
|
||
mock_save = MagicMock(return_value=True)
|
||
# Флаг use_curl=True, fetch_mode=browser (но флаг перекрывает)
|
||
fake_settings = MagicMock(
|
||
scraper_fetch_mode="browser",
|
||
avito_detail_backfill_use_curl=True,
|
||
)
|
||
with (
|
||
patch(_SETTINGS, fake_settings),
|
||
patch(_SESSION),
|
||
patch(_SCRAPER),
|
||
patch(_BROWSER_FETCHER) as mock_bf_cls,
|
||
patch(_RUNS, runs),
|
||
patch(_FETCH, mock_fetch),
|
||
patch(_SAVE, mock_save),
|
||
patch(_SLEEP, new_callable=AsyncMock),
|
||
):
|
||
result = await run_avito_detail_backfill(
|
||
db, run_id=9, params={"batch_size": 10, "budget_sec": 3600}
|
||
)
|
||
|
||
# BrowserFetcher не должен быть создан
|
||
mock_bf_cls.assert_not_called()
|
||
# fetch_detail вызван с browser_fetcher=None (curl путь)
|
||
assert mock_fetch.call_count == 1
|
||
_, kwargs = mock_fetch.call_args
|
||
assert kwargs.get("browser_fetcher") is None
|
||
assert result.enriched == 1
|
||
runs.mark_done.assert_called_once()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_backfill_use_curl_false_creates_browser_fetcher() -> None:
|
||
"""avito_detail_backfill_use_curl=False + scraper_fetch_mode='browser':
|
||
BrowserFetcher создаётся (legacy browser/auv путь).
|
||
"""
|
||
snapshot = _make_snapshot(1)
|
||
db = _mock_db(snapshot)
|
||
runs = MagicMock()
|
||
mock_enrichment = MagicMock()
|
||
mock_fetch = AsyncMock(return_value=mock_enrichment)
|
||
mock_save = MagicMock(return_value=True)
|
||
fake_settings = MagicMock(
|
||
scraper_fetch_mode="browser",
|
||
avito_detail_backfill_use_curl=False,
|
||
)
|
||
mock_bf_instance = AsyncMock()
|
||
mock_bf_instance.__aenter__ = AsyncMock(return_value=mock_bf_instance)
|
||
mock_bf_instance.__aexit__ = AsyncMock(return_value=False)
|
||
with (
|
||
patch(_SETTINGS, fake_settings),
|
||
patch(_SESSION),
|
||
patch(_SCRAPER),
|
||
patch(_BROWSER_FETCHER, return_value=mock_bf_instance) as mock_bf_cls,
|
||
patch(_RUNS, runs),
|
||
patch(_FETCH, mock_fetch),
|
||
patch(_SAVE, mock_save),
|
||
patch(_SLEEP, new_callable=AsyncMock),
|
||
):
|
||
result = await run_avito_detail_backfill(
|
||
db, run_id=10, params={"batch_size": 10, "budget_sec": 3600}
|
||
)
|
||
|
||
# BrowserFetcher должен быть создан (source="avito")
|
||
mock_bf_cls.assert_called_once_with(source="avito")
|
||
# fetch_detail вызван с browser_fetcher установленным (не None)
|
||
assert mock_fetch.call_count == 1
|
||
_, kwargs = mock_fetch.call_args
|
||
assert kwargs.get("browser_fetcher") is not None
|
||
assert result.enriched == 1
|
||
runs.mark_done.assert_called_once()
|