"""test_server.py — юниты per-provider concurrency для tradein-browser (#1793).

Проверяет ядро per-provider модели:
  1. _provider_for_url — корректный маппинг host → provider (avito/cian/yandex/generic);
  2. _provider_proxy — BROWSER_PROXY_* primary с фолбэком на legacy *_PROXY_URL;
  3. _resolve_provider — явный body["provider"] перебивает host, иначе host;
  4. конкурентность: два /fetch к РАЗНЫМ поставщикам идут параллельно (разные локи),
     два к ОДНОМУ — сериализованы (один лок). Через _do_fetch с asyncio.Event и
     замером перекрытия по времени.

camoufox НЕ запускается: _ensure_browser / _do_fetch мокаются. Хендлер вызывается
напрямую (make_mocked_request не поднимает сокет).

Запуск (из tradein-mvp/browser/)::

    python -m pytest test_server.py -q
"""

from __future__ import annotations

import asyncio
import importlib.util
import json
from pathlib import Path
from typing import Any

import pytest
from aiohttp.test_utils import make_mocked_request

# server.py — не пакет (отдельный сервис без __init__/pyproject). Грузим по пути.
_SERVER_PATH = Path(__file__).resolve().parent / "server.py"
_spec = importlib.util.spec_from_file_location("tradein_browser_server", _SERVER_PATH)
assert _spec is not None and _spec.loader is not None
server = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(server)


def _json_body(response: Any) -> dict[str, Any]:
    return json.loads(response.body.decode())


@pytest.fixture(autouse=True)
def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None:
    """Чистое per-provider состояние + инициализированный _locks_guard на каждый тест."""
    monkeypatch.setattr(server, "_browsers", {})
    monkeypatch.setattr(server, "_browser_cms", {})
    monkeypatch.setattr(server, "_page_counters", {})
    monkeypatch.setattr(server, "_locks", {})
    monkeypatch.setattr(server, "_retry_tasks", {})
    monkeypatch.setattr(server, "_locks_guard", asyncio.Lock())


# ── _provider_for_url ────────────────────────────────────────────────────────────


def test_provider_for_url_avito() -> None:
    assert server._provider_for_url("https://www.avito.ru/moskva/kvartiry") == "avito"
    assert server._provider_for_url("https://m.avito.ru/item/123") == "avito"


def test_provider_for_url_cian() -> None:
    assert server._provider_for_url("https://www.cian.ru/sale/flat/1/") == "cian"
    assert server._provider_for_url("https://spb.cian.ru/x") == "cian"


def test_provider_for_url_yandex() -> None:
    assert server._provider_for_url("https://realty.yandex.ru/offer/1/") == "yandex"


def test_provider_for_url_generic_fallback() -> None:
    assert server._provider_for_url("https://example.com/x") == "generic"
    assert server._provider_for_url("https://domclick.ru/x") == "generic"
    # Битый / пустой URL — без хоста → generic, не падаем.
    assert server._provider_for_url("not-a-url") == "generic"
    assert server._provider_for_url("") == "generic"


# ── _provider_proxy ──────────────────────────────────────────────────────────────


def test_provider_proxy_uses_browser_proxy_primary(monkeypatch: pytest.MonkeyPatch) -> None:
    """BROWSER_PROXY_* — primary источник для каждого поставщика."""
    monkeypatch.setenv("BROWSER_PROXY_AVITO", "http://u:p@avito-proxy:8080")
    monkeypatch.setenv("BROWSER_PROXY_CIAN", "http://u:p@cian-proxy:8080")
    monkeypatch.setenv("BROWSER_PROXY_YANDEX", "http://u:p@yandex-proxy:8080")
    assert server._provider_proxy("avito") == "http://u:p@avito-proxy:8080"
    assert server._provider_proxy("cian") == "http://u:p@cian-proxy:8080"
    assert server._provider_proxy("yandex") == "http://u:p@yandex-proxy:8080"


