"""test_server_instance_limit.py — потолок живых camoufox и выбор жертвы (#3412).

До #3412 инстанс поставщика поднимался лениво и не закрывался никогда: пять
поставщиков = пять вечных браузеров в контейнере с mem_limit 2560m. Каждый тест
ниже закрывает один дефект ревью прошлой попытки (ветка fix/3412-cian-instance-
relaunch) или сам потолок; проверка — по тому, какие инстансы ОСТАЛИСЬ живыми.

camoufox не запускается: модуль camoufox.async_api подменён заглушкой.

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

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

from __future__ import annotations

import asyncio
import importlib.util
import sys
import types
from pathlib import Path

import pytest

_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)


class _FakeCM:
    """AsyncCamoufox-заглушка: запуск по желанию медленный, закрытие пишет журнал."""

    launch_delay = 0.0
    closed: list[str] = []
    # provider → был ли он ещё в server._browsers в момент СВОЕГО закрытия
    visible_while_closing: dict[str, bool] = {}

    def __init__(self, **kwargs: object) -> None:
        pass

    async def __aenter__(self) -> object:
        await asyncio.sleep(_FakeCM.launch_delay)
        return object()

    async def __aexit__(self, *args: object) -> None:
        provider = next(p for p, cm in _cms_by_provider.items() if cm is self)
        _FakeCM.visible_while_closing[provider] = provider in server._browsers
        await asyncio.sleep(0)
        _FakeCM.closed.append(provider)


_cms_by_provider: dict[str, _FakeCM] = {}


@pytest.fixture(autouse=True)
def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None:
    for name in ("_browsers", "_browser_cms", "_contexts", "_anchor_pages", "_anchor_origins",
                 "_page_counters", "_locks", "_retry_tasks", "_last_goto_at",
                 "_launched_proxy", "_last_used"):
        monkeypatch.setattr(server, name, {})
    monkeypatch.setattr(server, "_launching", set())
    monkeypatch.setattr(server, "_locks_guard", asyncio.Lock())
    monkeypatch.setattr(server, "BROWSER_MAX_INSTANCES", 2)
    _FakeCM.launch_delay = 0.0
    _FakeCM.closed = []
    _FakeCM.visible_while_closing = {}
    _cms_by_provider.clear()

    fake_mod = types.ModuleType("camoufox.async_api")
    fake_mod.AsyncCamoufox = _FakeCM  # type: ignore[attr-defined]
    monkeypatch.setitem(sys.modules, "camoufox", types.ModuleType("camoufox"))
    monkeypatch.setitem(sys.modules, "camoufox.async_api", fake_mod)


async def _launch(provider: str) -> None:
    await server._launch_browser(provider, proxy_override="http://u:p@pool:8080")
    _cms_by_provider[provider] = server._browser_cms[provider]


def _live() -> set[str]:
    return set(server._browsers)


def test_launch_at_cap_closes_idle_instance() -> None:
    async def _run() -> None:
        await _launch("cian")
        await _launch("yandex")
        await _launch("generic")

    asyncio.run(_run())

    assert _live() == {"yandex", "generic"}
    assert _FakeCM.closed == ["cian"]


def test_victim_is_cheap_instance_not_avito_with_anchor() -> None:
    """Дефект ревью 1: LRU закрывал avito с якорной вкладкой, оставляя cian без context'а."""

    async def _run() -> None:
        await _launch("avito")  # использован ДАВНЕЕ всех, но с прогретым context'ом
        server._contexts["avito"] = types.SimpleNamespace(close=_noop)
        server._anchor_pages["avito"] = object()
        await _launch("cian")
        await _launch("yandex")

    asyncio.run(_run())

    assert _live() == {"avito", "yandex"}
    assert _FakeCM.closed == ["cian"]


async def _noop() -> None:
    return None


def test_eviction_keeps_victim_pacing() -> None:
    """Дефект ревью 2: вытеснение сбрасывало _last_goto_at чужого поставщика."""

    async def _run() -> None:
        await _launch("cian")
        server._last_goto_at["cian"] = 12345.0
        await _launch("yandex")
        await _launch("generic")

    asyncio.run(_run())

    assert "cian" not in _live()
    assert server._last_goto_at.get("cian") == 12345.0


def test_background_retry_path_respects_cap() -> None:
    """Дефект ревью 3: retry-путь _try_launch_browser поднимал инстанс мимо потолка."""

    async def _run() -> bool:
        await _launch("cian")
        await _launch("yandex")
        return await server._try_launch_browser("avito")

    assert asyncio.run(_run()) is True
    assert len(_live()) == 2
    assert "avito" in _live()


def test_fresh_instance_is_not_the_next_victim() -> None:
    """Дефект ревью 4: свежий инстанс без отметки использования уходил первым."""

    async def _run() -> None:
        await _launch("cian")
        assert await server._ensure_browser("cian", proxy_override="http://u:p@pool:8080")
        await _launch("yandex")  # свежий, после последнего обращения к cian
        await _launch("generic")  # должен закрыть давний cian, а не свежий yandex

    asyncio.run(_run())

    assert _FakeCM.closed == ["cian"]


def test_use_refreshes_lru_order() -> None:
    async def _run() -> None:
        await _launch("cian")
        await _launch("yandex")
        assert await server._ensure_browser("cian", proxy_override="http://u:p@pool:8080")
        await _launch("generic")

    asyncio.run(_run())

    assert _FakeCM.closed == ["yandex"]


def test_busy_instance_is_never_closed() -> None:
    async def _run() -> None:
        await _launch("cian")
        await _launch("yandex")
        for provider in ("cian", "yandex"):
            await (await server._lock_for(provider)).acquire()
        await _launch("generic")

    asyncio.run(_run())

    assert _FakeCM.closed == []
    assert _live() == {"cian", "yandex", "generic"}


def test_parallel_launches_reserve_their_slot() -> None:
    """Два запуска разных поставщиков одновременно не должны оба увидеть свободное место."""

    async def _run() -> None:
        await _launch("cian")
        _FakeCM.launch_delay = 0.01
        await asyncio.gather(_launch("avito"), _launch("yandex"))

    asyncio.run(_run())

    assert _live() == {"avito", "yandex"}


def test_evicted_instance_disappears_before_close_awaits() -> None:
    """Жертву закрывает чужая корутина без её лока: во время close() браузер уже снят."""

    async def _run() -> None:
        await _launch("cian")
        await _launch("yandex")
        await _launch("generic")

    asyncio.run(_run())

    assert _FakeCM.visible_while_closing == {"cian": False}