識別碼: ABW-123
' '')) def client(root, source, **kwargs): return client_for(root, source, transport=httpx.MockTransport(respond), **kwargs) with patch("peach.scraping_access.client_for", side_effect=client): check = w_scraping_check(SimpleNamespace(follow_secrets_root=self.root), {"source": "javbus"}) self.assertTrue(check["results"][0]["ok"]) transport = SourceTransport(self.root) self.addCleanup(transport.close) record = JavBusSource().query("ABW-123", session=Session(transport)) self.assertEqual(record.code, "ABW-123") self.assertEqual(record.cover_urls, ("https://www.javbus.com/imgs/cover/test.jpg",)) self.assertEqual(seen, ["/", "/ABW-123"]) def test_javbus_cover_language_overrides_caller_headers_without_changing_other_sources(self): from peach.jav_cover_fetch import _fetch for source in ("javbus", "javdb"): save(self.root, source, {"network": "direct"}) seen = [] def respond(request): seen.append(request) return httpx.Response(200, content=b"cover") def client(root, source, **kwargs): return client_for(root, source, transport=httpx.MockTransport(respond), **kwargs) transport = SourceTransport(self.root) self.addCleanup(transport.close) with patch("peach.scraping_access.client_for", side_effect=client): self.assertEqual(_fetch(transport, "https://www.javbus.com/imgs/cover/test.jpg", referer="https://www.javbus.com/ABW-123", limit=1024, ranged=True), b"cover") transport(HttpRequest("GET", "https://www.javbus.com/ABW-123", {"accept-language": "en-US", "X-Test": "preserved"}), 1, 1024) self.assertEqual(_fetch(transport, "https://javdb.com/test.jpg", referer="https://javdb.com/", limit=1024), b"cover") self.assertEqual([request.headers["Accept-Language"] for request in seen], ["zh-CN,zh;q=0.9", "zh-CN,zh;q=0.9", "ja,en;q=0.9"]) self.assertEqual(seen[0].headers["Range"], "bytes=0-65535") self.assertEqual(seen[0].headers["Referer"], "https://www.javbus.com/ABW-123") self.assertEqual(seen[1].headers.get_list("Accept-Language"), ["zh-CN,zh;q=0.9"]) self.assertEqual(seen[1].headers["X-Test"], "preserved") def test_browser_obeys_global_direct_proxy_and_environment_modes(self): transport = SourceTransport(self.root) self.addCleanup(transport.close) for mode, direct, proxy in (("direct", True, ""), ("proxy", False, "http://127.0.0.1:7890"), ("environment", False, "")): with self.subTest(mode=mode): peach_proxy.save(self.root, {"mode": mode, "proxy": "http://127.0.0.1:7890"}) with patch("peach.browser_transport.shared") as shared: transport._browser_for("javten") self.assertEqual(shared.call_args.kwargs["direct"], direct) self.assertEqual(shared.call_args.kwargs["proxy"], proxy) def test_switching_networks_releases_the_previous_exit_cooldown(self): from peach.scraping_access import pause_source, paused_until pause_source(self.root, 'airav', refused=True) self.assertGreater(paused_until(self.root, 'airav'), time.time()) save(self.root, 'airav', {'network': 'direct'}) self.assertEqual(paused_until(self.root, 'airav'), 0) def test_redirect_selects_destination_policy_and_strips_secrets(self): transport = SourceTransport(self.root) with patch.object(transport, "_request", side_effect=[ HttpResponse(302, {"location": "https://pics.dmm.co.jp/x"}, b""), HttpResponse(200, {}, b"image"), ]) as request: result = transport(HttpRequest("GET", "https://r18.dev/x", {"Cookie": "secret", "Authorization": "secret"}), 1, 20) self.assertEqual(result.body, b"image") self.assertEqual(request.call_args_list[1].args[0].headers, {}) self.assertEqual(source_for("https://evil-dmm.co.jp/x"), None) self.assertEqual(source_for("https://awsimgsrc.dmm.com/x"), "dmm") def test_connections_do_not_claim_session_validity_or_echo_exception_urls(self): contract = SimpleNamespace(follow_secrets_root=self.root) with patch("peach.source_checks.SourceTransport") as factory: factory.return_value.side_effect = httpx.ConnectError("https://secret:password@example.org/") result = w_scraping_check(contract, {"source": "fc2cmadb"}) self.assertFalse(result["session_verified"]) self.assertNotIn("password", json.dumps(result)) self.assertFalse(result["results"][0]["ok"]) def test_the_check_tells_a_login_wall_apart_from_an_unreachable_source(self): """页面上要分得开三种结果，因为用户的下一步各不相同。 登录墙不一定回 403。跟完重定向落在登录页的那次响应是 200，只看状态码的判据 会把一次「其实什么都没取到」报成可连接。反过来，只剩状态码可看的 403 也不许 报成登录墙：`docs/SOURCING.md` 记着那些 403 多半是出口 IP 被封，去官网登录 白做一场，正确动作是等或换出口。措辞与元数据来源同源，页面不另写一份。 """ contract = SimpleNamespace(follow_secrets_root=self.root) login = "https://fc2cmadb.com/" cases = ((HttpResponse(403, {}, b"", login), "auth", False, "出口 IP", "请在官网"), (HttpResponse(200, {}, b"login", "https://fc2cmadb.com/users/sign_in"), "auth", False, "请在官网完成登录后重试", "出口 IP"), (HttpResponse(503, {}, b"", login), "unavailable", False, "稍后重试", "登录"), (HttpResponse(200, {}, b"ok", login), "ok", True, "", "登录")) for response, kind, ok, says, avoids in cases: with self.subTest(status=response.status, url=response.url), \ patch("peach.source_checks.SourceTransport") as factory: factory.return_value.return_value = response result = w_scraping_check(contract, {"source": "fc2cmadb"})["results"][0] self.assertEqual((result["kind"], result["ok"]), (kind, ok)) self.assertIn(says, result.get("message", "")) self.assertNotIn(avoids, result.get("message", "")) def test_transport_ownership_is_explicit(self): for owned in (True, False): client = httpx.Client() transport = HttpxTransport(client, owns_client=owned) transport.close() self.assertEqual(client.is_closed, owned) client.close() def test_connection_checks_classify_failures_and_store_only_diagnostics(self): from peach.scraping_access import source_checks contract = SimpleNamespace(follow_secrets_root=self.root) cases = [ (httpx.ProxyError("http://user:secret@proxy.invalid"), "proxy"), (httpx.ReadTimeout("secret"), "timeout"), (HttpResponse(429, {}, b"", "https://airav.io/"), "rate_limited"), (HttpResponse(200, {}, b"申し訳ございません。お住まいの地域からはご利用になれません。
"), } for source, (url, page) in pages.items(): with self.subTest(source=source): cooldown = self.root / f"scraping-{source}.cooldown.json" waits = [] for _ in range(2): transport = SourceTransport(self.root) transport.transports[source] = lambda *args, page=page: HttpResponse( 200, {}, page.encode("utf-8")) with self.assertRaises(SourceRefused) as refused: transport(HttpRequest("GET", url, {}), 1, 100000) self.assertEqual(refused.exception.reason, "geo_restricted") self.assertIn("换一个它接受的地区的出口", str(refused.exception)) record = json.loads(cooldown.read_text(encoding="utf-8")) waits.append(round(record["until"] - time.time())) cooldown.write_text(json.dumps({"until": 0, "blocks": record["blocks"]}), encoding="utf-8") self.assertEqual(waits, [FIRST_BLOCKED_PAUSE, FIRST_BLOCKED_PAUSE]) with self.assertRaises(SourcePaused): cooldown.write_text(json.dumps({"until": time.time() + 60, "blocks": 0}), encoding="utf-8") SourceTransport(self.root)(HttpRequest("GET", url, {}), 1, 100) def test_a_ban_page_doubles_like_a_refusal_and_ordinary_pages_pass(self): """封禁说明页按 403 那一档翻倍；正文里没有这几句话的正常页原样交回。""" from peach.scraping_access import FIRST_BLOCKED_PAUSE, SourceRefused cooldown = self.root / "scraping-javbus.cooldown.json" request = HttpRequest("GET", "https://www.javbus.com/ABW-358", {}) waits = [] for _ in range(2): transport = SourceTransport(self.root) transport.transports["javbus"] = lambda *args: HttpResponse( 200, {}, b"The owner of this website has banned your access.") with self.assertRaises(SourceRefused) as refused: transport(request, 1, 100000) self.assertEqual(refused.exception.reason, "ip_banned") record = json.loads(cooldown.read_text(encoding="utf-8")) waits.append(round(record["until"] - time.time())) cooldown.write_text(json.dumps({"until": 0, "blocks": record["blocks"]}), encoding="utf-8") self.assertEqual(waits, [FIRST_BLOCKED_PAUSE, FIRST_BLOCKED_PAUSE * 2]) opened = SourceTransport(self.root) opened.transports["javbus"] = lambda *args: HttpResponse(200, {}, b"ABW-358 page") self.assertEqual(opened(request, 1, 100000).status, 200) self.assertFalse(cooldown.exists()) def test_a_connection_check_names_the_region_refusal(self): """来源检查撞上地区限制页：细档报 `geo_restricted`，话说到「换出口」，不是「稍后重试」。""" from peach.scraping_access import source_checks transport = SourceTransport(self.root) transport.transports["javdb"] = lambda *args: HttpResponse( 200, {}, "由於版權限制，本站禁止了你的網路所在國家的訪問。".encode("utf-8")) with patch("peach.source_checks.SourceTransport", return_value=transport): result = w_scraping_check(SimpleNamespace(follow_secrets_root=self.root), {"source": "javdb"})["results"][0] self.assertFalse(result["ok"]) self.assertEqual((result["kind"], result["reason"]), ("auth", "geo_restricted")) self.assertIn("换一个它接受的地区的出口", result["message"]) self.assertEqual(source_checks(self.root, "javdb")["connection"]["reason"], "geo_restricted") def test_community_sources_carry_the_pasted_cookie_on_public_requests(self): """JavBus 的年龄门和 javdb 的登录墙靠用户贴的 Cookie 过；别的来源公开采集不带会话。""" self.assertTrue(describe(self.root, "javdb")["accepts_cookie"]) save(self.root, "javbus", {"cookie": "existmag=all; age=verified"}) transport = SourceTransport(self.root) self.addCleanup(transport.close) with patch("peach.scraping_access.client_for") as factory: fake = factory.return_value fake.stream.return_value.__enter__.return_value.iter_bytes.return_value = [b"ok"] transport(HttpRequest("GET", "https://www.javbus.com/MIDE-594", {}), 1, 10) transport(HttpRequest("GET", "https://pics.dmm.co.jp/x", {}), 1, 10) self.assertEqual([call.args[1] for call in factory.call_args_list], ["javbus", "dmm"]) self.assertEqual([call.kwargs["session"] for call in factory.call_args_list], [True, False]) with httpx.Client(cookies=cookie_jar(values_for(self.root, "javbus"), "javbus")) as client: self.assertIn("age=verified", client.build_request("GET", "https://www.javbus.com/MIDE-594").headers["cookie"]) self.assertNotIn("cookie", client.build_request("GET", "https://pics.dmm.co.jp/x").headers) def test_the_fc2_mirror_carries_the_pasted_cookie_too(self): """女優那一栏在浏览器里只对登录用户显示，采集按登录用户看到的那一页取。""" save(self.root, "fc2cmadb", {"cookie": "peach_session=abc"}) transport = SourceTransport(self.root) self.addCleanup(transport.close) with patch("peach.scraping_access.client_for") as factory: fake = factory.return_value fake.stream.return_value.__enter__.return_value.iter_bytes.return_value = [b"ok"] transport(HttpRequest("GET", "https://fc2cmadb.com/articles/4030617", {}), 1, 10) self.assertEqual(factory.call_args.args[1], "fc2cmadb") self.assertTrue(factory.call_args.kwargs["session"]) with httpx.Client(cookies=cookie_jar(values_for(self.root, "fc2cmadb"), "fc2cmadb")) as client: self.assertIn("peach_session=abc", client.build_request( "GET", "https://fc2cmadb.com/articles/4030617").headers["cookie"]) def test_the_cloudflare_sources_carry_the_pasted_cookie_under_the_shared_browser_user_agent(self): """`cf_clearance` 绑着解题那台浏览器的 UA：整站 UA 与用户的 Chrome 一致，Cookie 由用户贴，卡上不另收 UA。""" from peach.user_agent import USER_AGENT shown = save(self.root, "fc2ppvdb", {"cookie": "cf_clearance=abc"}) self.assertEqual(set(shown), {"source", "label", "login", "accepts_cookie", "network", "cookie_saved", "browser", "cooldown_until"}) self.assertTrue(shown["cookie_saved"]) self.assertFalse(shown["browser"], "没有浏览器的机器上这张卡照旧收 Cookie") for name in ("fc2ppvdb", "javten"): with client_for(self.root, name) as client: self.assertEqual(client.headers["user-agent"], USER_AGENT, name) transport = SourceTransport(self.root) self.addCleanup(transport.close) with patch("peach.scraping_access.client_for") as factory: fake = factory.return_value fake.stream.return_value.__enter__.return_value.iter_bytes.return_value = [b"ok"] transport(HttpRequest("GET", "https://fc2ppv-db.com/ja/videos/4898837", {}), 1, 10) transport(HttpRequest("GET", "https://javten.com/search?kw=4898837", {}), 1, 10) self.assertEqual([(call.args[1], call.kwargs["session"]) for call in factory.call_args_list], [("fc2ppvdb", True), ("javten", True)]) def test_the_cloudflare_sources_go_through_the_local_browser_when_there_is_one(self): """有浏览器的机器上两站的请求在浏览器页面里发（ADR-0065）：连接方式换成启动参数，Cookie 不再是必需的。""" save(self.root, "fc2ppvdb", {"network": "direct"}) peach_proxy.save(self.root, {"mode": "proxy", "proxy": "socks5://127.0.0.1:7890"}) fake = lambda request, timeout, max_bytes: HttpResponse(200, {}, b"page", request.url) fake.close = lambda: self.fail("浏览器是进程共用的，SourceTransport.close 不许关它") transport = SourceTransport(self.root) with patch("peach.browser_transport.shared", return_value=fake) as shared: transport(HttpRequest("GET", "https://fc2ppv-db.com/ja/videos/1", {}), 1, 100) transport(HttpRequest("GET", "https://javten.com/search?kw=1", {}), 1, 100) self.assertEqual([(call.kwargs["direct"], call.kwargs["proxy"]) for call in shared.call_args_list], [(True, ""), (False, "socks5://127.0.0.1:7890")]) self.assertEqual(shared.call_args.args, (self.root / "browser",)) self.assertEqual(shared.call_args_list[0].kwargs["gates"], {"fc2ppv-db.com": {"path": "/age-verify", "button": r"18|はい|入場|同意|以上|Enter|Yes"}}, "FC2PPV-DB 的年龄门交给浏览器传输去点") self.assertEqual(shared.call_args_list[1].kwargs["gates"], {}, "JAVten 没有门") transport.close() self.assertFalse(describe(self.root, "javten")["browser"]) with patch("peach.browser_transport.find_browser", return_value="C:/edge.exe"): self.assertTrue(describe(self.root, "javten")["browser"]) with patch("peach.scraping_access.client_for") as factory, \ patch("peach.browser_transport.shared", return_value=None): factory.return_value.stream.return_value.__enter__.return_value.iter_bytes.return_value = [b"ok"] SourceTransport(self.root)(HttpRequest("GET", "https://javten.com/", {}), 1, 10) self.assertEqual(factory.call_args.kwargs["session"], True, "没有浏览器就退回带 Cookie 的直连") def test_an_unsolved_challenge_pauses_the_source_and_a_lost_browser_reads_as_a_connection_failure(self): from peach import browser_transport from peach.scraping_access import SourcePaused cooldown = self.root / "scraping-fc2ppvdb.cooldown.json" request = HttpRequest("GET", "https://fc2ppv-db.com/ja/videos/1", {}) def unsolved(*args): raise browser_transport.ChallengeUnsolved("fc2ppv-db.com 的人机验证在 150 秒内没有通过") transport = SourceTransport(self.root) transport.transports["fc2ppvdb"] = unsolved with self.assertRaises(SourcePaused) as caught: transport(request, 1, 100) self.assertIn("点一下验证", str(caught.exception)) self.assertTrue(cooldown.exists()) cooldown.unlink() def lost(*args): raise browser_transport.BrowserUnavailable("浏览器请求未取得") transport.transports["fc2ppvdb"] = lost with self.assertRaises(httpx.TransportError): transport(request, 1, 100) self.assertFalse(cooldown.exists(), "浏览器断了是连接问题，不记来源的账") def test_a_pause_earned_on_the_direct_path_does_not_hold_back_the_browser(self): """直连 403 攒下的冷却不压到浏览器那条路上：换了路就作废，浏览器撞验证没过从第一档重新数。""" from peach import browser_transport from peach.scraping_access import FIRST_BLOCKED_PAUSE, SourcePaused class FakeBrowser(browser_transport.BrowserTransport): def __init__(self, respond): self.respond = respond def __call__(self, request, timeout, max_bytes): return self.respond(request) cooldown = self.root / "scraping-fc2ppvdb.cooldown.json" request = HttpRequest("GET", "https://fc2ppv-db.com/ja/videos/1", {}) refused = SourceTransport(self.root) refused.transports["fc2ppvdb"] = lambda *args: HttpResponse(403, {}, b"") for _ in range(3): with self.assertRaises(SourcePaused): refused(request, 1, 100) record = json.loads(cooldown.read_text(encoding="utf-8")) cooldown.write_text(json.dumps({**record, "until": 0}), encoding="utf-8") cooldown.write_text(json.dumps({**record, "until": time.time() + 3600}), encoding="utf-8") self.assertEqual((record["blocks"], record["via"]), (3, "http")) opened = SourceTransport(self.root) opened.transports["fc2ppvdb"] = FakeBrowser(lambda request: HttpResponse(200, {}, b"page", request.url)) self.assertEqual(opened(request, 1, 100).status, 200, "直连那一小时的冷却不拦浏览器") self.assertFalse(cooldown.exists()) def unsolved(request): raise browser_transport.ChallengeUnsolved("fc2ppv-db.com 的人机验证在 160 秒内没有通过") cooldown.write_text(json.dumps({"until": 0, "blocks": 3, "via": "http"}), encoding="utf-8") challenged = SourceTransport(self.root) challenged.transports["fc2ppvdb"] = FakeBrowser(unsolved) with self.assertRaises(SourcePaused): challenged(request, 1, 100) record = json.loads(cooldown.read_text(encoding="utf-8")) self.assertEqual((round(record["until"] - time.time()), record["blocks"], record["via"]), (FIRST_BLOCKED_PAUSE, 1, "browser"), "浏览器那条路的第一次从第一档起") with self.assertRaises(SourcePaused): challenged(request, 1, 100) # 浏览器那条路写下的冷却，回到直连时同样作废：直连会被按自己的账重新判。 direct = SourceTransport(self.root) direct.transports["fc2ppvdb"] = lambda *args: HttpResponse(200, {}, b"page") self.assertEqual(direct(request, 1, 100).status, 200) self.assertFalse(cooldown.exists()) def test_a_new_cookie_lifts_the_pause_the_old_one_earned(self): """过期的 `cf_clearance` 撞出 403 后整站冷却；用户换了新 Cookie 就该立刻再试，不等旧账到期。""" from peach.scraping_access import SourcePaused cooldown = self.root / "scraping-javten.cooldown.json" request = HttpRequest("GET", "https://javten.com/search?kw=4898837", {}) refused = SourceTransport(self.root) refused.transports["javten"] = lambda *args: HttpResponse(403, {}, b"") with self.assertRaises(SourcePaused) as caught: refused(request, 1, 100) self.assertIn("更新它的 Cookie", str(caught.exception)) self.assertTrue(cooldown.exists()) save(self.root, "javten", {"network": "peach"}) self.assertTrue(cooldown.exists(), "只改连接方式不算换了 Cookie") save(self.root, "javten", {"cookie": "cf_clearance=new"}) self.assertFalse(cooldown.exists()) def test_full_quality_bytes_are_installed_and_cache_avoids_download(self): from peach.web_scraping import _fetch_cover from peach.jav_cover_fetch import Candidate output = io.BytesIO() Image.new("RGB", (1600, 1000)).save(output, "JPEG") original = output.getvalue() contract = SimpleNamespace(cover_root=self.root / "covers", candidate_root=self.root, follow_secrets_root=self.root / "secrets", follow_sources_root=self.root) with patch("peach.jav_cover_fetch.best_cover", return_value=( Candidate("dmm", "https://pics.dmm.co.jp/fixture.jpg"), (1600, 1000), original, )) as fetch: result = _fetch_cover(contract, "ABW-232") cached = _fetch_cover(contract, "ABW-232") self.assertTrue(result["ok"]) self.assertTrue(cached["ok"]) self.assertEqual(fetch.call_count, 1) self.assertEqual((contract.cover_root / "ABW-232.jpg").read_bytes(), original) def test_smaller_candidate_cannot_overwrite_a_full_quality_cover(self): from peach.web_scraping import _fetch_cover from peach.jav_cover_fetch import Candidate target = self.root / "ABW-232.jpg" Image.new("RGB", (2000, 1400)).save(target, "JPEG") original = target.read_bytes() contract = SimpleNamespace(cover_root=self.root, candidate_root=self.root, follow_secrets_root=self.root / "secrets", follow_sources_root=self.root) with patch("peach.jav_cover_fetch.best_cover", return_value=( Candidate("dmm", "https://pics.dmm.co.jp/fixture.jpg"), (800, 600), b"small", )): result = _fetch_cover(contract, "ABW-232") self.assertTrue(result["ok"]) self.assertEqual(result["reason"], "kept_existing") self.assertIn("2000 × 1400", result["result"]) self.assertIn("800 × 600", result["result"]) self.assertEqual(target.read_bytes(), original) def test_equal_size_product_image_replaces_a_preview(self): from peach.web_scraping import _fetch_cover from peach.jav_cover_fetch import candidate_for preview = ("https://image.mgstage.com/images/luxutv/sp/259luxu/1509/" "popsample2_sp-259luxu-1509.jpg") product = ("https://image.mgstage.com/images/luxutv/259luxu/1509/" "pb_e_259luxu-1509.jpg") target = self.root / "259LUXU-1509.jpg" old = io.BytesIO() Image.new("RGB", (840, 472), "black").save(old, "JPEG") target.write_bytes(old.getvalue()) target.with_suffix(".scraping.json").write_text(json.dumps({ "checked_at": 0, "source_url": preview, "raw_sha256": "old", "width": 840, "height": 472, }), encoding="utf-8") new = io.BytesIO() Image.new("RGB", (840, 472), "white").save(new, "JPEG") contract = SimpleNamespace(cover_root=self.root, candidate_root=self.root, follow_secrets_root=self.root / "secrets", follow_sources_root=self.root) with patch("peach.jav_cover_fetch.best_cover", return_value=( candidate_for(product), (840, 472), new.getvalue(), )), patch("peach.cover_artwork._sidecars", return_value={}): result = _fetch_cover(contract, "259LUXU-1509") self.assertTrue(result["ok"]) self.assertNotEqual(result.get("reason"), "kept_existing") self.assertEqual(target.read_bytes(), new.getvalue()) def test_cover_failure_explains_observed_cause_without_exposing_urls(self): from peach.web_scraping import _fetch_cover from peach.jav_cover_fetch import Unavailable contract = SimpleNamespace(cover_root=self.root, candidate_root=self.root, follow_secrets_root=self.root / "secrets", follow_sources_root=self.root) cases = [(403, "access_denied"), (503, "source_error"), (404, "no_candidate"), (None, "network")] for status, reason in cases: def fetch(transport, *args, **kwargs): try: transport(HttpRequest("GET", "https://example.test/private?token=secret", {}), 1, 100) except httpx.TransportError: pass raise Unavailable("所有渠道都没有候选") with self.subTest(reason=reason), patch("peach.jav_cover_fetch.best_cover", side_effect=fetch), \ patch("peach.web_scraping.SourceTransport") as factory: if status is None: factory.return_value.side_effect = httpx.ConnectError("private?token=secret") else: factory.return_value.return_value = HttpResponse(status, {}, b"") result = _fetch_cover(contract, "ABW-232") self.assertFalse(result["ok"]) self.assertEqual(result["reason"], reason) self.assertNotIn("secret", json.dumps(result)) self.assertNotIn("可能需要代理", result["error"]) def test_unreachable_dmm_image_hosts_point_at_the_connection_setting(self): """中国移动宽带直连 DMM 图片主机大多在握手后被断开：封面比较和连接检查都要把人 指到 DMM / FANZA 的连接方式上，而不是说官方没有图、或让人去官网登录。""" from peach.web_scraping import MOBILE_BROADBAND_HINT, _fetch_cover from peach.jav_cover_fetch import HOSTS_UNREACHABLE, CoverConnectError contract = SimpleNamespace(cover_root=self.root, candidate_root=self.root, follow_secrets_root=self.root / "secrets", follow_sources_root=self.root) with patch("peach.jav_cover_fetch.best_cover", side_effect=CoverConnectError(HOSTS_UNREACHABLE)), \ patch("peach.web_scraping.SourceTransport"): result = _fetch_cover(contract, "ABW-232") self.assertEqual(result["reason"], "network") self.assertIn(MOBILE_BROADBAND_HINT, result["error"]) with patch("peach.source_checks.SourceTransport") as factory: factory.return_value.side_effect = httpx.ConnectError("EOF occurred in violation of protocol") checks = {item["label"]: item for item in w_scraping_check(SimpleNamespace(follow_secrets_root=self.root), {"source": "dmm"})["results"]} self.assertIn(MOBILE_BROADBAND_HINT, checks["高清图片 CDN"]["message"]) self.assertNotIn(MOBILE_BROADBAND_HINT, checks["来源页面"]["message"]) class BrowserFallbackTests(unittest.TestCase): """登记了 `browser_fallback` 的来源（minnano-av）：HTTP 客户端被拦时由本机浏览器验证一页， 取到就固定走浏览器；浏览器也没取到才照 HTTP 客户端被拦那样冷却。""" SEARCH = "https://www.minnano-av.com/search_result.php?search_scope=actress&search_word=x" def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name).resolve() self.cooldown = self.root / "scraping-minnano-av.cooldown.json" self.request = HttpRequest("GET", self.SEARCH, {}) finder = patch("peach.browser_transport.find_browser", return_value=None) finder.start() self.addCleanup(finder.stop) def browser(self, respond): from peach import browser_transport class FakeBrowser(browser_transport.BrowserTransport): def __init__(self): self.calls = [] def __call__(self, request, timeout, max_bytes): self.calls.append(request.url) return respond(request) return FakeBrowser() def refused_transport(self, response=None): transport = SourceTransport(self.root) http_calls = [] def http(request, *_args): http_calls.append(request.url) return response or HttpResponse(403, {}, b"") transport.transports["minnano-av"] = http return transport, http_calls def record(self) -> dict: return json.loads(self.cooldown.read_text(encoding="utf-8")) def test_a_refused_http_request_is_retried_in_the_browser_and_the_source_stays_there(self): from peach.scraping_access import fixed_to_browser page = HttpResponse(200, {}, b"