"""test_server_smoke.py — смоук для resilience-логики tradein-browser.

Полный браузер (camoufox subprocess) юнит-тестить непрактично, поэтому здесь
только смоук: (1) модуль импортируется; (2) /fetch и /login отдают 503 когда
браузер не поднят (прокси лежит) — критичный resilience-инвариант, защищающий
от crash-storm. camoufox в окружении теста НЕ нужен: launch замокан no-op'ом,
имитируя «прокси всё ещё недоступен».

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

    python -m pytest test_server_smoke.py -q

aiohttp.test_utils.make_mocked_request не поднимает реальный сервер/сокет —
хендлер вызывается напрямую, без event-loop-сервера.
"""

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]:
    """Достаёт JSON-тело из aiohttp Response (body — bytes после json_response)."""
    return json.loads(response.body.decode())


def test_module_imports() -> None:
    """Смоук: модуль импортируется и экспонирует ключевые хендлеры."""
    assert callable(server.fetch_handler)
    assert callable(server.login_handler)
    assert callable(server.health_handler)


def test_health_reports_browser_not_ready(monkeypatch: pytest.MonkeyPatch) -> None:
    """/health всегда 200, browser_ready=False когда браузер не поднят."""
    monkeypatch.setattr(server, "_browser", None)
    request = make_mocked_request("GET", "/health")
    response = asyncio.run(server.health_handler(request))
    assert response.status == 200
    body = _json_body(response)
    assert body["status"] == "ok"
    assert body["browser_ready"] is False


