"""保活上下文（cf_clearance 复用）逻辑的单元测试（不启动真实 Chromium）。""" from __future__ import annotations from contextlib import asynccontextmanager from types import SimpleNamespace import pytest from app.core.config import Settings from app.services.cloudflare_session import CloudflareSessionManager from app.services.session_store import SessionStore _DETAIL_URL = "https://www.suruga-ya.jp/product/detail/{}" class _FakePage: """最小页面替身：goto 状态由所属上下文是否被标记失败决定。""" def __init__(self, pool: "_FakeBrowserPool", context: "_FakeContext"): self._pool = pool self._context = context async def goto(self, url: str, **_kw): status = 403 if self._context.idx in self._pool.fail_idxs else 200 return SimpleNamespace(status=status) async def title(self) -> str: return "ok" async def content(self) -> str: return self._pool.content_html async def screenshot(self, **_kw): # pragma: no cover - 非 200 分支才用到 return None async def wait_for_timeout(self, _ms): # pragma: no cover return None class _FakeContext: def __init__(self, idx: int): self.idx = idx self.closed = False async def cookies(self): return [{"name": "cf_clearance", "value": f"cv-{self.idx}"}] async def close(self): self.closed = True class _FakeBrowserPool: """记录上下文创建次数，并允许把指定上下文标记为「复用必失败」。""" def __init__(self): self.content_html = "ok" self.create_calls = 0 self.contexts: list[_FakeContext] = [] self.fail_idxs: set[int] = set() async def create_persistent_context(self, storage_state=None) -> _FakeContext: self.create_calls += 1 ctx = _FakeContext(self.create_calls) self.contexts.append(ctx) return ctx @asynccontextmanager async def page_in_context(self, context: _FakeContext): yield _FakePage(self, context) async def close_context(self, context: _FakeContext): await context.close() def _make_manager(pool: _FakeBrowserPool) -> CloudflareSessionManager: settings = Settings(session_ttl_seconds=900, cloudflare_reuse_wait_timeout_seconds=8.0) return CloudflareSessionManager(settings, pool, SessionStore(settings)) @pytest.mark.asyncio async def test_warm_context_is_reused_without_rebuild(): pool = _FakeBrowserPool() mgr = _make_manager(pool) html1, sess1 = await mgr.fetch_html(_DETAIL_URL.format(1)) assert html1 == pool.content_html assert pool.create_calls == 1 # 首次构建 # 第二次请求（不同商品）应复用同一保活上下文，不再新建 html2, sess2 = await mgr.fetch_html(_DETAIL_URL.format(2)) assert html2 == pool.content_html assert pool.create_calls == 1 assert sess2.session_id == sess1.session_id @pytest.mark.asyncio async def test_warm_context_rebuilds_and_closes_on_failure(): pool = _FakeBrowserPool() mgr = _make_manager(pool) _, sess1 = await mgr.fetch_html(_DETAIL_URL.format(1)) assert pool.create_calls == 1 # 让首个上下文在后续复用导航时返回 403（模拟 clearance 失效） pool.fail_idxs.add(1) html2, sess2 = await mgr.fetch_html(_DETAIL_URL.format(2)) assert html2 == pool.content_html assert pool.create_calls == 2 # 复用失败后重建 assert pool.contexts[0].closed is True # 旧上下文已退役关闭 assert sess2.session_id != sess1.session_id # 换了新会话 @pytest.mark.asyncio async def test_expired_warm_context_is_closed_on_rebuild(): pool = _FakeBrowserPool() # 存活时长设为 0：任何已存在的保活上下文都立即判定过期 settings = Settings(session_ttl_seconds=0, cloudflare_reuse_wait_timeout_seconds=8.0) mgr = CloudflareSessionManager(settings, pool, SessionStore(settings)) await mgr.fetch_html(_DETAIL_URL.format(1)) assert pool.create_calls == 1 # 第二次请求：旧上下文过期 -> 退役关闭 -> 重建，旧上下文不应泄漏 await mgr.fetch_html(_DETAIL_URL.format(2)) assert pool.create_calls == 2 assert pool.contexts[0].closed is True assert len(mgr._warm_contexts) == 1 # 注册表中只保留新的那个 @pytest.mark.asyncio async def test_close_releases_all_warm_contexts(): pool = _FakeBrowserPool() mgr = _make_manager(pool) await mgr.fetch_html(_DETAIL_URL.format(1)) assert pool.contexts[0].closed is False await mgr.close() assert pool.contexts[0].closed is True assert mgr._warm_contexts == {}