"""curl_cffi 403/firewall reconnect-retry под backconnect-прокси.
Под backconnect rotating-прокси (mproxy.site, нет changeip-URL) одиночный HTTP 403
или firewall-интерстициал ронял весь avito_full_load sweep через AvitoBlockedError →
mark_banned. Каждое новое соединение через backconnect даёт свежий exit-IP, поэтому
ретраим пересозданием curl_cffi-сессии до _AVITO_403_MAX_RETRIES раз.
backconnect = scraper_proxy_url задан И avito_proxy_rotate_url НЕ задан. Включаем через
scraper_proxy_url_env (scraper_proxy_url — read-only property, его не monkeypatch'ишь).
"""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from app.services.scrapers import avito as avito_mod
from app.services.scrapers.avito import AvitoScraper
from app.services.scrapers.avito_exceptions import AvitoBlockedError
# Валидная карточка SERP (одна) — successful page после reconnect.
_CARD_HTML = (
""
'"
""
)
# Firewall-интерстициал (HTTP 200, не 403).
_FIREWALL_HTML = (
"Доступ ограничен: проблема с IP"
'...
'
)
def _enable_backconnect(monkeypatch: pytest.MonkeyPatch) -> None:
"""scraper_proxy_url задан (через env-поле), changeip-URL — нет → backconnect."""
monkeypatch.setattr(avito_mod.settings, "scraper_proxy_url_env", "http://u:p@mproxy.site:14619")
monkeypatch.setattr(avito_mod.settings, "avito_proxy_url", None)
monkeypatch.setattr(avito_mod.settings, "avito_proxy_rotate_url", None)
@pytest.mark.asyncio
async def test_backconnect_403_403_200_reconnect_succeeds(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""backconnect: 403 → 403 → 200 → reconnect-retry возвращает HTML, _reset_cffi дёрнут 2×."""
_enable_backconnect(monkeypatch)
scraper = AvitoScraper()
r403 = MagicMock()
r403.status_code = 403
r403.text = ""
ok = MagicMock()
ok.status_code = 200
ok.text = _CARD_HTML
mock_session = AsyncMock()
mock_session.get = AsyncMock(side_effect=[r403, r403, ok])
scraper._cffi = mock_session
# _reset_cffi не должен реально пересоздавать сессию (иначе .get-side_effect
# потеряется) — патчим на no-op счётчик, проверяем что вызвался.
reset = AsyncMock()
with (
patch("app.services.scrapers.avito.asyncio.sleep", AsyncMock()),
patch.object(AvitoScraper, "_reset_cffi", reset),
patch.object(AvitoScraper, "_rotate_ip", AsyncMock(return_value=True)) as rot,
):
html = await scraper._fetch_serp_html("https://www.avito.ru/x", page=1)
assert html == _CARD_HTML
assert reset.await_count == 2 # 403, 403 → два reconnect перед 200
rot.assert_not_awaited() # backconnect не дёргает changeip-ротацию
assert mock_session.get.await_count == 3
@pytest.mark.asyncio
async def test_backconnect_firewall_reconnect_succeeds(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""backconnect: firewall-200 → 200 → reconnect-retry (firewall тоже триггерит)."""
_enable_backconnect(monkeypatch)
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
reset = AsyncMock()
with (
patch("app.services.scrapers.avito.asyncio.sleep", AsyncMock()),
patch.object(AvitoScraper, "_reset_cffi", reset),
):
html = await scraper._fetch_serp_html("https://www.avito.ru/x", page=1)
assert html == _CARD_HTML
assert reset.await_count == 1
@pytest.mark.asyncio
async def test_backconnect_403_exhausted_raises_blocked(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""backconnect: 403 _AVITO_403_MAX_RETRIES+1 раз → всё равно AvitoBlockedError."""
_enable_backconnect(monkeypatch)
scraper = AvitoScraper()
r403 = MagicMock()
r403.status_code = 403
r403.text = ""
mock_session = AsyncMock()
mock_session.get = AsyncMock(return_value=r403)
scraper._cffi = mock_session
reset = AsyncMock()
with (
patch("app.services.scrapers.avito.asyncio.sleep", AsyncMock()),
patch.object(AvitoScraper, "_reset_cffi", reset),
):
with pytest.raises(AvitoBlockedError, match="IP banned"):
await scraper._fetch_serp_html("https://www.avito.ru/x", page=1)
# ровно _AVITO_403_MAX_RETRIES reconnect, потом raise
assert reset.await_count == avito_mod._AVITO_403_MAX_RETRIES
# get: 1 первичный + _AVITO_403_MAX_RETRIES повторов
assert mock_session.get.await_count == avito_mod._AVITO_403_MAX_RETRIES + 1
@pytest.mark.asyncio
async def test_non_backconnect_with_rotate_url_no_reconnect_path(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""rotate_url задан → backconnect=False → одиночный 403 (max_rotations=0) raise сразу.
Новая reconnect-ветка должна быть выключена когда есть changeip-URL: legacy
single-IP mobile-proxy путь без изменений.
"""
monkeypatch.setattr(
avito_mod.settings, "avito_proxy_rotate_url", "https://changeip.example/?proxy_key=x"
)
monkeypatch.setattr(avito_mod.settings, "avito_proxy_max_rotations", 0)
monkeypatch.setattr(avito_mod.settings, "scraper_proxy_url_env", "http://u:p@mproxy.site:14619")
scraper = AvitoScraper()
r403 = MagicMock()
r403.status_code = 403
r403.text = ""
mock_session = AsyncMock()
mock_session.get = AsyncMock(return_value=r403)
scraper._cffi = mock_session
reset = AsyncMock()
with (
patch("app.services.scrapers.avito.asyncio.sleep", AsyncMock()),
patch.object(AvitoScraper, "_reset_cffi", reset),
patch.object(AvitoScraper, "_rotate_ip", AsyncMock(return_value=True)),
):
with pytest.raises(AvitoBlockedError):
await scraper._fetch_serp_html("https://www.avito.ru/x", page=1)
reset.assert_not_awaited() # reconnect-ветка выключена при наличии rotate_url
assert mock_session.get.await_count == 1 # немедленный raise