"""会话层测试：cookie 预热、失败升级、浏览器兜底降级全部用 httpx.MockTransport 拦截，不触达真实站点。 """ from __future__ import annotations import httpx import pytest from app.scraping.core import site from app.shared.config import Settings from app.shared.errors import ( ItemNotFoundError, OffIchibaRedirectError, UpstreamBlockedError, UpstreamRequestError, ) from app.scraping.parsers.subsites import build_item_page_validator from app.scraping.services.browser_fallback import BrowserVisit from app.scraping.services.site_session import SiteSession GOOD_PAGE = '' BLOCK_PAGE = "Access Denied. Reference #18.abc" TARGET = "https://search.rakuten.co.jp/search/mall/x/" class FakeBrowser: """可控的浏览器兜底替身""" def __init__(self, visit: BrowserVisit | None = None): self.visit_result = visit self.calls: list[tuple[str, bool]] = [] self.enabled = True self.unavailable_reason = None if visit else "playwright is not installed" self.ready = visit is not None async def visit(self, url: str, *, mobile: bool) -> BrowserVisit | None: self.calls.append((url, mobile)) return self.visit_result async def close(self) -> None: pass def make_settings(**overrides) -> Settings: base = { "http_max_attempts": 3, "request_timeout_seconds": 5.0, "max_site_concurrency": 4, "browser_fallback_enabled": True, } base.update(overrides) return Settings(**base) async def build_session(handler, *, browser=None, settings=None) -> SiteSession: """构建 SiteSession 并把两条通道的 client 换成 MockTransport 版本""" session = SiteSession(settings or make_settings(), browser or FakeBrowser()) await session.start() for profile in session._profiles.values(): await profile.client.aclose() profile.client = httpx.AsyncClient( headers=site.default_headers(mobile=profile.mobile), transport=httpx.MockTransport(handler), follow_redirects=True, ) return session async def test_warmup_happens_before_first_fetch_and_is_reused(): seen: list[str] = [] def handler(request: httpx.Request) -> httpx.Response: seen.append(str(request.url)) headers = {} if request.url.host == "www.rakuten.co.jp": headers["set-cookie"] = "ak_bmsc=abc; Domain=.rakuten.co.jp; Path=/" return httpx.Response(200, text=GOOD_PAGE, headers=headers) session = await build_session(handler) try: await session.fetch_html(TARGET, mobile=False) await session.fetch_html(TARGET, mobile=False) finally: await session.close() # 首页预热只做一次，第二次抓取直接复用 cookie assert seen.count("https://www.rakuten.co.jp/") == 1 assert seen.count(TARGET) == 2 async def test_missing_state_marker_is_treated_as_blocked(): def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(200, text="no state here") session = await build_session(handler) try: with pytest.raises(UpstreamBlockedError): await session.fetch_html(TARGET, mobile=False) finally: await session.close() async def test_retry_recovers_when_a_later_attempt_succeeds(): attempts = {"n": 0} def handler(request: httpx.Request) -> httpx.Response: if request.url.host == "www.rakuten.co.jp": return httpx.Response(200, text="home") attempts["n"] += 1 if attempts["n"] == 1: return httpx.Response(200, text=BLOCK_PAGE) return httpx.Response(200, text=GOOD_PAGE) session = await build_session(handler) try: assert await session.fetch_html(TARGET, mobile=False) == GOOD_PAGE finally: await session.close() assert attempts["n"] == 2 async def test_browser_fallback_supplies_cookies_and_page_on_persistent_block(): """浏览器已经取到页面时应直接采用，不再多打一次 HTTP""" browser = FakeBrowser( BrowserVisit( cookies=[{"name": "bm_sv", "value": "xyz", "domain": ".rakuten.co.jp", "path": "/"}], html=GOOD_PAGE, ) ) def handler(request: httpx.Request) -> httpx.Response: if request.url.host == "www.rakuten.co.jp": return httpx.Response(200, text="home") return httpx.Response(200, text=BLOCK_PAGE) session = await build_session(handler, browser=browser) try: assert await session.fetch_html(TARGET, mobile=False) == GOOD_PAGE finally: cookie_names = {cookie.name for cookie in session._profiles["pc"].client.cookies.jar} await session.close() assert browser.calls == [(TARGET, False)] assert "bm_sv" in cookie_names # cookie 已回灌，后续请求可复用 async def test_missing_playwright_degrades_to_blocked_error_not_crash(): browser = FakeBrowser(None) # 模拟未安装 playwright def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(200, text=BLOCK_PAGE) session = await build_session(handler, browser=browser) try: with pytest.raises(UpstreamBlockedError): await session.fetch_html(TARGET, mobile=False) finally: await session.close() assert browser.calls # 尝试过兜底 async def test_404_fails_fast_without_retrying(): attempts = {"n": 0} def handler(request: httpx.Request) -> httpx.Response: if request.url.host == "www.rakuten.co.jp": return httpx.Response(200, text="home") attempts["n"] += 1 return httpx.Response(404, text="not found") session = await build_session(handler) try: with pytest.raises(ItemNotFoundError): await session.fetch_html("https://item.rakuten.co.jp/s/c/", mobile=True) finally: await session.close() assert attempts["n"] == 1 async def test_unsupported_subsite_redirect_fails_fast_without_retrying(): """跳到未登记的乐天子站时重试没有意义，应立刻以专门的错误返回""" attempts = {"n": 0} requested = "https://item.rakuten.co.jp/somewhere/1/" def handler(request: httpx.Request) -> httpx.Response: if request.url.host == "www.rakuten.co.jp": return httpx.Response(200, text="home") if request.url.host == "item.rakuten.co.jp": attempts["n"] += 1 return httpx.Response(302, headers={"location": "https://unknown.rakuten.co.jp/x/1/"}) return httpx.Response(200, text="unknown site") session = await build_session(handler) try: with pytest.raises(OffIchibaRedirectError) as exc: await session.fetch( requested, mobile=True, validator=build_item_page_validator(requested) ) finally: await session.close() assert attempts["n"] == 1 assert exc.value.final_url == "https://unknown.rakuten.co.jp/x/1/" async def test_known_subsite_redirect_is_accepted_and_reports_final_url(): """已登记的子站应正常通过校验，并把落地地址回报给调用方用于分派""" requested = "https://item.rakuten.co.jp/biccamera/1/" landed = "https://biccamera.rakuten.co.jp/item/1/" def handler(request: httpx.Request) -> httpx.Response: if request.url.host == "www.rakuten.co.jp": return httpx.Response(200, text="home") if request.url.host == "item.rakuten.co.jp": return httpx.Response(302, headers={"location": landed}) return httpx.Response(200, text='') session = await build_session(handler) try: page = await session.fetch( requested, mobile=True, validator=build_item_page_validator(requested) ) finally: await session.close() assert page.url == landed assert "__NUXT__" in page.html async def test_upstream_5xx_surfaces_as_request_error(): def handler(request: httpx.Request) -> httpx.Response: if request.url.host == "www.rakuten.co.jp": return httpx.Response(200, text="home") return httpx.Response(503, text="service unavailable") session = await build_session(handler) try: with pytest.raises(UpstreamRequestError): await session.fetch_html(TARGET, mobile=False) finally: await session.close() async def test_search_and_detail_use_separate_cookie_jars(): """PC 与手机两条通道的 cookie 不能互相污染""" def handler(request: httpx.Request) -> httpx.Response: mobile = request.headers.get("sec-ch-ua-mobile") == "?1" headers = {} if request.url.host == "www.rakuten.co.jp": headers["set-cookie"] = f"ak_bmsc={'sp' if mobile else 'pc'}; Domain=.rakuten.co.jp; Path=/" return httpx.Response(200, text=GOOD_PAGE, headers=headers) session = await build_session(handler) try: await session.fetch_html(TARGET, mobile=False) await session.fetch_html("https://item.rakuten.co.jp/s/c/", mobile=True) pc_cookie = session._profiles["pc"].client.cookies.get("ak_bmsc") sp_cookie = session._profiles["sp"].client.cookies.get("ak_bmsc") finally: await session.close() assert pc_cookie == "pc" assert sp_cookie == "sp" async def test_profile_status_reports_warmup_state(): def handler(request: httpx.Request) -> httpx.Response: headers = {"set-cookie": "ak_bmsc=abc; Domain=.rakuten.co.jp; Path=/"} return httpx.Response(200, text=GOOD_PAGE, headers=headers) session = await build_session(handler) try: assert session.profile_status()["pc"]["warmed"] is False await session.fetch_html(TARGET, mobile=False) status = session.profile_status() assert status["pc"]["warmed"] is True assert "ak_bmsc" in status["pc"]["cookies"] assert status["sp"]["warmed"] is False finally: await session.close()