gendesign/tradein-mvp/backend/tests/test_2700_cian_detail_403_node.py

"""#2700: 403 Циана перестаёт умирать внутри провайдера — узел снимается, прогон честен.

Живая различающая проба на проде 2026-08-10 (один и тот же detail-URL, один и тот же
код, менялся ТОЛЬКО прокси-узел):

    узел 1  (asocks-residential-1, 46.8.110.92)  → HTTP 403,  21 564 б, `cian_waf_block`
    узел 9  (asocks-mobile-1,       5.227.16.0)  → HTTP 200, 617 352 б, state ok
    узел 10 (asocks-mobile-2,     95.104.183.29) → HTTP 200, 617 355 б, state ok
    узел 11 (asocks-mobile-3,      95.55.49.98)  → HTTP 200, 617 407 б, state ok

То есть отбита была ПАРА «узел × Циан», а не площадка (и не наши заголовки: те же 200
пришли без единой куки — detail-страница авторизации не требует). Пятнадцать суток
подряд это выглядело как «Циан нас забанил» ровно потому, что 403 гасился в
`return None`: пул получал `mark_health(ok=True)` на отбитый узел и продолжал выдавать
его Циану, а прогон писал `detail_failed=50` при `errors_count=0` и статусе `done`.

Тесты красные на старом коде:
  * `fetch_detail` возвращал None и НЕ звал `mark_banned` → узел оставался в выдаче;
  * `mark_done` писал `status='done'` прогону, у которого отказали все 50 попыток.
"""

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from scraper_kit.cian_exceptions import CianBlockedError
from scraper_kit.contracts import ProxyLease
from scraper_kit.orchestration import runs as kit_runs
from scraper_kit.providers.cian import detail as cian_detail
from scraper_kit.proxy_errors import ProxyBanError

from app.services import scrape_runs as app_runs

_MODULES = {"kit": kit_runs, "app": app_runs}
_LEASE = ProxyLease(id=1, url="http://user:pass@node-1:10423", kind="http", rotate_url=None)


@dataclass
class _FakeConfig:
    use_proxy_pool_curl: bool = True
    cian_proxy_url: str | None = None
    environment: str = "production"


class _SpyProvider:
    """ProxyProvider-заглушка (тот же контракт, что в test_proxy_pool_curl_paths)."""

    def __init__(self) -> None:
        self.mark_health_calls: list[tuple[int, bool]] = []
        self.mark_banned_calls: list[tuple[int, str]] = []
        self.release_calls: list[int] = []

    def acquire(self, provider: str) -> ProxyLease:
        return _LEASE

    def release(self, lease: ProxyLease) -> None:
        self.release_calls.append(lease.id)

    def mark_health(
        self, lease: ProxyLease, ok: bool, *, exit_ip: Any = None, latency_ms: Any = None
    ) -> None:
        self.mark_health_calls.append((lease.id, ok))

    def mark_banned(self, lease: ProxyLease, *, source: str) -> None:
        self.mark_banned_calls.append((lease.id, source))


def _session_returning(status_code: int, text: str = "") -> MagicMock:
    session = MagicMock()
    session.get = AsyncMock(return_value=MagicMock(status_code=status_code, text=text))
    session.close = AsyncMock()
    return session


async def _fetch(status_code: int, spy: _SpyProvider) -> Any:
    with patch.object(
        cian_detail, "build_curl_cffi_session", return_value=_session_returning(status_code)
    ):
        return await cian_detail.fetch_detail(
            "https://ekb.cian.ru/sale/flat/332775238/",
            config=_FakeConfig(),
            proxy_provider=spy,
        )


# ── 1. 403 доходит до пула ────────────────────────────────────────────────────


async def test_403_bans_the_node_for_cian_only() -> None:
    """Красный на старом коде: было `return None`, бана узла не происходило.

    Проверяется ПОВЕДЕНИЕ пула (`mark_banned` на паре «узел × cian»), а не наличие
    нового имени в коде.
    """
    spy = _SpyProvider()
    with pytest.raises(CianBlockedError):
        await _fetch(403, spy)
    assert spy.mark_banned_calls == [(1, "cian")]
    assert spy.mark_health_calls == [(1, False)]
    assert spy.release_calls == [1]  # lease не течёт даже на бане


