125 lines
3.9 KiB
Python
125 lines
3.9 KiB
Python
"""保活上下文（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 = "<html>ok</html>"
|
|
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_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 == {}
|