"""test_server_reuse_context.py — тёплый переиспользуемый browser context (#3118). Проблема: browser.new_page() создаёт НОВЫЙ изолированный context на КАЖДЫЙ /fetch, поэтому залитый ОДИН РАЗ снимок cookies (см. cookies в /fetch) не видит Set-Cookie от площадки — живой qrator_jsid2 DomClick (~2.5ч TTL) протухает после первого же запроса. reuse_context=True держит per-provider context живым между вызовами; куки вливаются ТОЛЬКО в момент его создания. reset_context=True выбрасывает сожжённый context ДО работы. Проверяет: 1. reuse_context=True — context создаётся ОДИН раз на несколько _fetch_once, куки вливаются только при создании (не повторно); 2. reset_context=True — старый context закрывается и выбрасывается, следующий _fetch_once создаёт новый (с новой инъекцией кук); 3. reuse_context=False (дефолт) — поведение НЕ меняется: browser.new_page() на каждый вызов, context вообще не создаётся/не трогается; 4. _close_browser закрывает и выбрасывает сохранённый context провайдера. camoufox НЕ запускается — browser/context/page подделаны. Запуск (из tradein-mvp/browser/):: python -m pytest test_server_reuse_context.py -q """ from __future__ import annotations import asyncio import importlib.util from pathlib import Path from typing import Any 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) @pytest.fixture(autouse=True) def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(server, "_browsers", {}) monkeypatch.setattr(server, "_browser_cms", {}) monkeypatch.setattr(server, "_contexts", {}) monkeypatch.setattr(server, "_page_counters", {}) monkeypatch.setattr(server, "_locks", {}) monkeypatch.setattr(server, "_last_goto_at", {}) # #3205: recycle threshold теперь per-provider (_RECYCLE_PAGES_BY_PROVIDER), не # единый BROWSER_RECYCLE_PAGES — патчим весь словарь высоким порогом, чтобы тесты # reuse/reset context (несколько _fetch_once подряд, включая domclick=1 по коду) # не триггерили настоящий _relaunch_browser/camoufox. monkeypatch.setattr( server, "_RECYCLE_PAGES_BY_PROVIDER", {p: 10_000 for p in server.PROVIDERS} ) class _FakePage: """Поддельная page: фиксирует goto/close, минимальный контракт _fetch_once.""" def __init__(self) -> None: self.goto_urls: list[str] = [] self.closed = 0 async def route(self, pattern: str, handler: Any) -> None: return None async def goto(self, url: str, **kwargs: Any) -> None: self.goto_urls.append(url) async def wait_for_timeout(self, ms: int) -> None: return None async def content(self) -> str: return "ok" async def close(self) -> None: self.closed += 1 class _FakeContext: """Поддельный BrowserContext: фиксирует add_cookies/new_page/close.""" def __init__(self) -> None: self.add_cookies_calls: list[list[dict]] = [] self.pages: list[_FakePage] = [] self.closed = 0 async def add_cookies(self, cookies: list[dict]) -> None: self.add_cookies_calls.append(cookies) async def new_page(self) -> _FakePage: page = _FakePage() self.pages.append(page) return page async def close(self) -> None: self.closed += 1 class _FakeReuseBrowser: """Поддельный Browser: считает new_context()/new_page() вызовы, копит контексты.""" def __init__(self) -> None: self.contexts_created: list[_FakeContext] = [] self.new_page_calls = 0 async def new_context(self) -> _FakeContext: ctx = _FakeContext() self.contexts_created.append(ctx) return ctx async def new_page(self) -> _FakePage: self.new_page_calls += 1 return _FakePage() # ── reuse_context=True: один context на несколько fetch ─────────────────────────── def test_reuse_context_creates_context_once_across_fetches() -> None: browser = _FakeReuseBrowser() server._browsers["avito"] = browser async def _run() -> None: await server._fetch_once("avito", "https://www.avito.ru/1", reuse_context=True) await server._fetch_once("avito", "https://www.avito.ru/2", reuse_context=True) await server._fetch_once("avito", "https://www.avito.ru/3", reuse_context=True) asyncio.run(_run()) assert len(browser.contexts_created) == 1, "context должен создаваться РОВНО один раз" ctx = browser.contexts_created[0] assert len(ctx.pages) == 3, "каждый fetch — своя страница ИЗ переиспользуемого context" assert browser.new_page_calls == 0, "browser.new_page() не должен зваться при reuse_context" # Все страницы закрыты, context — нет (остаётся жить). assert all(p.closed == 1 for p in ctx.pages) assert ctx.closed == 0 assert server._contexts["avito"] is ctx def test_reuse_context_injects_cookies_only_on_creation() -> None: browser = _FakeReuseBrowser() server._browsers["domclick"] = browser cookies = {"qrator_jsid2": "live-token"} async def _run() -> None: await server._fetch_once( "domclick", "https://domclick.ru/card/1", cookies=cookies, reuse_context=True ) await server._fetch_once( "domclick", "https://domclick.ru/card/2", cookies=cookies, reuse_context=True ) asyncio.run(_run()) ctx = browser.contexts_created[0] assert len(ctx.add_cookies_calls) == 1, "повторный fetch НЕ должен вливать куки заново" injected = ctx.add_cookies_calls[0] assert { "name": "qrator_jsid2", "value": "live-token", "domain": ".domclick.ru", "path": "/", } in injected # ── reset_context=True: старый context выбрасывается, следующий — заново ────────── def test_reset_context_discards_old_context_and_creates_new() -> None: browser = _FakeReuseBrowser() server._browsers["domclick"] = browser async def _run() -> None: await server._fetch_once( "domclick", "https://domclick.ru/1", cookies={"a": "1"}, reuse_context=True ) await server._fetch_once( "domclick", "https://domclick.ru/2", cookies={"a": "2"}, reuse_context=True, reset_context=True, ) asyncio.run(_run()) assert len(browser.contexts_created) == 2, "reset_context должен вызвать пересоздание" old_ctx, new_ctx = browser.contexts_created assert old_ctx.closed == 1, "старый context должен быть закрыт" assert new_ctx.add_cookies_calls == [ [{"name": "a", "value": "2", "domain": ".domclick.ru", "path": "/"}] ] assert server._contexts["domclick"] is new_ctx def test_reset_context_without_reuse_is_noop_for_new_page_path() -> None: """reset_context=True без reuse_context (не должно ломать текущий new_page-путь).""" browser = _FakeReuseBrowser() server._browsers["avito"] = browser html = asyncio.run( server._fetch_once("avito", "https://www.avito.ru/1", reset_context=True) ) assert html == "ok" assert browser.new_page_calls == 1 assert browser.contexts_created == [] # ── reuse_context=False (дефолт): golden-parity, поведение не меняется ──────────── def test_reuse_context_false_uses_new_page_every_time() -> None: browser = _FakeReuseBrowser() server._browsers["cian"] = browser async def _run() -> None: await server._fetch_once("cian", "https://www.cian.ru/1") await server._fetch_once("cian", "https://www.cian.ru/2") asyncio.run(_run()) assert browser.new_page_calls == 2 assert browser.contexts_created == [], "reuse_context=False не должен создавать context" assert "cian" not in server._contexts def test_reuse_context_false_still_injects_cookies_into_page_context( monkeypatch: pytest.MonkeyPatch, ) -> None: """Golden-parity: cookies без reuse_context вливаются в page.context, как раньше.""" class _FakePageContext: def __init__(self) -> None: self.add_cookies_calls: list[list[dict]] = [] async def add_cookies(self, cookies: list[dict]) -> None: self.add_cookies_calls.append(cookies) class _PageWithContext(_FakePage): def __init__(self) -> None: super().__init__() self.context = _FakePageContext() class _Browser: def __init__(self) -> None: self.page = _PageWithContext() async def new_page(self) -> _PageWithContext: return self.page browser = _Browser() server._browsers["yandex"] = browser asyncio.run( server._fetch_once("yandex", "https://realty.yandex.ru/1", cookies={"sid": "x"}) ) assert browser.page.context.add_cookies_calls == [ [{"name": "sid", "value": "x", "domain": ".realty.yandex.ru", "path": "/"}] ] # ── _close_browser закрывает переиспользуемый context (#3118) ───────────────────── def test_close_browser_closes_and_pops_reusable_context() -> None: ctx = _FakeContext() server._contexts["avito"] = ctx asyncio.run(server._close_browser("avito")) assert ctx.closed == 1 assert "avito" not in server._contexts def test_close_browser_noop_when_no_context() -> None: """Нет сохранённого context — _close_browser не падает (no-op ветка).""" asyncio.run(server._close_browser("generic")) assert "generic" not in server._contexts