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

"""Offline tests для unified scrapers-страницы admin-API (epic консолидации UI).

Покрытие эндпоинтов (db/httpx мокаются, NO live network/DB):
  - GET  /api/v1/admin/scrape/runs        — unified runs + total + фильтры
  - GET  /api/v1/admin/scraper/health     — fetch_mode + browser + providers

POST /scraper/{src}/rotate-ip (changeip mobileproxy) удалён #2616 шаг 3 — мёртвая
подписка (#2613); её тесты удалены вместе с ручкой.
"""

from __future__ import annotations

import os

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

from datetime import UTC, datetime
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient


@pytest.fixture
def client() -> TestClient:
    from app.api.v1 import admin as admin_module
    from app.core.db import get_db

    app = FastAPI()
    app.include_router(admin_module.router, prefix="/api/v1/admin")

    def fake_db():
        yield MagicMock()

    app.dependency_overrides[get_db] = fake_db
    return TestClient(app)


# ── API 1: GET /scrape/runs ──────────────────────────────────────────────────


def test_unified_runs_returns_rows_and_total(client: TestClient) -> None:
    """list_all → total + rows; ISO-форматирование дат + error_text.

    run_type из выдачи убран (#2674): вырожденная колонка (одно значение во всех
    3244 прогонах), вид прогона называет source.
    """
    fake_rows = [
        {
            "run_id": 5,
            "source": "avito_city_sweep",
            "status": "done",
            "params": {"pages_per_anchor": 3},
            "counters": {"lots_fetched": 120},
            "total_seen": 120,
            "new_count": 30,
            "started_at": datetime(2026, 6, 1, 10, tzinfo=UTC),
            "finished_at": datetime(2026, 6, 1, 11, tzinfo=UTC),
            "heartbeat_at": datetime(2026, 6, 1, 11, tzinfo=UTC),
            "error_text": None,
        }
    ]
    with patch("app.services.scrape_runs.list_all", return_value=(42, fake_rows)) as m:
        r = client.get("/api/v1/admin/scrape/runs")
    assert r.status_code == 200
    body = r.json()
    assert body["total"] == 42
    assert len(body["rows"]) == 1
    row = body["rows"][0]
    assert row["run_id"] == 5
    assert row["source"] == "avito_city_sweep"
    assert "run_type" not in row
    assert row["total_seen"] == 120
    assert row["new_count"] == 30
    assert row["error_text"] is None
    assert "2026-06-01" in row["started_at"]
    # defaults передаются в сервис
    _, kwargs = m.call_args
    assert kwargs["source"] is None
    assert kwargs["status"] is None
    assert kwargs["limit"] == 50
    assert kwargs["offset"] == 0


def test_unified_runs_passes_filters(client: TestClient) -> None:
    """source/status/limit/offset прокидываются в list_all."""
    with patch("app.services.scrape_runs.list_all", return_value=(0, [])) as m:
        r = client.get(
            "/api/v1/admin/scrape/runs",
            params={
                "source": "cian_city_sweep",
                "status": "banned",
                "limit": 10,
                "offset": 20,
            },
        )
    assert r.status_code == 200
    _, kwargs = m.call_args
    assert kwargs["source"] == "cian_city_sweep"
    assert kwargs["status"] == "banned"
    assert kwargs["limit"] == 10
    assert kwargs["offset"] == 20


def test_unified_runs_limit_over_max_422(client: TestClient) -> None:
    """limit=999 > max=200 → 422."""
    r = client.get("/api/v1/admin/scrape/runs", params={"limit": 999})
    assert r.status_code == 422


def test_unified_runs_negative_offset_422(client: TestClient) -> None:
    """offset=-1 < 0 → 422."""
    r = client.get("/api/v1/admin/scrape/runs", params={"offset": -1})
    assert r.status_code == 422


def test_unified_runs_invalid_status_422(client: TestClient) -> None:
    """status с непредусмотренным значением → 422 (Literal-валидация)."""
    r = client.get("/api/v1/admin/scrape/runs", params={"status": "unknown_status"})
    assert r.status_code == 422


def test_unified_runs_valid_statuses_200(client: TestClient) -> None:
    """Все допустимые значения status проходят валидацию (ожидаем 200, не 422)."""
    valid_statuses = ["done", "running", "banned", "zombie", "failed", "cancelled"]
    for s in valid_statuses:
        with patch("app.services.scrape_runs.list_all", return_value=(0, [])):
            r = client.get("/api/v1/admin/scrape/runs", params={"status": s})
        assert r.status_code == 200, f"status={s!r} должен возвращать 200, получили {r.status_code}"


