From 308eeac21428a0bafcac0a69438aa2168985e342 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Fri, 3 Jul 2026 21:48:15 +0000 Subject: [PATCH] fix(tradein/scrapers): wire proxy_pool into Avito legacy pipeline, rotate on ban (#2202) (#2328) --- .../backend/app/services/scrape_pipeline.py | 265 +++++++++++- .../backend/tests/test_ban_rotation_1790.py | 4 + .../tests/test_scrape_pipeline_proxy_pool.py | 408 ++++++++++++++++++ 3 files changed, 655 insertions(+), 22 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_scrape_pipeline_proxy_pool.py diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 434d26ac..e3312b69 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -42,7 +42,7 @@ from sqlalchemy.orm import Session from app.core.config import settings from app.core.shutdown import shutdown_requested -from app.services import scrape_runs +from app.services import proxy_pool, scrape_runs from app.services.scrapers.avito import AvitoScraper from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError @@ -250,6 +250,188 @@ def _avito_proxies() -> dict[str, str] | None: return {"http": url, "https": url} if url else None +def _mask_proxy_url(url: str) -> str: + """Скрыть пароль в proxy-url для логов (scheme://user:***@host).""" + if "@" not in url or "//" not in url: + return url + scheme, rest = url.split("//", 1) + creds, host = rest.split("@", 1) + if ":" in creds: + user, _pwd = creds.split(":", 1) + creds = f"{user}:***" + return f"{scheme}//{creds}@{host}" + + +def _pool_proxy_dict( + db: Session, provider: str, *, run_id: int | None = None +) -> tuple[dict[str, str] | None, proxy_pool.ProxyLease | None]: + """#2202: health-filtered lease из scrape_proxies вместо статического env-прокси. + + За флагом settings.use_proxy_pool_curl (ship-dark, default False — прод не + меняется, пока флаг не включён явно). Флаг off ИЛИ пул пуст/ошибка acquire → + fallback на статический _avito_proxies() (settings.scraper_proxy_url), поведение + как раньше — сбор никогда не падает из-за проблем пула. + + run_id — id текущего scrape_runs-прогона (если он есть у вызывающего, см. + run_avito_city_sweep/run_avito_newbuilding_sweep). Прокидывается в + proxy_pool.acquire(), чтобы scrape_proxies.leased_by нёс реальный run_id для + диагностики банов вместо NON_RUN_LEASE_MARKER. None (default) — прежнее + поведение для вызывающих без run_id в scope (run_avito_pipeline standalone). + + Returns (proxies-dict для curl_cffi AsyncSession, lease | None). lease нужен + вызывающему для release+mark_health в конце run (см. _release_pool_lease) и для + последующей ротации на бан (см. _rotate_pool_proxy). + """ + # getattr-defensive: некоторые тестовые/adapter settings-объекты (SimpleNamespace) + # не несут этого поля — трактуем отсутствие как off (то же ship-dark поведение). + if not getattr(settings, "use_proxy_pool_curl", False): + return _avito_proxies(), None + try: + lease = proxy_pool.acquire(db, provider, run_id=run_id) + except Exception: + logger.warning( + "pipeline: proxy_pool.acquire(%s) failed — fallback to env proxy", + provider, + exc_info=True, + ) + lease = None + if lease is None: + logger.warning( + "pipeline: proxy pool empty/unhealthy for provider=%s — fallback to env proxy", + provider, + ) + return _avito_proxies(), None + logger.info( + "pipeline: using pool proxy id=%d provider=%s url=%s for this run", + lease.id, + provider, + _mask_proxy_url(lease.url), + ) + return {"http": lease.url, "https": lease.url}, lease + + +def _release_pool_lease(db: Session, lease: proxy_pool.ProxyLease | None, ok: bool) -> None: + """Best-effort release+mark_health в конце run (#2202) — проблема пула не должна + ронять уже собранные данные. lease=None (пул выключен/не выдал прокси) → no-op. + """ + if lease is None: + return + try: + proxy_pool.mark_health(db, lease.id, ok) + except Exception: + logger.warning("pipeline: mark_health failed for lease id=%d", lease.id, exc_info=True) + try: + proxy_pool.release(db, lease.id) + except Exception: + logger.warning("pipeline: release failed for lease id=%d", lease.id, exc_info=True) + + +async def _rotate_pool_proxy( + db: Session, + provider: str, + session: AsyncSession, + current_lease: proxy_pool.ProxyLease | None, +) -> proxy_pool.ProxyLease | None: + """#2202: переключиться на ДРУГОЙ здоровый прокси из пула при бане (не changeip + одного и того же IP). Помечает current_lease нездоровым (mark_health ok=False), + затем acquire()-ит следующий свободный здоровый прокси, и ТОЛЬКО ПОСЛЕ этого + освобождает current_lease (release). Порядок здесь принципиален (#2202 MAJOR-фикс, + review-репро): release ДО acquire() убирает current_lease из leased_by-фильтра + немедленно, и один бан поднимает consecutive_fails только до 1 (порог исключения — + proxy_pool.MAX_CONSECUTIVE_FAILS=3) — только что забаненный прокси тут же снова + проходит health-фильтр и acquire() может вернуть его же обратно (ORDER BY + last_ok_at NULLS LAST, id — никак не гарантирует, что он не окажется первым). + Пока current_lease НЕ освобождён, acquire()'s `leased_by IS NULL` условие + гарантированно его исключает — никакой отдельный exclude_id-параметр не нужен. + + Успешная ротация мутирует session.proxies — curl_cffi AsyncSession читает + self.proxies заново на каждый request (не кэширует на конструкторе), так что + переприсвоение атрибута действует на следующий вызов без пересоздания сессии. + + Returns новый lease, или None если пул пуст/ошибка (вызывающий код в этом случае + падает обратно на changeip-ротацию одного IP — см. _rotate_avito_proxy). + current_lease освобождается в обоих случаях (успех/провал) — он подтверждённо + нездоров, дальше эта функция за него не отвечает. + """ + if current_lease is not None: + try: + proxy_pool.mark_health(db, current_lease.id, ok=False) + except Exception: + logger.warning( + "pipeline: mark_health(False) failed for lease id=%d", + current_lease.id, + exc_info=True, + ) + + try: + new_lease = proxy_pool.acquire(db, provider) + except Exception: + logger.warning( + "pipeline: proxy_pool.acquire(%s) failed during rotation", provider, exc_info=True + ) + new_lease = None + + if current_lease is not None: + try: + proxy_pool.release(db, current_lease.id) + except Exception: + logger.warning( + "pipeline: release failed for lease id=%d", current_lease.id, exc_info=True + ) + + if new_lease is None: + logger.warning( + "pipeline: proxy pool exhausted for provider=%s during rotation — " + "no healthy proxy to rotate to", + provider, + ) + return None + + session.proxies = {"http": new_lease.url, "https": new_lease.url} + logger.info( + "pipeline: ROTATED to pool proxy id=%d provider=%s url=%s (previous lease id=%s)", + new_lease.id, + provider, + _mask_proxy_url(new_lease.url), + current_lease.id if current_lease is not None else None, + ) + return new_lease + + +async def _rotate_avito_proxy( + db: Session, + session: AsyncSession | None, + current_lease: proxy_pool.ProxyLease | None, + *, + reason: str, + rotations_done: int, +) -> tuple[bool, proxy_pool.ProxyLease | None]: + """#2202: единая точка ротации на бан для Avito. + + Сначала пробует переключить прокси ПУЛА (health-фильтр, гарантированно другой + узел) — только когда settings.use_proxy_pool_curl включён И есть curl-сессия + (session is not None, т.е. не browser-mode). Если пул выключен, пуст, или это + browser-mode (session=None) — падает на старый changeip (смена exit-IP ОДНОГО + прокси-аккаунта, как раньше). + + NB: пока делит общий rotations_done/avito_proxy_max_rotations бюджет с + changeip-путём — раздельные бюджеты (пул своё, changeip своё) можно завести + отдельным follow-up, если пул-ротации на практике окажутся частым событием. + + Returns (rotated, lease для последующих вызовов — обновлён при успешной + пул-ротации, иначе тот же current_lease). + """ + if getattr(settings, "use_proxy_pool_curl", False) and session is not None: + new_lease = await _rotate_pool_proxy(db, "avito", session, current_lease) + if new_lease is not None: + return True, new_lease + logger.warning( + "pipeline: pool rotation unavailable (reason=%s) — falling back to changeip", reason + ) + rotated = await _rotate_proxy_ip(reason=reason, rotations_done=rotations_done, source="avito") + return rotated, current_lease + + # ── Result dataclasses ────────────────────────────────────────── @@ -332,6 +514,9 @@ async def run_avito_pipeline( browser_fetcher: BrowserFetcher | None = None own_session = False own_browser = False + # #2202: pool-lease текущей curl-сессии (None если browser-mode/pool off/shared + # session переданный извне — тогда владелец проксей не мы, не трогаем). + _lease: proxy_pool.ProxyLease | None = None scraper = AvitoScraper() if browser_mode: @@ -343,12 +528,16 @@ async def run_avito_pipeline( scraper._browser = browser_fetcher else: own_session = shared_session is None - session = shared_session or AsyncSession( - impersonate="chrome120", - timeout=25, - headers=_CHROME_HEADERS, - proxies=_avito_proxies(), - ) + if shared_session is not None: + session = shared_session + else: + _proxies, _lease = _pool_proxy_dict(db, "avito") + session = AsyncSession( + impersonate="chrome120", + timeout=25, + headers=_CHROME_HEADERS, + proxies=_proxies, + ) scraper._cffi = session try: @@ -429,7 +618,10 @@ async def run_avito_pipeline( # После ротации сбрасываем счётчик и продолжаем — НЕ abort. # Если ротации исчерпаны — abort enrichment (listings уже сохранены). if enrich_rotations_done < settings.avito_proxy_max_rotations: - rotated = await _rotate_proxy_ip( + rotated, _lease = await _rotate_avito_proxy( + db, + session, + _lease, reason=f"house consecutive={enrich_consecutive_blocks}", rotations_done=enrich_rotations_done, ) @@ -437,7 +629,7 @@ async def run_avito_pipeline( if rotated: enrich_consecutive_blocks = 0 logger.info( - "pipeline:enrich ROTATE — IP changed, " + "pipeline:enrich ROTATE — proxy changed, " "reset consecutive blocks (rotation #%d/%d)", enrich_rotations_done, settings.avito_proxy_max_rotations, @@ -545,7 +737,10 @@ async def run_avito_pipeline( # После ротации сбрасываем счётчик — продолжаем detail-обход. # Если ротации исчерпаны — abort detail-фазы (listings уже сохранены). if enrich_rotations_done < settings.avito_proxy_max_rotations: - rotated = await _rotate_proxy_ip( + rotated, _lease = await _rotate_avito_proxy( + db, + session, + _lease, reason=f"detail consecutive={enrich_consecutive_blocks}", rotations_done=enrich_rotations_done, ) @@ -553,7 +748,7 @@ async def run_avito_pipeline( if rotated: enrich_consecutive_blocks = 0 logger.info( - "pipeline:enrich ROTATE — IP changed, " + "pipeline:enrich ROTATE — proxy changed, " "reset consecutive blocks (rotation #%d/%d)", enrich_rotations_done, settings.avito_proxy_max_rotations, @@ -625,7 +820,10 @@ async def run_avito_pipeline( if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT: # #1790: попытка ротации перед abort'ом (общий бюджет run). if enrich_rotations_done < settings.avito_proxy_max_rotations: - rotated = await _rotate_proxy_ip( + rotated, _lease = await _rotate_avito_proxy( + db, + session, + _lease, reason=f"house-detail consecutive={enrich_consecutive_blocks}", rotations_done=enrich_rotations_done, ) @@ -633,7 +831,7 @@ async def run_avito_pipeline( if rotated: enrich_consecutive_blocks = 0 logger.info( - "pipeline:enrich ROTATE — IP changed, " + "pipeline:enrich ROTATE — proxy changed, " "reset consecutive blocks (rotation #%d/%d)", enrich_rotations_done, settings.avito_proxy_max_rotations, @@ -696,6 +894,8 @@ async def run_avito_pipeline( await session.close() if own_browser and browser_fetcher is not None: await browser_fetcher.__aexit__(None, None, None) + if own_session: + _release_pool_lease(db, _lease, ok=not enrichment_aborted) @dataclass @@ -776,6 +976,8 @@ async def run_avito_city_sweep( browser_mode = settings.scraper_fetch_mode == "browser" # #1790: общий бюджет IP-ротаций на весь sweep (разделяется по всем anchor'ам). sweep_rotations_done = 0 + # #2202: pool-lease текущей curl-сессии (None в browser-mode/pool off). + _lease: proxy_pool.ProxyLease | None = None async with AsyncExitStack() as stack: session: AsyncSession | None = None @@ -783,12 +985,13 @@ async def run_avito_city_sweep( if browser_mode: shared_bf = await stack.enter_async_context(BrowserFetcher(source="avito")) else: + _proxies, _lease = _pool_proxy_dict(db, "avito", run_id=run_id) session = await stack.enter_async_context( AsyncSession( impersonate="chrome120", timeout=25, headers=_CHROME_HEADERS, - proxies=_avito_proxies(), + proxies=_proxies, ) ) try: @@ -844,7 +1047,7 @@ async def run_avito_city_sweep( НЕМЕДЛЕННО после SERP+save (до detail-фазы). Это гарантирует, что TimeoutError из wait_for теряет только detail-счётчики, но не SERP. """ - nonlocal all_touched_house_ids, sweep_rotations_done + nonlocal all_touched_house_ids, sweep_rotations_done, _lease # ── Phase 1+2: SERP + save ───────────────────────────────── scraper = AvitoScraper() @@ -954,7 +1157,10 @@ async def run_avito_city_sweep( >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT and sweep_rotations_done < settings.avito_proxy_max_rotations ): - rotated = await _rotate_proxy_ip( + rotated, _lease = await _rotate_avito_proxy( + db, + session, + _lease, reason=( f"city-sweep house " f"consecutive={sweep_consecutive_house_blocks}" @@ -966,7 +1172,7 @@ async def run_avito_city_sweep( sweep_consecutive_house_blocks = 0 logger.info( "city-sweep run_id=%d: house ROTATE — " - "IP changed, reset consecutive " + "proxy changed, reset consecutive " "(rotation #%d/%d)", run_id, sweep_rotations_done, @@ -1077,7 +1283,10 @@ async def run_avito_city_sweep( if consecutive_blocks >= 3: # #1790: попытка ротации перед abort'ом. if sweep_rotations_done < settings.avito_proxy_max_rotations: - rotated = await _rotate_proxy_ip( + rotated, _lease = await _rotate_avito_proxy( + db, + session, + _lease, reason=( f"city-sweep detail " f"consecutive={consecutive_blocks}" @@ -1089,7 +1298,7 @@ async def run_avito_city_sweep( consecutive_blocks = 0 logger.info( "city-sweep run_id=%d: detail ROTATE — " - "IP changed, reset consecutive " + "proxy changed, reset consecutive " "(rotation #%d/%d)", run_id, sweep_rotations_done, @@ -1177,7 +1386,10 @@ async def run_avito_city_sweep( and sweep_rotations_done < settings.avito_proxy_max_rotations ): - rotated = await _rotate_proxy_ip( + rotated, _lease = await _rotate_avito_proxy( + db, + session, + _lease, reason=( "city-sweep house(detail) " f"consecutive=" @@ -1190,7 +1402,7 @@ async def run_avito_city_sweep( sweep_consecutive_detail_house_blocks = 0 logger.info( "city-sweep run_id=%d: house(detail) " - "ROTATE — IP changed (rotation #%d/%d)", + "ROTATE — proxy changed (rotation #%d/%d)", run_id, sweep_rotations_done, settings.avito_proxy_max_rotations, @@ -1390,6 +1602,10 @@ async def run_avito_city_sweep( logger.exception("city-sweep run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise + finally: + # #2202: best-effort release+mark_health — не должен ронять уже + # собранный sweep. ok=False только если весь run завершился с ошибками. + _release_pool_lease(db, _lease, ok=(counters.errors_count == 0)) # ── Avito newbuilding (novostroyka) citywide sweep ──────────────── @@ -1430,18 +1646,21 @@ async def run_avito_newbuilding_sweep( counters = NewbuildingSweepCounters() browser_mode = settings.scraper_fetch_mode == "browser" + # #2202: pool-lease текущей curl-сессии (None в browser-mode/pool off). + _lease: proxy_pool.ProxyLease | None = None async with AsyncExitStack() as stack: session: AsyncSession | None = None shared_bf: BrowserFetcher | None = None if browser_mode: shared_bf = await stack.enter_async_context(BrowserFetcher(source="avito")) else: + _proxies, _lease = _pool_proxy_dict(db, "avito", run_id=run_id) session = await stack.enter_async_context( AsyncSession( impersonate="chrome120", timeout=25, headers=_CHROME_HEADERS, - proxies=_avito_proxies(), + proxies=_proxies, ) ) try: @@ -1508,6 +1727,8 @@ async def run_avito_newbuilding_sweep( logger.exception("nb-sweep run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise + finally: + _release_pool_lease(db, _lease, ok=(counters.errors_count == 0)) # ── Yandex city sweep ─────────────────────────────────────────── diff --git a/tradein-mvp/backend/tests/test_ban_rotation_1790.py b/tradein-mvp/backend/tests/test_ban_rotation_1790.py index b434ac68..b4c35122 100644 --- a/tradein-mvp/backend/tests/test_ban_rotation_1790.py +++ b/tradein-mvp/backend/tests/test_ban_rotation_1790.py @@ -151,6 +151,10 @@ async def test_rotation_bounded_abort_when_budget_exhausted() -> None: mock_settings.scraper_fetch_mode = "curl_cffi" mock_settings.scraper_proxy_url = None mock_settings.scraper_skip_seen_today = False + # #2202: bare MagicMock settings auto-truthy on any attribute access — pin + # explicitly off so _rotate_avito_proxy takes the changeip path this test + # asserts on (pool path is covered separately in test_scrape_pipeline_proxy_pool.py). + mock_settings.use_proxy_pool_curl = False result = await run_avito_pipeline( MagicMock(), 56.8, 60.6, 1000, enrich_houses=True, enrich_detail_top_n=0 diff --git a/tradein-mvp/backend/tests/test_scrape_pipeline_proxy_pool.py b/tradein-mvp/backend/tests/test_scrape_pipeline_proxy_pool.py new file mode 100644 index 00000000..5497ae2b --- /dev/null +++ b/tradein-mvp/backend/tests/test_scrape_pipeline_proxy_pool.py @@ -0,0 +1,408 @@ +"""Tests for scrape_pipeline.py proxy-pool wiring (#2202). + +Issue #2202: боевые скрейперы (scrape_pipeline.py Avito orchestrator) читали только +статический settings.scraper_proxy_url — один mobile-IP на все прогоны, SPOF. Этот +файл покрывает новые pool-aware helpers: + + - _pool_proxy_dict: acquire() health-filtered lease за флагом USE_PROXY_POOL_CURL, + fallback на статический env при флаге off / пустом пуле / ошибке. + - _rotate_pool_proxy: переключение на ДРУГОЙ здоровый прокси из пула при бане + (mark_health(False)+release текущего, acquire() следующего, мутация session.proxies). + - _rotate_avito_proxy: единая точка ротации — пул first, changeip fallback. + +Acceptance-тест (issue AC4): пул с 1 «забаненным» прокси — ротация выбирает +СЛЕДУЮЩИЙ здоровый, не зависший на забаненном и не переиспользующий только что +провалившийся lease раньше более свежего здорового кандидата. + +Offline: FakeSession эмулирует scrape_proxies-таблицу (тот же паттерн, что +tests/services/test_proxy_pool.py), без live БД/сети. +""" + +from __future__ import annotations + +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from datetime import UTC, datetime +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, patch + +import pytest + +from app.services import proxy_pool +from app.services.proxy_pool import MAX_CONSECUTIVE_FAILS +from app.services.scrape_pipeline import ( + _pool_proxy_dict, + _release_pool_lease, + _rotate_avito_proxy, + _rotate_pool_proxy, +) + +# ── stateful fake session (mirrors tests/services/test_proxy_pool.py) ───────── + + +class _FakeResult: + def __init__(self, rows: list[dict[str, Any]]): + self._rows = rows + + def mappings(self) -> _FakeResult: + return self + + def fetchone(self) -> dict[str, Any] | None: + return self._rows[0] if self._rows else None + + def all(self) -> list[dict[str, Any]]: + return list(self._rows) + + +class FakeSession: + """Эмуляция Session поверх in-memory списка scrape_proxies-строк.""" + + def __init__(self, rows: list[dict[str, Any]]): + self.rows = rows + + def _by_id(self, pid: int) -> dict[str, Any] | None: + return next((r for r in self.rows if r["id"] == pid), None) + + def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult: + sql = str(stmt) + p = params or {} + + if "FOR UPDATE SKIP LOCKED" in sql: # acquire SELECT + provider = p["provider"] + max_fails = p["max_fails"] + cands = [ + r + for r in self.rows + if r["enabled"] + and r["consecutive_fails"] < max_fails + and r["provider_affinity"] in (provider, "any") + and r["leased_by"] is None + ] + cands.sort( + key=lambda r: ( + r["last_ok_at"] is None, + r["last_ok_at"] or datetime.min.replace(tzinfo=UTC), + r["id"], + ) + ) + return _FakeResult(cands[:1]) + + if "SET leased_by = CAST(:run_id" in sql: # acquire lease UPDATE + row = self._by_id(p["id"]) + if row is not None: + row["leased_by"] = p["run_id"] + row["leased_at"] = datetime.now(UTC) + return _FakeResult([]) + + if "SET leased_by = NULL" in sql: # release + row = self._by_id(p["id"]) + if row is not None: + row["leased_by"] = None + row["leased_at"] = None + return _FakeResult([]) + + if "SET consecutive_fails = 0" in sql: # mark_health ok + row = self._by_id(p["id"]) + if row is not None: + row["consecutive_fails"] = 0 + row["exit_ip"] = p["exit_ip"] + row["latency_ms"] = p["latency_ms"] + row["last_ok_at"] = datetime.now(UTC) + return _FakeResult([]) + + if "consecutive_fails = consecutive_fails + 1" in sql: # mark_health fail + row = self._by_id(p["id"]) + if row is not None: + row["consecutive_fails"] += 1 + if row["consecutive_fails"] >= p["disable_threshold"]: + row["enabled"] = False + return _FakeResult([]) + + raise AssertionError(f"unhandled SQL: {sql}") + + def commit(self) -> None: + pass + + def rollback(self) -> None: + pass + + +def _proxy( + pid: int, + *, + affinity: str = "avito", + enabled: bool = True, + fails: int = 0, + leased_by: int | None = None, + last_ok_at: datetime | None = None, +) -> dict[str, Any]: + return { + "id": pid, + "url": f"http://u:p@h{pid}:8080", + "kind": "http", + "rotate_url": None, + "provider_affinity": affinity, + "enabled": enabled, + "consecutive_fails": fails, + "leased_by": leased_by, + "leased_at": None, + "last_ok_at": last_ok_at, + "exit_ip": None, + "latency_ms": None, + } + + +def _mock_settings(*, use_proxy_pool_curl: bool, scraper_proxy_url: str | None = None): + return SimpleNamespace( + use_proxy_pool_curl=use_proxy_pool_curl, + scraper_proxy_url=scraper_proxy_url, + ) + + +# ── _pool_proxy_dict ──────────────────────────────────────────────────────── + + +def test_pool_proxy_dict_falls_back_when_flag_off(): + db = FakeSession([_proxy(1)]) + with patch( + "app.services.scrape_pipeline.settings", + _mock_settings(use_proxy_pool_curl=False, scraper_proxy_url="http://env:3128"), + ): + proxies, lease = _pool_proxy_dict(db, "avito") # type: ignore[arg-type] + + assert proxies == {"http": "http://env:3128", "https": "http://env:3128"} + assert lease is None + + +def test_pool_proxy_dict_returns_lease_when_flag_on_and_healthy(): + db = FakeSession([_proxy(1)]) + with patch( + "app.services.scrape_pipeline.settings", + _mock_settings(use_proxy_pool_curl=True), + ): + proxies, lease = _pool_proxy_dict(db, "avito") # type: ignore[arg-type] + + assert lease is not None and lease.id == 1 + assert proxies == {"http": "http://u:p@h1:8080", "https": "http://u:p@h1:8080"} + assert db._by_id(1)["leased_by"] is not None # lease проставлен + + +def test_pool_proxy_dict_falls_back_when_pool_empty(): + db = FakeSession([]) # пустой пул + with patch( + "app.services.scrape_pipeline.settings", + _mock_settings(use_proxy_pool_curl=True, scraper_proxy_url="http://env:3128"), + ): + proxies, lease = _pool_proxy_dict(db, "avito") # type: ignore[arg-type] + + assert lease is None + assert proxies == {"http": "http://env:3128", "https": "http://env:3128"} + + +# ── _rotate_pool_proxy — acceptance-тест AC4 ─────────────────────────────── + + +@pytest.mark.asyncio +async def test_rotate_pool_proxy_skips_banned_picks_next_healthy(): + """#2202 AC4: пул с 1 забаненным прокси — ротация выбирает следующий здоровый. + + BAD (id=1) — уже забанен (fails>=MAX_CONSECUTIVE_FAILS), исключён acquire() на + уровне SQL-фильтра. CURRENT (id=3) — прокси активного прогона, только что словивший + бан (403/429) — вызывающий код зовёт rotate. GOOD (id=2) — свободный здоровый, + единственный валидный кандидат после того как CURRENT будет помечен/освобождён. + + NB (#2202 MAJOR-fix): гарантия исключения CURRENT из acquire() держится на том, + что release(current_lease) вызывается ПОСЛЕ acquire(), не до — пока current_lease + leased_by-занят, SQL `leased_by IS NULL` исключает его вне зависимости от id/ + last_ok_at сортировки. См. зеркальный тест ниже + (test_rotate_pool_proxy_skips_banned_picks_next_healthy_current_id_lower) — + та же гарантия при CURRENT id НИЖЕ GOOD id (иначе тай-брейк по id в ORDER BY + "last_ok_at NULLS LAST, id" мог случайно фейворить эту фикстуру). + """ + db = FakeSession( + [ + _proxy(1, fails=MAX_CONSECUTIVE_FAILS), # BAD — уже забанен + _proxy(2, fails=0), # GOOD — ожидаемый выбор + _proxy(3, fails=0, leased_by=999), # CURRENT — в работе, банится сейчас + ] + ) + current_lease = proxy_pool.ProxyLease( + id=3, url="http://u:p@h3:8080", kind="http", rotate_url=None + ) + session = SimpleNamespace(proxies=None) + + new_lease = await _rotate_pool_proxy(db, "avito", session, current_lease) # type: ignore[arg-type] + + assert new_lease is not None + assert new_lease.id == 2 # GOOD, не BAD (excluded) и не CURRENT (только что провален) + assert session.proxies == {"http": "http://u:p@h2:8080", "https": "http://u:p@h2:8080"} + + # CURRENT (id=3) помечен нездоровым и освобождён + assert db._by_id(3)["consecutive_fails"] == 1 + assert db._by_id(3)["leased_by"] is None + # BAD (id=1) не тронут этой операцией + assert db._by_id(1)["consecutive_fails"] == MAX_CONSECUTIVE_FAILS + # GOOD (id=2) теперь зализан под новую ротацию + assert db._by_id(2)["leased_by"] is not None + + +@pytest.mark.asyncio +async def test_rotate_pool_proxy_skips_banned_picks_next_healthy_current_id_lower(): + """#2202 AC4 regression guard (MAJOR-fix): гарантия исключения current_lease НЕ + должна зависеть от того, чей id меньше в фикстуре. + + Зеркало test_rotate_pool_proxy_skips_banned_picks_next_healthy с CURRENT/GOOD id + в обратном порядке: CURRENT=1 < GOOD=2 (оригинальный тест имел CURRENT=3 > GOOD=2, + поэтому "тай-брейк по id" в ORDER BY last_ok_at NULLS LAST, id случайно совпадал с + правильным ответом и не доказывал реальную гарантию — см. review-репро на #2202). + + До фикса (release(current_lease) ДО acquire()): CURRENT освобождается, попадает в + leased_by IS NULL кандидаты с fails=1 (< MAX_CONSECUTIVE_FAILS=3, т.е. "здоров"), + и при равном last_ok_at (оба NULL) выигрывает тай-брейк по id (1 < 2) — acquire() + возвращает ТОТ ЖЕ прокси, который только что забанили. Этот тест ловит именно этот + регресс: если release() снова передвинуть перед acquire(), new_lease.id будет 1 + (CURRENT), а не 2 (GOOD), и assert упадёт. + """ + db = FakeSession( + [ + _proxy(1, fails=0, leased_by=999), # CURRENT — id НИЖЕ GOOD, банится сейчас + _proxy(2, fails=0), # GOOD — единственный здоровый свободный кандидат + ] + ) + current_lease = proxy_pool.ProxyLease( + id=1, url="http://u:p@h1:8080", kind="http", rotate_url=None + ) + session = SimpleNamespace(proxies=None) + + new_lease = await _rotate_pool_proxy(db, "avito", session, current_lease) # type: ignore[arg-type] + + assert new_lease is not None + assert new_lease.id == 2 # GOOD — НЕ CURRENT, несмотря на id=1 < id=2 + assert session.proxies == {"http": "http://u:p@h2:8080", "https": "http://u:p@h2:8080"} + + # CURRENT (id=1) помечен нездоровым и освобождён (после успешной ротации) + assert db._by_id(1)["consecutive_fails"] == 1 + assert db._by_id(1)["leased_by"] is None + # GOOD (id=2) теперь зализан под новую ротацию + assert db._by_id(2)["leased_by"] is not None + + +@pytest.mark.asyncio +async def test_rotate_pool_proxy_exhausted_returns_none(): + """Все кандидаты нездоровы после ротации → None (caller падает на changeip). + + BAD (id=1) уже забанен. CURRENT (id=3) на MAX_CONSECUTIVE_FAILS-1 fails — ЭТА + ротация (mark_health ok=False) толкает его ровно до порога → тоже исключается + из acquire(). Ни одного здорового кандидата не остаётся. + """ + db = FakeSession( + [ + _proxy(1, fails=MAX_CONSECUTIVE_FAILS), # BAD — уже забанен + _proxy(3, fails=MAX_CONSECUTIVE_FAILS - 1, leased_by=999), # CURRENT — на грани + ] + ) + current_lease = proxy_pool.ProxyLease( + id=3, url="http://u:p@h3:8080", kind="http", rotate_url=None + ) + session = SimpleNamespace(proxies=None) + + new_lease = await _rotate_pool_proxy(db, "avito", session, current_lease) # type: ignore[arg-type] + + assert new_lease is None + assert session.proxies is None # не тронут — caller остаётся на старом/changeip + assert db._by_id(3)["leased_by"] is None # текущий всё равно освобождён + + +# ── _rotate_avito_proxy — pool-first, changeip fallback ──────────────────── + + +@pytest.mark.asyncio +async def test_rotate_avito_proxy_uses_pool_when_flag_on(): + db = FakeSession( + [ + _proxy(1, fails=0), # свободный здоровый — цель ротации + _proxy(3, fails=0, leased_by=999), # CURRENT + ] + ) + current_lease = proxy_pool.ProxyLease( + id=3, url="http://u:p@h3:8080", kind="http", rotate_url=None + ) + session = SimpleNamespace(proxies=None) + + with patch("app.services.scrape_pipeline.settings", _mock_settings(use_proxy_pool_curl=True)): + with patch("app.services.scrape_pipeline._rotate_proxy_ip", AsyncMock()) as changeip_mock: + rotated, new_lease = await _rotate_avito_proxy( + db, # type: ignore[arg-type] + session, # type: ignore[arg-type] + current_lease, + reason="test", + rotations_done=0, + ) + + assert rotated is True + assert new_lease is not None and new_lease.id == 1 + changeip_mock.assert_not_awaited() # пул сработал — changeip не нужен + + +@pytest.mark.asyncio +async def test_rotate_avito_proxy_falls_back_to_changeip_when_flag_off(): + session = SimpleNamespace(proxies=None) + + with patch("app.services.scrape_pipeline.settings", _mock_settings(use_proxy_pool_curl=False)): + with patch( + "app.services.scrape_pipeline._rotate_proxy_ip", AsyncMock(return_value=True) + ) as changeip_mock: + rotated, new_lease = await _rotate_avito_proxy( + None, # type: ignore[arg-type] + session, # type: ignore[arg-type] + None, + reason="test", + rotations_done=0, + ) + + assert rotated is True + assert new_lease is None + changeip_mock.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_rotate_avito_proxy_falls_back_to_changeip_when_pool_exhausted(): + """Пул on, но acquire() пуст (например все забанены) — падаем на changeip, не молчим.""" + db = FakeSession([_proxy(1, fails=MAX_CONSECUTIVE_FAILS)]) # единственный — забанен + session = SimpleNamespace(proxies=None) + + with patch("app.services.scrape_pipeline.settings", _mock_settings(use_proxy_pool_curl=True)): + with patch( + "app.services.scrape_pipeline._rotate_proxy_ip", AsyncMock(return_value=True) + ) as changeip_mock: + rotated, new_lease = await _rotate_avito_proxy( + db, # type: ignore[arg-type] + session, # type: ignore[arg-type] + None, + reason="test", + rotations_done=0, + ) + + assert rotated is True + assert new_lease is None + changeip_mock.assert_awaited_once() # pool exhausted → fallback path taken + + +# ── _release_pool_lease ────────────────────────────────────────────────────── + + +def test_release_pool_lease_noop_when_none(): + _release_pool_lease(None, None, ok=True) # type: ignore[arg-type] # must not raise + + +def test_release_pool_lease_marks_health_and_releases(): + db = FakeSession([_proxy(1, leased_by=100)]) + lease = proxy_pool.ProxyLease(id=1, url="http://u:p@h1:8080", kind="http", rotate_url=None) + + _release_pool_lease(db, lease, ok=True) # type: ignore[arg-type] + + assert db._by_id(1)["leased_by"] is None + assert db._by_id(1)["consecutive_fails"] == 0