def test_provider_proxy_falls_back_to_legacy(monkeypatch: pytest.MonkeyPatch) -> None:
    """Без BROWSER_PROXY_* берётся legacy *_PROXY_URL фолбэк."""
    monkeypatch.delenv("BROWSER_PROXY_AVITO", raising=False)
    monkeypatch.delenv("BROWSER_PROXY_CIAN", raising=False)
    monkeypatch.delenv("BROWSER_PROXY_YANDEX", raising=False)
    monkeypatch.setenv("AVITO_PROXY_URL", "http://u:p@legacy-avito:8080")
    monkeypatch.setenv("CIAN_PROXY_URL", "http://u:p@legacy-cian:8080")
    monkeypatch.setenv("YANDEX_PROXY_URL", "http://u:p@legacy-yandex:8080")
    assert server._provider_proxy("avito") == "http://u:p@legacy-avito:8080"
    assert server._provider_proxy("cian") == "http://u:p@legacy-cian:8080"
    assert server._provider_proxy("yandex") == "http://u:p@legacy-yandex:8080"


def test_provider_proxy_generic(monkeypatch: pytest.MonkeyPatch) -> None:
    """generic: SCRAPER_PROXY_URL primary, AVITO_PROXY_URL фолбэк."""
    monkeypatch.delenv("SCRAPER_PROXY_URL", raising=False)
    monkeypatch.setenv("AVITO_PROXY_URL", "http://u:p@avito:8080")
    assert server._provider_proxy("generic") == "http://u:p@avito:8080"
    monkeypatch.setenv("SCRAPER_PROXY_URL", "http://u:p@scraper:8080")
    assert server._provider_proxy("generic") == "http://u:p@scraper:8080"


def test_provider_proxy_none_when_unset(monkeypatch: pytest.MonkeyPatch) -> None:
    """Ни одной env → None (direct, как одиночный браузер без прокси)."""
    for var in (
        "BROWSER_PROXY_CIAN",
        "CIAN_PROXY_URL",
    ):
        monkeypatch.delenv(var, raising=False)
    assert server._provider_proxy("cian") is None


