"""Test anti-bot hardening: blocked exceptions + early abort."""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from app.services.scrapers.avito_exceptions import (
AvitoBlockedError,
AvitoContentBlockedError,
AvitoRateLimitedError,
)
# ── fetch_detail tests ──────────────────────────────────────────
@pytest.mark.asyncio
async def test_fetch_detail_raises_blocked_on_403() -> None:
"""fetch_detail должен raise AvitoBlockedError на HTTP 403."""
from app.services.scrapers.avito_detail import fetch_detail
mock_session = AsyncMock()
mock_response = MagicMock()
mock_response.status_code = 403
mock_session.get = AsyncMock(return_value=mock_response)
with pytest.raises(AvitoBlockedError, match="403"):
await fetch_detail("/ekaterinburg/kvartiry/test-123", cffi_session=mock_session)
@pytest.mark.asyncio
async def test_fetch_detail_raises_rate_limited_on_429() -> None:
"""fetch_detail должен raise AvitoRateLimitedError на HTTP 429."""
from app.services.scrapers.avito_detail import fetch_detail
mock_session = AsyncMock()
mock_response = MagicMock()
mock_response.status_code = 429
mock_session.get = AsyncMock(return_value=mock_response)
with pytest.raises(AvitoRateLimitedError, match="429"):
await fetch_detail("/ekaterinburg/kvartiry/test-123", cffi_session=mock_session)
@pytest.mark.asyncio
async def test_fetch_detail_raises_value_error_on_other_non_200() -> None:
"""fetch_detail должен raise ValueError на прочие non-200 (не 403/429)."""
from app.services.scrapers.avito_detail import fetch_detail
mock_session = AsyncMock()
mock_response = MagicMock()
mock_response.status_code = 503
mock_session.get = AsyncMock(return_value=mock_response)
with pytest.raises(ValueError, match="503"):
await fetch_detail("/ekaterinburg/kvartiry/test-123", cffi_session=mock_session)
# ── SERP (avito.py) tests ───────────────────────────────────────
@pytest.mark.asyncio
async def test_fetch_around_raises_content_blocked_on_page1_empty_html() -> None:
"""fetch_around должен raise AvitoContentBlockedError на HTTP 200 + 0 карточек на странице 1."""
from app.services.scrapers.avito import AvitoScraper
scraper = AvitoScraper()
mock_session = AsyncMock()
mock_response = MagicMock()
mock_response.status_code = 200
# Garbage HTML: нет data-marker="item" → _parse_html вернёт []
mock_response.text = "captcha challenge
"
mock_session.get = AsyncMock(return_value=mock_response)
scraper._cffi = mock_session
with pytest.raises(AvitoContentBlockedError, match="content-block"):
await scraper.fetch_around(56.84, 60.60, 1000, pages=1)
@pytest.mark.asyncio
async def test_fetch_around_content_blocked_is_subtype_of_blocked() -> None:
"""AvitoContentBlockedError должен быть поймать как AvitoBlockedError (наследование)."""
from app.services.scrapers.avito import AvitoScraper
scraper = AvitoScraper()
mock_session = AsyncMock()
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.text = ""
mock_session.get = AsyncMock(return_value=mock_response)
scraper._cffi = mock_session
with pytest.raises(AvitoBlockedError):
await scraper.fetch_around(56.84, 60.60, 1000, pages=1)
@pytest.mark.asyncio
async def test_fetch_around_page_gt1_empty_returns_gracefully() -> None:
"""fetch_around на page>1 с 0 карточек должен тихо завершить пагинацию (не raise)."""
from app.services.scrapers.avito import AvitoScraper
scraper = AvitoScraper()
mock_session = AsyncMock()
# Страница 1: нормальный HTML с одной карточкой
valid_html = (
""
'"
""
)
# Страница 2: пустой HTML → end of pagination
empty_html = "no listings
"
page1_resp = MagicMock()
page1_resp.status_code = 200
page1_resp.text = valid_html
page2_resp = MagicMock()
page2_resp.status_code = 200
page2_resp.text = empty_html
mock_session.get = AsyncMock(side_effect=[page1_resp, page2_resp])
scraper._cffi = mock_session
with patch("app.services.scrapers.avito.AvitoScraper.sleep_between_requests", AsyncMock()):
result = await scraper.fetch_around(56.84, 60.60, 1000, pages=2)
# Должен вернуть лоты со страницы 1, не raise на странице 2
assert len(result) >= 1
@pytest.mark.asyncio
async def test_fetch_around_raises_blocked_on_403() -> None:
"""AvitoScraper.fetch_around должен raise AvitoBlockedError на HTTP 403."""
from app.services.scrapers.avito import AvitoScraper
scraper = AvitoScraper()
mock_session = AsyncMock()
mock_response = MagicMock()
mock_response.status_code = 403
mock_session.get = AsyncMock(return_value=mock_response)
scraper._cffi = mock_session
with pytest.raises(AvitoBlockedError):
await scraper.fetch_around(56.84, 60.60, 1000, pages=1)
@pytest.mark.asyncio
async def test_fetch_around_raises_rate_limited_on_429() -> None:
"""AvitoScraper.fetch_around должен raise AvitoRateLimitedError на persistent HTTP 429.
429 (backconnect conn-limit) сначала проходит короткие retry; при отсутствии
rotate-url после исчерпания retry сохраняется старый fallback — AvitoRateLimitedError.
"""
from app.services.scrapers.avito import AvitoScraper
scraper = AvitoScraper()
mock_session = AsyncMock()
mock_response = MagicMock()
mock_response.status_code = 429
mock_session.get = AsyncMock(return_value=mock_response)
scraper._cffi = mock_session
with patch("app.services.scrapers.avito.asyncio.sleep", AsyncMock()):
with pytest.raises(AvitoRateLimitedError):
await scraper.fetch_around(56.84, 60.60, 1000, pages=1)
# ── #623 firewall-детект + ротация IP ────────────────────────────
#
# Legacy `run_avito_pipeline` / `run_avito_city_sweep` orchestration-level tests
# (pipeline consecutive-block abort, inter-detail sleep, city-sweep mark_banned)
# удалены вместе с `app.services.scrape_pipeline` (#2397 Part E1) — эквивалентная
# regression-coverage теперь в `test_scraper_kit_pipeline_parity.py` (kit
# `run_avito_city_sweep`: ban/rotation state-machine, detail-block propagation).
_CARD_HTML = (
""
'"
""
)
# Firewall-интерстициал Avito (HTTP 200, не 403) — IP-блок datacenter.
_FIREWALL_HTML = (
"Доступ ограничен: проблема с IP"
'...
'
)
@pytest.mark.asyncio
async def test_fetch_around_raises_blocked_on_firewall_html_200() -> None:
"""HTTP 200 + firewall-страница «Доступ ограничен» → AvitoBlockedError (#623).
Avito банит datacenter-IP отдачей firewall-интерстициала с кодом 200 (не 403) —
раньше это молча проходило как «0 cards / content-block», теперь явный IP-блок.
"""
from app.services.scrapers.avito import AvitoScraper
scraper = AvitoScraper()
mock_session = AsyncMock()
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.text = _FIREWALL_HTML
mock_session.get = AsyncMock(return_value=mock_response)
scraper._cffi = mock_session
with pytest.raises(AvitoBlockedError, match="firewall"):
await scraper.fetch_around(56.84, 60.60, 1000, pages=1)
@pytest.mark.asyncio
async def test_fetch_around_rotates_ip_then_succeeds(monkeypatch: pytest.MonkeyPatch) -> None:
"""При firewall + заданной changeip-ссылке: ротация IP и повтор страницы (#623)."""
from app.services.scrapers import avito as avito_mod
from app.services.scrapers.avito import AvitoScraper
monkeypatch.setattr(
avito_mod.settings, "avito_proxy_rotate_url", "https://changeip.example/?proxy_key=x"
)
monkeypatch.setattr(avito_mod.settings, "avito_proxy_max_rotations", 1)
scraper = AvitoScraper()
fw = MagicMock()
fw.status_code = 200
fw.text = _FIREWALL_HTML
ok = MagicMock()
ok.status_code = 200
ok.text = _CARD_HTML
mock_session = AsyncMock()
mock_session.get = AsyncMock(side_effect=[fw, ok])
scraper._cffi = mock_session
with (
patch.object(AvitoScraper, "_rotate_ip", AsyncMock(return_value=True)) as rot,
patch.object(AvitoScraper, "sleep_between_requests", AsyncMock()),
):
result = await scraper.fetch_around(56.84, 60.60, 1000, pages=1)
rot.assert_awaited_once()
assert len(result) == 1
assert mock_session.get.await_count == 2 # firewall → rotate → retry OK
@pytest.mark.asyncio
async def test_fetch_around_blocked_when_rotation_exhausted(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Если ротация не сняла бан (max_rotations исчерпан) → AvitoBlockedError (#623)."""
from app.services.scrapers import avito as avito_mod
from app.services.scrapers.avito import AvitoScraper
monkeypatch.setattr(
avito_mod.settings, "avito_proxy_rotate_url", "https://changeip.example/?proxy_key=x"
)
monkeypatch.setattr(avito_mod.settings, "avito_proxy_max_rotations", 1)
scraper = AvitoScraper()
fw = MagicMock()
fw.status_code = 200
fw.text = _FIREWALL_HTML
mock_session = AsyncMock()
mock_session.get = AsyncMock(return_value=fw) # всегда firewall
scraper._cffi = mock_session
with patch.object(AvitoScraper, "_rotate_ip", AsyncMock(return_value=True)) as rot:
with pytest.raises(AvitoBlockedError, match="firewall"):
await scraper.fetch_around(56.84, 60.60, 1000, pages=1)
assert rot.await_count == 1 # одна попытка ротации, потом сдаёмся
# ── #623 sweep-pipeline proxy (SERP+detail через mobile proxy) ───
def test_avito_proxies_helper() -> None:
"""Kit _avito_proxies(config): None без proxy, dict {http,https} при заданном URL.
Регрессия: sweep создаёт shared-сессию, через неё идут и SERP, и detail —
без прокси detail летит с datacenter-IP → 429 → run banned (#623 followup).
"""
from types import SimpleNamespace
from scraper_kit.orchestration.pipeline import _avito_proxies
assert _avito_proxies(SimpleNamespace(scraper_proxy_url=None)) is None
url = "http://u:p@mproxy.site:16928"
assert _avito_proxies(SimpleNamespace(scraper_proxy_url=url)) == {"http": url, "https": url}
# ── curl_cffi 429 short-retry (backconnect conn-limit transient) ──────────────
@pytest.mark.asyncio
async def test_fetch_around_429_short_retry_then_succeeds() -> None:
"""429 дважды → 200: короткий retry снимает transient conn-limit БЕЗ ротации IP.
backconnect-прокси (mproxy.site, лимит 5 соединений) отдаёт 429 при параллельной
пагинации — это transient, проходит на коротком retry. _rotate_ip НЕ должен дёргаться.
"""
from app.services.scrapers.avito import AvitoScraper
scraper = AvitoScraper()
r429 = MagicMock()
r429.status_code = 429
ok = MagicMock()
ok.status_code = 200
ok.text = _CARD_HTML
mock_session = AsyncMock()
mock_session.get = AsyncMock(side_effect=[r429, r429, ok])
scraper._cffi = mock_session
with (
patch("app.services.scrapers.avito.asyncio.sleep", AsyncMock()) as slp,
patch.object(AvitoScraper, "_rotate_ip", AsyncMock(return_value=True)) as rot,
patch.object(AvitoScraper, "sleep_between_requests", AsyncMock()),
):
result = await scraper.fetch_around(56.84, 60.60, 1000, pages=1)
assert len(result) == 1
assert mock_session.get.await_count == 3 # 429, 429, 200
rot.assert_not_awaited() # короткий retry, не ротация
# короткий backoff (_AVITO_429_BACKOFF_SEC ~1.5s), не 9с-settle ротации
assert slp.await_count >= 2
for c in slp.await_args_list:
assert c.args[0] < 5.0
@pytest.mark.asyncio
async def test_fetch_around_429_exhausted_no_rotate_url_raises() -> None:
"""429 постоянно + НЕТ rotate-url → после исчерпания коротких retry: старый fallback."""
from app.services.scrapers import avito as avito_mod
from app.services.scrapers.avito import AvitoScraper
scraper = AvitoScraper()
r429 = MagicMock()
r429.status_code = 429
mock_session = AsyncMock()
mock_session.get = AsyncMock(return_value=r429)
scraper._cffi = mock_session
with (
patch("app.services.scrapers.avito.asyncio.sleep", AsyncMock()) as slp,
patch.object(AvitoScraper, "_rotate_ip", AsyncMock(return_value=True)) as rot,
):
with pytest.raises(AvitoRateLimitedError):
await scraper.fetch_around(56.84, 60.60, 1000, pages=1)
rot.assert_not_awaited() # нет rotate-url → ротация не дёргается
# ровно _AVITO_429_MAX_RETRIES коротких retry, потом fallback
assert slp.await_count == avito_mod._AVITO_429_MAX_RETRIES
@pytest.mark.asyncio
async def test_fetch_around_429_exhausted_with_rotate_url_rotates(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""429 постоянно + ЕСТЬ rotate-url → короткие retry, затем ротация (старый путь)."""
from app.services.scrapers import avito as avito_mod
from app.services.scrapers.avito import AvitoScraper
monkeypatch.setattr(
avito_mod.settings, "avito_proxy_rotate_url", "https://changeip.example/?proxy_key=x"
)
monkeypatch.setattr(avito_mod.settings, "avito_proxy_max_rotations", 1)
scraper = AvitoScraper()
r429 = MagicMock()
r429.status_code = 429
ok = MagicMock()
ok.status_code = 200
ok.text = _CARD_HTML
mock_session = AsyncMock()
# 429 на всех коротких retry (1 первичный + _AVITO_429_MAX_RETRIES), после
# исчерпания short-retry — ротация → 200. backconnect-reconnect здесь НЕ
# срабатывает (scraper_proxy_url не задан), поэтому сразу rotate-путь.
n_429 = 1 + avito_mod._AVITO_429_MAX_RETRIES
mock_session.get = AsyncMock(side_effect=[r429] * n_429 + [ok])
scraper._cffi = mock_session
with (
patch("app.services.scrapers.avito.asyncio.sleep", AsyncMock()),
patch.object(AvitoScraper, "_rotate_ip", AsyncMock(return_value=True)) as rot,
patch.object(AvitoScraper, "sleep_between_requests", AsyncMock()),
):
result = await scraper.fetch_around(56.84, 60.60, 1000, pages=1)
rot.assert_awaited_once() # после исчерпания коротких retry — ротация
assert len(result) == 1
@pytest.mark.asyncio
async def test_fetch_around_403_no_429_retry_path(monkeypatch: pytest.MonkeyPatch) -> None:
"""403 (реальный IP-ban) НЕ проходит через 429-retry — поведение без изменений."""
from app.services.scrapers import avito as avito_mod
from app.services.scrapers.avito import AvitoScraper
monkeypatch.setattr(avito_mod.settings, "avito_proxy_rotate_url", None)
scraper = AvitoScraper()
r403 = MagicMock()
r403.status_code = 403
mock_session = AsyncMock()
mock_session.get = AsyncMock(return_value=r403)
scraper._cffi = mock_session
with (
patch("app.services.scrapers.avito.asyncio.sleep", AsyncMock()) as slp,
patch.object(AvitoScraper, "_rotate_ip", AsyncMock(return_value=True)),
):
with pytest.raises(AvitoBlockedError):
await scraper.fetch_around(56.84, 60.60, 1000, pages=1)
# 403 без rotate-url → немедленный raise, ни одного 429-backoff sleep
slp.assert_not_awaited()
assert mock_session.get.await_count == 1