# ── API 2: GET /scraper/health ───────────────────────────────────────────────


def test_health_assembles_browser_and_providers(client: TestClient) -> None:
    """browser /health + ipify пробинг → fetch_mode + browser + 3 провайдера."""
    from app.api.v1 import admin as admin_module

    async def fake_browser_health() -> admin_module.BrowserHealth:
        return admin_module.BrowserHealth(
            reachable=True,
            browsers={"avito": False, "cian": True, "yandex": False, "generic": True},
        )

    async def fake_current_ip(proxy_url: str | None) -> str | None:
        return "1.2.3.4" if proxy_url else None

    proxy_urls = {
        "avito": "http://u:p@host.a:10049",
        "cian": None,
        "yandex": "http://u:p@host.y:10051",
    }

    with (
        patch.object(admin_module, "_probe_browser_health", fake_browser_health),
        patch.object(admin_module, "_probe_current_ip", fake_current_ip),
        # #2830: узел резолвится из пула по (db, source), а не из статичного env
        patch.object(admin_module, "_provider_proxy_url", lambda _db, s: proxy_urls[s]),
        patch.object(admin_module.settings, "scraper_fetch_mode", "browser"),
    ):
        r = client.get("/api/v1/admin/scraper/health")

    assert r.status_code == 200
    body = r.json()
    assert body["fetch_mode"] == "browser"
    assert body["browser"]["reachable"] is True
    assert body["browser"]["browsers"]["cian"] is True
    assert body["browser"]["browsers"]["avito"] is False

    by_source = {p["source"]: p for p in body["providers"]}
    assert set(by_source) == {"avito", "cian", "yandex"}
    assert by_source["avito"]["proxy_host"] == "host.a"
    assert by_source["avito"]["proxy_port"] == 10049
    # #2616 шаг 2: changeip mobileproxy-ротация снята — rotate_supported всегда
    # False (живая ASocks-ротация — /admin/proxies/{id}/rotate, не per-source).
    assert by_source["avito"]["rotate_supported"] is False
    assert by_source["avito"]["current_ip"] == "1.2.3.4"
    # cian без proxy → host/port/ip null
    assert by_source["cian"]["proxy_host"] is None
    assert by_source["cian"]["current_ip"] is None
    assert by_source["cian"]["rotate_supported"] is False
    assert by_source["yandex"]["proxy_host"] == "host.y"
    assert by_source["yandex"]["rotate_supported"] is False


def test_health_browser_unreachable(client: TestClient) -> None:
    """httpx-ошибка на /health → reachable=False, browsers пуст."""
    from app.api.v1 import admin as admin_module

    async def fake_current_ip(proxy_url: str | None) -> str | None:
        return None

    class _BoomClient:
        def __init__(self, *a: Any, **k: Any) -> None:
            pass

        async def __aenter__(self) -> _BoomClient:
            return self

        async def __aexit__(self, *a: Any) -> None:
            return None

        async def get(self, *a: Any, **k: Any) -> Any:
            raise RuntimeError("browser down")

    with (
        patch.object(admin_module.httpx, "AsyncClient", _BoomClient),
        patch.object(admin_module, "_probe_current_ip", fake_current_ip),
        patch.object(admin_module, "_provider_proxy_url", lambda _db, _s: None),
    ):
        r = client.get("/api/v1/admin/scraper/health")

    assert r.status_code == 200
    body = r.json()
    assert body["browser"]["reachable"] is False
    assert body["browser"]["browsers"] == {}


# ── API 4: GET /scraper/pacing ───────────────────────────────────────────────


def test_pacing_get_proxies_browser_response(client: TestClient) -> None:
    """GET /scraper/pacing → парсит browser /pacing и возвращает PacingResponse."""
    from app.api.v1 import admin as admin_module

    fake_browser_resp = {
        "providers": [
            {"source": "avito", "interval_s": 3.0, "env_default_s": 2.0},
            {"source": "cian", "interval_s": 18.0, "env_default_s": 2.0},
            {"source": "yandex", "interval_s": 2.0, "env_default_s": 2.0},
            {"source": "generic", "interval_s": 2.0, "env_default_s": 2.0},
        ]
    }

    async def _fake_proxy() -> dict:
        return fake_browser_resp

    with patch.object(admin_module, "_proxy_browser_pacing_get", _fake_proxy):
        r = client.get("/api/v1/admin/scraper/pacing")

    assert r.status_code == 200
    body = r.json()
    assert "providers" in body
    by_source = {p["source"]: p for p in body["providers"]}
    assert set(by_source) == {"avito", "cian", "yandex", "generic"}
    assert by_source["cian"]["interval_s"] == 18.0
    assert by_source["avito"]["env_default_s"] == 2.0