def test_fetch_returns_503_when_browser_unavailable(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """/fetch отдаёт 503 (не 500/краш) когда браузер не поднят и launch не удался.

    Имитирует «прокси лежит»: _browser=None + _ensure_browser no-op,
    оставляющий None. Никакого реального camoufox-launch'а / direct-fetch.
    """
    monkeypatch.setattr(server, "_browser", None)
    monkeypatch.setattr(server, "_fetch_sem", asyncio.Semaphore(4))

    async def _no_ensure() -> bool:
        return False  # прокси всё ещё недоступен → браузер остаётся None

    monkeypatch.setattr(server, "_ensure_browser", _no_ensure)

    request = make_mocked_request(
        "POST", "/fetch", payload=b'{"url": "https://example.com"}'
    )

    async def _run() -> Any:
        # make_mocked_request не парсит тело — подменяем .json() на нужный payload.
        request.json = lambda: _coro({"url": "https://example.com"})  # type: ignore[method-assign]
        return await server.fetch_handler(request)

    response = asyncio.run(_run())
    assert response.status == 503
    body = _json_body(response)
    assert "browser unavailable" in body["error"]


def test_login_returns_503_when_browser_unavailable(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """/login тоже отдаёт 503 когда браузер не поднят (без credential-логирования)."""
    monkeypatch.setattr(server, "_browser", None)
    monkeypatch.setattr(server, "_fetch_sem", asyncio.Semaphore(4))

    async def _no_ensure() -> bool:
        return False

    monkeypatch.setattr(server, "_ensure_browser", _no_ensure)

    login_body = {
        "url": "https://cian.example/login",
        "email": "u@example.com",
        "password": "secret",
        "email_selector": "#email",
        "password_selector": "#password",
        "submit_selector": "#submit",
    }
    request = make_mocked_request("POST", "/login")

    async def _run() -> Any:
        request.json = lambda: _coro(login_body)  # type: ignore[method-assign]
        return await server.login_handler(request)

    response = asyncio.run(_run())
    assert response.status == 503
    body = _json_body(response)
    assert "browser unavailable" in body["error"]


# ── BROWSER_CONCURRENCY config parse ────────────────────────────────────────────


def test_concurrency_default_is_4() -> None:
    """BROWSER_CONCURRENCY не задан → дефолт 4."""
    assert server._parse_concurrency(None) == 4


def test_concurrency_env_override() -> None:
    """Валидное число из env применяется как есть."""
    assert server._parse_concurrency("8") == 8
    assert server._parse_concurrency("1") == 1


def test_concurrency_clamps_to_min_1() -> None:
    """0 / отрицательное зажимается до >=1 (семафор с 0 завесил бы сервис)."""
    assert server._parse_concurrency("0") == 1
    assert server._parse_concurrency("-5") == 1


def test_concurrency_garbage_falls_back_to_4() -> None:
    """Нечисловой мусор → дефолт 4 (кривой конфиг не валит сервис)."""
    assert server._parse_concurrency("abc") == 4
    assert server._parse_concurrency("") == 4


def test_startup_creates_semaphore_with_concurrency(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """_on_startup создаёт _fetch_sem с ёмкостью BROWSER_CONCURRENCY + launch-лок."""
    monkeypatch.setattr(server, "BROWSER_CONCURRENCY", 3)
    monkeypatch.setattr(server, "_fetch_sem", None)
    monkeypatch.setattr(server, "_launch_lock", None)

    async def _no_launch() -> bool:
        return True  # не запускаем реальный camoufox в тесте

    monkeypatch.setattr(server, "_try_launch_browser", _no_launch)

    app = server.web.Application()
    asyncio.run(server._on_startup(app))

    assert isinstance(server._fetch_sem, asyncio.Semaphore)
    assert server._fetch_sem._value == 3  # стартовая ёмкость == concurrency
    assert isinstance(server._launch_lock, asyncio.Lock)


# ── semaphore limits concurrency + per-page isolation ────────────────────────────


class _FakePage:
    """Поддельная page: считает open/close и отмечает пик параллелизма."""

    def __init__(self, tracker: "_Tracker") -> None:
        self._tracker = tracker

    async def route(self, pattern: str, handler: Any) -> None:
        # _apply_resource_block навешивает route-interception перед goto (perf:
        # блокировка image/media/font). Заглушка no-op — поведение route проверяется
        # отдельно в test_server_resource_block.py.
        return None

    async def goto(self, url: str, **kwargs: Any) -> None:
        # Внутри goto держим «активную» страницу — здесь меряем параллелизм.
        self._tracker.active += 1
        self._tracker.peak = max(self._tracker.peak, self._tracker.active)
        await asyncio.sleep(0.02)
        self._tracker.active -= 1

    async def wait_for_timeout(self, ms: int) -> None:
        return None

    async def content(self) -> str:
        return "<html>ok</html>"

    async def close(self) -> None:
        self._tracker.closed += 1


class _FakeBrowser:
    def __init__(self, tracker: "_Tracker") -> None:
        self._tracker = tracker

    async def new_page(self) -> _FakePage:
        self._tracker.opened += 1
        return _FakePage(self._tracker)


class _Tracker:
    def __init__(self) -> None:
        self.opened = 0
        self.closed = 0
        self.active = 0
        self.peak = 0


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


def test_semaphore_limits_concurrency_to_n(monkeypatch: pytest.MonkeyPatch) -> None:
    """Семафор(N) не даёт больше N одновременных goto, даже при M>N запросах."""
    tracker = _Tracker()
    browser = _FakeBrowser(tracker)
    monkeypatch.setattr(server, "_browser", browser)
    monkeypatch.setattr(server, "_fetch_sem", asyncio.Semaphore(2))
    monkeypatch.setattr(server, "_inflight", 0)
    # recycle отключаем — высокий порог, чтобы relaunch не вмешивался.
    monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000)

    async def _ensure() -> bool:
        return True

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

    async def _run() -> None:
        await asyncio.gather(*[server.fetch_handler(_make_fetch_request()) for _ in range(6)])

    asyncio.run(_run())

    assert tracker.peak <= 2, f"параллелизм {tracker.peak} превысил лимит семафора 2"
    assert tracker.opened == 6
    assert tracker.closed == 6  # каждая страница своя, закрыта в finally


def test_fetch_opens_and_closes_own_page(monkeypatch: pytest.MonkeyPatch) -> None:
    """Один /fetch открывает ровно одну собственную страницу и закрывает её."""
    tracker = _Tracker()
    browser = _FakeBrowser(tracker)
    monkeypatch.setattr(server, "_browser", browser)
    monkeypatch.setattr(server, "_fetch_sem", asyncio.Semaphore(4))
    monkeypatch.setattr(server, "_inflight", 0)
    monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000)

    async def _ensure() -> bool:
        return True

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

    response = asyncio.run(server.fetch_handler(_make_fetch_request()))
    assert response.status == 200
    assert _json_body(response)["html"] == "<html>ok</html>"
    assert tracker.opened == 1
    assert tracker.closed == 1


def test_ensure_browser_serialized_by_launch_lock(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """_ensure_browser сериализует launch: при N параллельных вызовах launch один раз."""
    monkeypatch.setattr(server, "_browser", None)
    monkeypatch.setattr(server, "_launch_lock", asyncio.Lock())
    calls = {"n": 0}

    async def _launch_once() -> bool:
        calls["n"] += 1
        await asyncio.sleep(0.01)
        # имитируем успешный launch — выставляем браузер
        monkeypatch.setattr(server, "_browser", object())
        return True

    monkeypatch.setattr(server, "_try_launch_browser", _launch_once)

    async def _run() -> list[bool]:
        return await asyncio.gather(*[server._ensure_browser() for _ in range(5)])

    results = asyncio.run(_run())
    assert all(results)
    assert calls["n"] == 1, "launch должен произойти ровно один раз (double-check под локом)"


# ── _safe_relaunch resets _page_counter (recycle starvation fix) ─────────────────


def test_safe_relaunch_resets_page_counter_when_proceeds(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """_safe_relaunch сбрасывает _page_counter=0 когда relaunch проходит (inflight<=1).

    Регрессия: без сброса под конкуренцией counter копился вечно (>= порога на
    каждом fetch) → recycle голодал. Сброс атомарен с relaunch'ем под локом.
    """
    monkeypatch.setattr(server, "_launch_lock", asyncio.Lock())
    monkeypatch.setattr(server, "_inflight", 1)  # мы единственный in-flight
    monkeypatch.setattr(server, "_page_counter", 999)

    relaunched = {"n": 0}

    async def _fake_relaunch() -> None:
        relaunched["n"] += 1

    monkeypatch.setattr(server, "_relaunch_browser", _fake_relaunch)

    asyncio.run(server._safe_relaunch("recycle"))

    assert relaunched["n"] == 1, "relaunch должен выполниться при inflight<=1"
    assert server._page_counter == 0, "counter должен сброситься при реальном relaunch"


def test_safe_relaunch_keeps_page_counter_when_deferred(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """_safe_relaunch НЕ трогает _page_counter когда relaunch отложен (inflight>1).

    Отложенный relaunch оставляет counter, чтобы следующий запрос (когда нагрузка
    спадёт) дожал recycle — иначе recycle потерялся бы навсегда.
    """
    monkeypatch.setattr(server, "_launch_lock", asyncio.Lock())
    monkeypatch.setattr(server, "_inflight", 3)  # есть параллельные запросы
    monkeypatch.setattr(server, "_page_counter", 42)

    relaunched = {"n": 0}

    async def _fake_relaunch() -> None:
        relaunched["n"] += 1

    monkeypatch.setattr(server, "_relaunch_browser", _fake_relaunch)

    asyncio.run(server._safe_relaunch("recycle"))

    assert relaunched["n"] == 0, "relaunch должен быть отложен при inflight>1"
    assert server._page_counter == 42, "counter НЕ трогаем при отложенном relaunch"


# ── _login_once closes page on CancelledError (BaseException leak fix) ────────────


class _CancellingLoginPage:
    """Page для /login, которая бросает CancelledError внутри логина."""

    def __init__(self, tracker: "_Tracker") -> None:
        self._tracker = tracker
        self.url = "https://cian.example/login"

    async def goto(self, url: str, **kwargs: Any) -> None:
        return None

    async def wait_for_timeout(self, ms: int) -> None:
        return None

    async def fill(self, selector: str, value: str, **kwargs: Any) -> None:
        # Отмена таска (BaseException, не Exception) — обходит except Exception.
        raise asyncio.CancelledError()

    async def close(self) -> None:
        self._tracker.closed += 1


class _CancellingLoginBrowser:
    def __init__(self, tracker: "_Tracker") -> None:
        self._tracker = tracker

    async def new_page(self) -> _CancellingLoginPage:
        self._tracker.opened += 1
        return _CancellingLoginPage(self._tracker)


def test_login_closes_page_on_cancelled_error(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """_login_once закрывает страницу даже при CancelledError (finally, не except).

    CancelledError — BaseException, обходит except Exception, без finally страница
    утекала бы при отмене таска (shutdown / dropped client).
    """
    tracker = _Tracker()
    browser = _CancellingLoginBrowser(tracker)
    monkeypatch.setattr(server, "_browser", browser)

    params = {
        "url": "https://cian.example/login",
        "email": "u@example.com",
        "password": "secret",
        "email_selector": "#email",
        "password_selector": "#password",
        "submit_selector": "#submit",
    }

    with pytest.raises(asyncio.CancelledError):
        asyncio.run(server._login_once(params))

    assert tracker.opened == 1
    assert tracker.closed == 1, "страница должна закрыться в finally при CancelledError"


async def _coro(value: Any) -> Any:
    """Хелпер: оборачивает значение в awaitable для подмены request.json()."""
    return value