def test_blocked_error_is_recognised_by_generic_proxy_layer() -> None:
    """Generic curl-слой узнаёт бан по `ProxyBanError`, не зная про Циан."""
    assert issubclass(CianBlockedError, ProxyBanError)


# ── 2. Не-бан остаётся не-баном ───────────────────────────────────────────────


async def test_404_does_not_ban_the_node() -> None:
    """Снятое объявление — не бан: наказывать за него здоровый узел нельзя."""
    spy = _SpyProvider()
    assert await _fetch(404, spy) is None
    assert spy.mark_banned_calls == []
    assert spy.mark_health_calls == [(1, True)]


# ── 3. Прогон с полностью отказавшей фазой перестаёт быть 'done' ──────────────


def _capture_status(mod: Any, counters: dict[str, int]) -> list[str]:
    """Статусы всех UPDATE'ов, которые сделал mark_done на фейковой сессии.

    Читаем СТАТУС В SQL (как в test_2625_run_that_did_nothing), а не имя вызванной
    функции: тест обязан краснеть на поведении финализатора.
    """
    statuses: list[str] = []

    def _execute(stmt: Any, *args: Any, **kwargs: Any) -> MagicMock:
        sql = str(stmt)
        for status in ("done", "failed", "banned"):
            if f"status = '{status}'" in sql:
                statuses.append(status)
        return MagicMock()

    db = MagicMock()
    db.execute.side_effect = _execute
    with patch.object(mod, "sentry_sdk", MagicMock()):
        mod.mark_done(db, 3258, dict(counters))
    return statuses


# Реальные counters с прода, не выдуманные.
# Прогон 3258 (cian_city_sweep, 06.08): 50 из 50 detail'ов отказали, статус 'done'.
PROD_3258_ALL_DETAIL_FAILED = {
    "anchors_done": 5,
    "anchors_total": 5,
    "errors_count": 0,
    "lots_fetched": 1680,
    "lots_inserted": 59,
    "lots_updated": 626,
    "detail_attempted": 50,
    "detail_failed": 50,
    "detail_enriched": 0,
    "houses_attempted": 40,
    "houses_failed": 40,
    "houses_enriched": 0,
}
# Прогон 3597 (cian_city_sweep, 10.08, уже после того как 403 ушёл): 10 из 11 удачны.
PROD_3597_MOSTLY_OK = {
    **PROD_3258_ALL_DETAIL_FAILED,
    "errors_count": 1,
    "detail_attempted": 11,
    "detail_failed": 1,
    "detail_enriched": 10,
    "houses_attempted": 0,
    "houses_failed": 0,
}
# Одиночная попытка, и та неудачна — шум, а не диагноз: прогон остаётся 'done'.
SINGLE_ATTEMPT_FAILED = {
    **PROD_3597_MOSTLY_OK,
    "detail_attempted": 1,
    "detail_failed": 1,
    "detail_enriched": 0,
}


@pytest.mark.parametrize("name", list(_MODULES))
def test_phase_failed_100_percent_is_not_done(name: str) -> None:
    """Прод-прогон 3258: detail 50/50 отказ → 'failed'. Красный на старом коде."""
    assert _capture_status(_MODULES[name], PROD_3258_ALL_DETAIL_FAILED) == ["failed"]


@pytest.mark.parametrize("name", list(_MODULES))
def test_partial_phase_failure_stays_done(name: str) -> None:
    """Прод-прогон 3597: 1 отказ из 11 → остаётся 'done' (частичный отказ — не отказ)."""
    assert _capture_status(_MODULES[name], PROD_3597_MOSTLY_OK) == ["done"]


@pytest.mark.parametrize("name", list(_MODULES))
def test_single_failed_attempt_stays_done(name: str) -> None:
    """Порог _PHASE_MIN_ATTEMPTS: одна неудачная попытка прогон не роняет."""
    assert _capture_status(_MODULES[name], SINGLE_ATTEMPT_FAILED) == ["done"]


@pytest.mark.parametrize("name", list(_MODULES))
def test_zero_attempts_stays_done(name: str) -> None:
    """Фаза не запускалась (0 попыток) — 0 == 0 не должно читаться как отказ."""
    counters = {**PROD_3597_MOSTLY_OK, "detail_attempted": 0, "detail_failed": 0}
    assert _capture_status(_MODULES[name], counters) == ["done"]