def test_pacing_get_browser_unreachable_503(client: TestClient) -> None:
    """GET /scraper/pacing при недоступном browser → 503."""
    from fastapi import HTTPException

    from app.api.v1 import admin as admin_module

    async def _boom() -> dict:
        raise HTTPException(status_code=503, detail="browser unreachable: ConnectError")

    with patch.object(admin_module, "_proxy_browser_pacing_get", _boom):
        r = client.get("/api/v1/admin/scraper/pacing")

    assert r.status_code == 503


# ── API 5: PUT /scraper/pacing/{source} ─────────────────────────────────────


def test_pacing_put_proxies_to_browser(client: TestClient) -> None:
    """PUT /scraper/pacing/avito → проксирует на browser и отдаёт {ok,source,interval_s}."""
    from app.api.v1 import admin as admin_module

    captured: dict[str, Any] = {}

    class _FakeResp:
        def raise_for_status(self) -> None:
            return None

        def json(self) -> dict:
            return {"ok": True, "source": "avito", "interval_s": 7.5}

    class _FakeClient:
        def __init__(self, *a: Any, **k: Any) -> None:
            pass

        async def __aenter__(self) -> _FakeClient:
            return self

        async def __aexit__(self, *a: Any) -> None:
            return None

        async def put(self, url: str, json: Any = None, **k: Any) -> _FakeResp:
            captured["url"] = url
            captured["json"] = json
            return _FakeResp()

    with patch.object(admin_module.httpx, "AsyncClient", _FakeClient):
        r = client.put(
            "/api/v1/admin/scraper/pacing/avito",
            json={"interval_s": 7.5},
        )

    assert r.status_code == 200
    body = r.json()
    assert body["ok"] is True
    assert body["source"] == "avito"
    assert body["interval_s"] == 7.5
    assert captured["json"]["source"] == "avito"
    assert captured["json"]["interval_s"] == 7.5


def test_pacing_put_invalid_source_422(client: TestClient) -> None:
    """PUT /scraper/pacing/domclick → 422 (Literal-валидация)."""
    r = client.put("/api/v1/admin/scraper/pacing/domclick", json={"interval_s": 5.0})
    assert r.status_code == 422


def test_pacing_put_negative_interval_422(client: TestClient) -> None:
    """PUT /scraper/pacing/cian с interval_s < 0 → 422 (ge=0 валидация)."""
    r = client.put("/api/v1/admin/scraper/pacing/cian", json={"interval_s": -1.0})
    assert r.status_code == 422


def test_pacing_put_over_max_interval_422(client: TestClient) -> None:
    """PUT /scraper/pacing/yandex с interval_s > 120 → 422 (le=120 валидация)."""
    r = client.put("/api/v1/admin/scraper/pacing/yandex", json={"interval_s": 200.0})
    assert r.status_code == 422


def test_pacing_put_browser_unreachable_503(client: TestClient) -> None:
    """PUT /scraper/pacing/avito при недоступном browser → 503."""
    from app.api.v1 import admin as admin_module

    class _BoomClient:
        def __init__(self, *a: Any, **k: Any) -> None:
            pass

        async def __aenter__(self) -> _BoomClient:
            return self

        async def __aexit__(self, *a: Any) -> None:
            return None

        async def put(self, *a: Any, **k: Any) -> Any:
            raise RuntimeError("connection refused")

    with patch.object(admin_module.httpx, "AsyncClient", _BoomClient):
        r = client.put(
            "/api/v1/admin/scraper/pacing/avito",
            json={"interval_s": 5.0},
        )

    assert r.status_code == 503


# ── API 6: GET /scraper/data-quality ────────────────────────────────────────