def test_provider_proxy_browser_proxy_wins_over_legacy(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """BROWSER_PROXY_* перебивает legacy *_PROXY_URL (http vs возможный socks5)."""
    monkeypatch.setenv("BROWSER_PROXY_CIAN", "http://u:p@http-cian:8080")
    monkeypatch.setenv("CIAN_PROXY_URL", "socks5://u:p@socks-cian:1080")
    assert server._provider_proxy("cian") == "http://u:p@http-cian:8080"


# ── _resolve_provider ────────────────────────────────────────────────────────────


def test_resolve_provider_by_host() -> None:
    assert server._resolve_provider({}, "https://www.avito.ru/x") == "avito"
    assert server._resolve_provider({}, "https://www.cian.ru/x") == "cian"


def test_resolve_provider_explicit_overrides_host() -> None:
    """Явный валидный body["provider"] перебивает host."""
    assert server._resolve_provider({"provider": "yandex"}, "https://www.avito.ru/x") == "yandex"


def test_resolve_provider_invalid_explicit_falls_back_to_host() -> None:
    """Невалидный body["provider"] игнорируется → берём по host."""
    assert server._resolve_provider({"provider": "bogus"}, "https://www.cian.ru/x") == "cian"
    assert server._resolve_provider({"provider": 123}, "https://www.cian.ru/x") == "cian"


def test_resolve_provider_uses_source_field() -> None:
    """browser_fetcher шлёт {"url","source"} — explicit берётся и из "source"."""
    assert server._resolve_provider({"source": "avito"}, "https://example.com/x") == "avito"
    assert server._resolve_provider({"source": "yandex"}, "https://www.avito.ru/x") == "yandex"


def test_resolve_provider_domclick_source_to_generic() -> None:
    """domclick scraper шлёт source='domklik' (нет в PROVIDERS) + хост domclick.ru.

    'domklik' не валиден → host-detect domclick.ru → generic. Намеренно: в prod
    нет выделенного BROWSER_PROXY_DOMCLICK.
    """
    assert (
        server._resolve_provider({"source": "domklik"}, "https://domclick.ru/x") == "generic"
    )


# ── _lock_for: один лок на поставщика ────────────────────────────────────────────


def test_lock_for_is_stable_per_provider() -> None:
    """_lock_for возвращает один и тот же объект-лок для поставщика, разные — для разных."""

    async def _run() -> tuple[Any, Any, Any]:
        a1 = await server._lock_for("avito")
        a2 = await server._lock_for("avito")
        c1 = await server._lock_for("cian")
        return a1, a2, c1

    a1, a2, c1 = asyncio.run(_run())
    assert a1 is a2, "лок одного поставщика стабилен"
    assert a1 is not c1, "локи разных поставщиков различны"


# ── concurrency: параллель между поставщиками, сериализация внутри ───────────────


def _make_fetch_request(url: str) -> Any:
    request = make_mocked_request("POST", "/fetch")
    request.json = lambda: _coro({"url": url})  # type: ignore[method-assign]
    return request


class _OverlapProbe:
    """Мок _do_fetch: меряет максимальное число одновременно активных fetch per-provider."""

    def __init__(self, hold: float = 0.05) -> None:
        self.hold = hold
        self.active: dict[str, int] = {}
        self.peak: dict[str, int] = {}
        self.global_active = 0
        self.global_peak = 0

    async def __call__(self, provider: str, url: str) -> str:
        self.active[provider] = self.active.get(provider, 0) + 1
        self.peak[provider] = max(self.peak.get(provider, 0), self.active[provider])
        self.global_active += 1
        self.global_peak = max(self.global_peak, self.global_active)
        try:
            await asyncio.sleep(self.hold)
        finally:
            self.active[provider] -= 1
            self.global_active -= 1
        return f"<html>{provider}</html>"


def test_same_provider_fetches_are_serialized(monkeypatch: pytest.MonkeyPatch) -> None:
    """Два /fetch к ОДНОМУ поставщику сериализованы (один лок) — peak==1."""
    probe = _OverlapProbe()

    async def _ensure(provider: str) -> bool:
        server._browsers[provider] = object()
        return True

    monkeypatch.setattr(server, "_ensure_browser", _ensure)
    monkeypatch.setattr(server, "_do_fetch", probe)

    async def _run() -> None:
        await asyncio.gather(
            server.fetch_handler(_make_fetch_request("https://www.avito.ru/a")),
            server.fetch_handler(_make_fetch_request("https://www.avito.ru/b")),
            server.fetch_handler(_make_fetch_request("https://www.avito.ru/c")),
        )

    asyncio.run(_run())
    assert probe.peak["avito"] == 1, "внутри поставщика строго ≤1 одновременный fetch"


def test_different_providers_fetch_concurrently(monkeypatch: pytest.MonkeyPatch) -> None:
    """Два /fetch к РАЗНЫМ поставщикам идут параллельно (разные локи) — global peak>1."""
    probe = _OverlapProbe()

    async def _ensure(provider: str) -> bool:
        server._browsers[provider] = object()
        return True

    monkeypatch.setattr(server, "_ensure_browser", _ensure)
    monkeypatch.setattr(server, "_do_fetch", probe)

    async def _run() -> list[Any]:
        return await asyncio.gather(
            server.fetch_handler(_make_fetch_request("https://www.avito.ru/a")),
            server.fetch_handler(_make_fetch_request("https://www.cian.ru/b")),
            server.fetch_handler(_make_fetch_request("https://realty.yandex.ru/c")),
        )

    responses = asyncio.run(_run())
    assert all(r.status == 200 for r in responses)
    # >=2 (не ==3): на нагруженном CI планировщик может не вывести все три корутины
    # в overlap одновременно; перекрытие хотя бы двух уже доказывает параллельность
    # между поставщиками. Same-provider тест (peak==1) остаётся строгим.
    assert probe.global_peak >= 2, "разные поставщики должны фетчить конкурентно"
    # У каждого поставщика по одному запросу — внутренний peak == 1.
    assert probe.peak == {"avito": 1, "cian": 1, "yandex": 1}


def test_mixed_load_serializes_within_parallelizes_across(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Смешанная нагрузка: avito×2 сериализованы, но параллельны с cian×1."""
    probe = _OverlapProbe()

    async def _ensure(provider: str) -> bool:
        server._browsers[provider] = object()
        return True

    monkeypatch.setattr(server, "_ensure_browser", _ensure)
    monkeypatch.setattr(server, "_do_fetch", probe)

    async def _run() -> None:
        await asyncio.gather(
            server.fetch_handler(_make_fetch_request("https://www.avito.ru/a")),
            server.fetch_handler(_make_fetch_request("https://www.avito.ru/b")),
            server.fetch_handler(_make_fetch_request("https://www.cian.ru/c")),
        )

    asyncio.run(_run())
    assert probe.peak["avito"] == 1, "avito сериализован сам с собой"
    assert probe.peak["cian"] == 1
    assert probe.global_peak >= 2, "avito и cian должны перекрываться во времени"


async def _coro(value: Any) -> Any:
    return value