272 lines
9.7 KiB
Python
272 lines
9.7 KiB
Python
from __future__ import annotations
|
|
|
|
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.tasks.avito_detail_backfill import ( # noqa: E402
|
|
AvitoDetailBackfillResult,
|
|
run_avito_detail_backfill,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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"
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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_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
|
|
|
|
|
|
@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()
|