def _make_dq_db_mock() -> MagicMock:
    """Мок db-сессии для GET /scraper/data-quality."""
    db = MagicMock()

    # listings rows: avito=1000 active (все свежие), cian=500 active / 300 протухших
    # (#2660: stale_count идёт рядом с active_count — «активно» ≠ «живо»).
    avito_row = {
        "source": "avito",
        "active_count": 1000,
        "stale_count": 0,
        "f_description": 900,
        "f_photo_urls": 980,
        "f_address": 1000,
        "f_lat": 950,
        "f_lon": 950,
        "f_kitchen_area_m2": 600,
        "f_living_area_m2": 100,
        "f_ceiling_height_m": 300,
        "f_metro_stations": 700,
    }
    cian_row = {
        "source": "cian",
        "active_count": 500,
        "stale_count": 300,
        "f_description": 450,
        "f_photo_urls": 490,
        "f_address": 500,
        "f_lat": 480,
        "f_lon": 480,
        "f_kitchen_area_m2": 400,
        "f_living_area_m2": 400,
        # #2699: cian тоже пишет в канон ceiling_height_m — раньше в фикстуре
        # стояло f_ceiling_height=350 / f_ceiling_height_m=0, ровно тот прод-раскол,
        # из-за которого 855 циановских значений не видел ни один потребитель.
        "f_ceiling_height_m": 350,
        "f_metro_stations": 100,
    }

    houses_row = {
        "total": 2000,
        "validated_cnt": 1800,
        "rating_cnt": 1200,
        "house_type_cnt": 1900,
    }

    listing_mappings = MagicMock()
    listing_mappings.all.return_value = [avito_row, cian_row]

    houses_mappings = MagicMock()
    houses_mappings.one.return_value = houses_row

    reviews_scalar = 4500

    # db.execute() вызывается 3 раза: listings SQL, houses SQL, house_reviews COUNT
    execute_results = iter(
        [
            MagicMock(mappings=lambda: listing_mappings),
            MagicMock(mappings=lambda: houses_mappings),
            MagicMock(scalar=lambda: reviews_scalar),
        ]
    )
    db.execute.side_effect = lambda *a, **k: next(execute_results)
    return db


def test_data_quality_shape(client: TestClient) -> None:
    """GET /scraper/data-quality → StructureCheck: sources + houses + reviews_count."""
    from app.api.v1 import admin as admin_module  # noqa: F401
    from app.core.db import get_db

    app = client.app
    app.dependency_overrides[get_db] = lambda: _make_dq_db_mock()

    r = client.get("/api/v1/admin/scraper/data-quality")
    assert r.status_code == 200
    body = r.json()

    assert "sources" in body
    assert "houses" in body

    # two sources
    sources = {s["source"]: s for s in body["sources"]}
    assert set(sources) >= {"avito", "cian"}

    avito = sources["avito"]
    assert avito["active_count"] == 1000
    assert avito["stale_count"] == 0  # #2660
    fields = avito["fields"]
    assert 0.0 <= fields["description"] <= 100.0
    assert fields["description"] == pytest.approx(90.0, abs=0.1)
    assert fields["photo_urls"] == pytest.approx(98.0, abs=0.1)
    assert fields["kitchen_area_m2"] == pytest.approx(60.0, abs=0.1)

    cian = sources["cian"]
    assert cian["active_count"] == 500
    assert cian["stale_count"] == 300  # #2660: 60% «активных» на деле не виделись
    assert cian["fields"]["ceiling_height_m"] == pytest.approx(70.0, abs=0.1)
    assert "ceiling_height" not in cian["fields"]  # #2699: DEPRECATED, убрана

    houses = body["houses"]
    assert houses["total"] == 2000
    assert houses["validated_pct"] == pytest.approx(90.0, abs=0.1)
    assert houses["rating_pct"] == pytest.approx(60.0, abs=0.1)
    assert houses["house_type_pct"] == pytest.approx(95.0, abs=0.1)
    assert houses["reviews_count"] == 4500


def test_data_quality_pct_in_range(client: TestClient) -> None:
    """Все fill% в [0,100]."""
    from app.core.db import get_db

    app = client.app
    app.dependency_overrides[get_db] = lambda: _make_dq_db_mock()

    r = client.get("/api/v1/admin/scraper/data-quality")
    assert r.status_code == 200
    body = r.json()

    for src in body["sources"]:
        for field_name, pct in src["fields"].items():
            assert (
                0.0 <= pct <= 100.0
            ), f"source={src['source']} field={field_name} pct={pct} вне [0,100]"


# ── Security: текст исключения не утекает в HTTP-ответ (#2234) ────────────────


def test_admin_error_detail_hides_exception_text(client: TestClient) -> None:
    """Внутренняя зависимость кидает исключение с 'секретным' текстом →

    в теле ответа НЕТ секрета, есть generic detail + error-id; статус-код прежний.
    Роут /scrape/avito-detail: fetch_detail падает с DSN-подобным сообщением.
    """
    secret = "postgres://user:SECRETPASS@internal-host:5432/db"
    with patch(
        "app.api.v1.admin.fetch_detail",
        new=AsyncMock(side_effect=Exception(secret)),
    ):
        r = client.post(
            "/api/v1/admin/scrape/avito-detail",
            params={"item_url": "/ekaterinburg/kvartiry/1-k_kvartira_100"},
        )

    assert r.status_code == 502
    body = r.json()
    detail = body["detail"]
    assert "SECRETPASS" not in detail
    assert secret not in detail
    assert "error-id:" in detail
    assert detail.startswith("fetch failed")