tests/test_follow_connector_base.py
Rama master · réplica de lectura del repositorio original.
"""连接器基类共享骨架的隔离测试。
「条件请求 → 304 短路 → 状态检查」由 `_BaseConnector._request()` 一处实现，
八个连接器的 fetch 开头不各写一份，所以这些规则也只需要在这里守住；
各站的解析差异仍由 `test_follow_sources.py` 覆盖。测试不联网，transport 全部注入。
"""
import unittest
import httpx
from peach.follow import FollowSourceError
from peach.follow_sources import (
CONNECTORS, KemonoConnector, _BaseConnector, display_thumb_url,
is_history_end_error, official_profile_handle,
)
from peach.follow_store import _OFFICIAL_IDENTITY_PROVIDERS
from peach.http import HttpResponse
def _transport(status=200, headers=None, body=b"", record=None):
def call(request, timeout, max_bytes):
if record is not None:
record.append(request)
return HttpResponse(status, headers or {}, body)
return call
class _Probe(_BaseConnector):
provider = "probe"
class SharedRequestSkeletonTests(unittest.TestCase):
def test_handshake_timeout_recovers_with_bounded_backoff(self):
calls, waits, progress = [], [], []
def transport(request, timeout, max_bytes):
calls.append(request)
if len(calls) < 5:
raise httpx.ConnectTimeout('_ssl.c:1064: The handshake operation timed out')
return HttpResponse(200, {}, b'{}')
connector = _Probe(transport=transport, sleeper=waits.append)
connector.progress = lambda **state: progress.append(state)
self.assertEqual(connector._get('https://example.test').status, 200)
self.assertEqual(waits, [1, 2, 4, 8])
self.assertEqual(progress[-1]['attempt'], 5)
def test_retries_exhaust_and_post_is_not_replayed(self):
for method, expected in [('GET', 5), ('POST', 1)]:
calls = []
def transport(*args, calls=calls):
calls.append(1)
raise httpx.ReadTimeout('temporary')
connector = _Probe(transport=transport, sleeper=lambda _: None)
with self.assertRaises(FollowSourceError):
connector._send(method, 'https://example.test', None, headers={}, base={})
self.assertEqual(len(calls), expected)
def test_access_denied_is_not_retried(self):
calls, waits = [], []
connector = _Probe(transport=_transport(status=403, record=calls), sleeper=waits.append)
with self.assertRaises(FollowSourceError):
connector._request('https://example.test', ref='r')
self.assertEqual(len(calls), 1)
self.assertEqual(waits, [])
def test_first_page_sends_conditional_headers(self):
seen = []
connector = _Probe(transport=_transport(body=b"{}", record=seen))
common, response = connector._request(
"https://example.test/a", ref="r", etag='"v0"',
last_modified="Wed, 13 Aug 2026 08:00:00 GMT")
self.assertIsNotNone(response)
self.assertEqual(seen[0].headers.get("If-None-Match"), '"v0"')
self.assertEqual(seen[0].headers.get("If-Modified-Since"),
"Wed, 13 Aug 2026 08:00:00 GMT")
self.assertEqual(common["ref"], "r")
self.assertEqual(common["request_url"], "https://example.test/a")
def test_paging_back_drops_conditional_headers(self):
"""往回翻页不能带条件请求头。
`If-None-Match` 存的是第一页的 etag，拿它去问第二页，站点很可能回 304，
表现就是「点了没反应」。这条规则过去在每个连接器里各写一遍，现在只有一处。"""
seen = []
connector = _Probe(transport=_transport(body=b"{}", record=seen))
connector._request("https://example.test/a?o=50", ref="r", etag='"v0"',
last_modified="Wed, 13 Aug 2026 08:00:00 GMT", page=1)
self.assertNotIn("If-None-Match", seen[0].headers)
self.assertNotIn("If-Modified-Since", seen[0].headers)
def test_not_modified_short_circuits_with_usable_common(self):
connector = _Probe(transport=_transport(status=304, headers={"ETag": '"v1"'}))
common, response = connector._request("https://example.test/a", ref="r")
self.assertIsNone(response, "304 必须短路，调用方据此直接返回 not_modified")
self.assertEqual(common["etag"], '"v1"')
self.assertEqual(common["provider"], "probe")
def test_request_url_can_be_masked_so_credentials_never_reach_evidence(self):
"""带凭据的真实 URL 绝不能落进证据档案。
rule34xxx 的 `api_key` 在查询串里，必须传脱敏版本；`_request` 请求用真实 URL，
`common` 里放脱敏 URL。"""
seen = []
connector = _Probe(transport=_transport(body=b"{}", record=seen))
common, _ = connector._request(
"https://example.test/a?api_key=secret", ref="r",
request_url="https://example.test/a")
self.assertIn("api_key=secret", seen[0].url, "请求本身仍要带真实凭据")
self.assertNotIn("secret", str(common["request_url"]))
def test_non_200_raises_before_the_caller_parses_anything(self):
waits = []
connector = _Probe(transport=_transport(status=500), sleeper=waits.append)
with self.assertRaises(FollowSourceError):
connector._request("https://example.test/a", ref="r")
self.assertEqual(waits, [1, 2, 4, 8])
class SharedSendContractTests(unittest.TestCase):
"""`_get` 与 `_post` 只在动词和请求体上不同，其余规则必须是同一份实现。
两份复制品已经开始分岔：`connector_headers=False`（跳到第三方主机时不带来源站
Cookie）当初只加进了 GET 那一份。那是安全语义，不该取决于用的是哪个动词。
"""
class _Blocked(_BaseConnector):
provider = "blocked"
blocked_reason = "站点有机器人验证"
def test_both_verbs_refuse_a_blocked_site(self):
connector = self._Blocked(transport=_transport())
for call in (lambda: connector._get("https://example.test/a"),
lambda: connector._post("https://example.test/a", b"x")):
with self.assertRaises(FollowSourceError):
call()
def test_both_verbs_reject_a_body_over_the_size_bound(self):
connector = _Probe(transport=_transport(body=b"x" * 40), max_bytes=8)
for call in (lambda: connector._get("https://example.test/a"),
lambda: connector._post("https://example.test/a", b"x")):
with self.assertRaises(FollowSourceError):
call()
def test_both_verbs_can_drop_the_source_site_headers(self):
"""跨站请求不带来源站的 Cookie，POST 也一样。"""
class _WithCookie(_BaseConnector):
provider = "withcookie"
def _headers(self):
return {**super()._headers(), "Cookie": "session=secret"}
seen = []
connector = _WithCookie(transport=_transport(body=b"{}", record=seen))
connector._get("https://third.test/a", connector_headers=False)
connector._post("https://third.test/a", b"x", connector_headers=False)
for request in seen:
self.assertNotIn("Cookie", request.headers)
connector._post("https://own.test/a", b"x")
self.assertEqual(seen[-1].headers.get("Cookie"), "session=secret")
def test_post_carries_the_body_and_get_does_not(self):
seen = []
connector = _Probe(transport=_transport(body=b"{}", record=seen))
connector._get("https://example.test/a")
connector._post("https://example.test/a", b"payload")
self.assertEqual([(row.method, row.body) for row in seen],
[("GET", None), ("POST", b"payload")])
def test_the_conditional_header_still_overrides_an_explicit_one(self):
"""`etag` 盖掉调用方显式给的同名头，合并顺序与抽取前一致。"""
seen = []
connector = _Probe(transport=_transport(body=b"{}", record=seen))
connector._get("https://example.test/a", headers={"If-None-Match": '"old"'},
etag='"new"')
self.assertEqual(seen[0].headers.get("If-None-Match"), '"new"')
class _Item:
"""已入库的一行里，读取时投影用得到的那几个字段。"""
def __init__(self, provider, *, thumb_url=None, media_url=None, item_id=1, metadata=None):
self.provider = provider
self.thumb_url = thumb_url
self.media_url = media_url
self.id = item_id
self.metadata = metadata or {}
class DisplayThumbUrlTests(unittest.TestCase):
"""「这个站的缩略图 URL 长什么样」属于连接器，不属于 Web 层。
rule34.xxx 的 preview→sample 改写、归档站的 `img.` 子域、归档站旧行的缩略图
推导，都归连接器。站点行为变了要改的是站点知识，而 Web 层根本不该知道
`api-cdn.rule34.xxx` 这种主机名。
"""
def test_rule34xxx_images_use_the_source_file_and_extension(self):
"""图片使用来源声明的原文件，视频使用来源缩略图。"""
thumb = ("https://api-cdn.rule34.xxx/thumbnails/1234/"
"thumbnail_" + "a" * 32 + ".jpg")
image = "https://api-cdn.rule34.xxx/images/1234/" + "a" * 32 + ".png"
self.assertEqual(display_thumb_url(_Item("rule34xxx", thumb_url=thumb,
media_url=image)), image)
self.assertEqual(display_thumb_url(_Item("rule34xxx", thumb_url=thumb,
media_url='https://api-cdn.rule34.xxx/images/128/' + 'c' * 32 + '.mp4')), thumb)
def test_a_rule34xxx_thumb_in_another_shape_is_left_alone(self):
"""认不出的形状原样返回，不猜一个 bucket 出来。"""
other = "https://api-cdn.rule34.xxx/images/1/x.jpg"
self.assertEqual(display_thumb_url(_Item("rule34xxx", thumb_url=other)), other)
def test_archive_thumbnails_use_the_img_subdomain(self):
"""2026-08-30 实测：主域 kemono.cr 回 302、pawchive.pw 回 404，img. 都回 200。"""
for provider, host in (("kemono", "kemono.cr"), ("pawchive", "pawchive.pw"),
("coomer", "coomer.st")):
recorded = f"https://{host}/thumbnail/data/a/b/c.jpg"
fixed = f"https://img.{host}/thumbnail/data/a/b/c.jpg"
self.assertEqual(display_thumb_url(_Item(provider, thumb_url=recorded)),
fixed)
# 已经是 img. 的不再叠加
self.assertEqual(display_thumb_url(_Item(provider, thumb_url=fixed)), fixed)
def test_an_archive_thumb_on_a_foreign_host_is_not_rewritten(self):
self.assertEqual(
display_thumb_url(_Item("kemono", thumb_url="https://example.test/a.jpg")),
"https://example.test/a.jpg")
def test_an_archive_row_without_a_thumb_derives_one_from_the_image_media(self):
"""封面修复前入库的旧行 thumb_url 为空，但 media_url 就是图片本身。
库里两种 `media_url` 形状都有：那次媒体主机修复之前拼的没有 `/data` 前缀，
之后拼的有。`/thumbnail/data` 里已经带了一个 `data`，所以拼之前要剥掉，
否则新形状的行会得到 `/thumbnail/data/data/...` 这种必然 404 的地址。
"""
for media in ("https://pawchive.pw/a/b/c.jpg",
"https://file.pawchive.pw/data/a/b/c.jpg"):
self.assertEqual(
display_thumb_url(_Item("pawchive", media_url=media)),
"https://img.pawchive.pw/thumbnail/data/a/b/c.jpg", media)
def test_a_video_only_archive_row_gets_no_derived_thumb(self):
"""视频没有可推导的缩略图，给了也是 404——卡片宁可显示干净的占位。"""
self.assertIsNone(display_thumb_url(_Item(
"kemono", media_url="https://kemono.cr/data/a/b/c.mp4")))
def test_a_provider_without_a_rule_uses_the_recorded_thumb(self):
for provider in ("rule34video", "f95zone", "nyaa"):
self.assertEqual(
display_thumb_url(_Item(provider, thumb_url="https://x.test/a.jpg")),
"https://x.test/a.jpg")
self.assertIsNone(display_thumb_url(_Item("rule34video")))
class HistoryEndRecognitionTests(unittest.TestCase):
"""哪些状态码代表「往回翻到尽头」只声明一次，翻页判定和旧行识别共用。
`record_history_end` 之前的版本把翻到尽头记成了 `error`，正文就是
`_check_status` 那句话。旧的识别代码是在 Web 层照站点名硬编码中文串比较，
新增一个可回填来源时没人会想到还要改那一处。
"""
def test_a_declared_status_is_recognised_and_others_are_not(self):
self.assertTrue(is_history_end_error("kemono", "kemono 返回 HTTP 400"))
self.assertTrue(is_history_end_error("coomer", "coomer 返回 HTTP 404"))
self.assertTrue(is_history_end_error("rule34paheal",
"rule34paheal 返回 HTTP 404"))
self.assertFalse(is_history_end_error("kemono", "kemono 返回 HTTP 500"))
self.assertFalse(is_history_end_error("rule34video",
"rule34video 返回 HTTP 400"))
def test_a_provider_that_never_pages_back_recognises_nothing(self):
for message in ("f95zone 返回 HTTP 404", "f95zone 返回 HTTP 400"):
self.assertFalse(is_history_end_error("f95zone", message))
self.assertFalse(is_history_end_error("nyaa", "nyaa 返回 HTTP 404"))
def test_it_matches_the_message_check_status_actually_writes(self):
"""判据必须对上真正落盘的那句话。
识别的是 `_check_status` 生成的文本。这两处一旦分家，回填到底的来源会
永远显示成红色错误行，而且不会有任何测试变红——所以这里现场生成一次。
"""
connector = KemonoConnector(provider="kemono",
transport=_transport(status=400))
with self.assertRaises(FollowSourceError) as raised:
connector._check_status(HttpResponse(400, {}, b""))
self.assertTrue(is_history_end_error("kemono", str(raised.exception)),
f"没认出来：{raised.exception}")
def test_an_unrelated_error_is_not_swallowed(self):
for message in ("kemono 请求失败：connect timeout",
"kemono 的帖子列表格式不符", ""):
self.assertFalse(is_history_end_error("kemono", message))
class PublicProbeApiTests(unittest.TestCase):
"""`follow_discovery` 与 `web_follow` 要的是探测能力，不该去摸私有方法。"""
def test_probe_returns_the_response_without_checking_status(self):
"""发现流程按状态码判断「作者页在不在」，404 是答案不是故障。"""
connector = _Probe(transport=_transport(status=404))
self.assertEqual(connector.probe("https://example.test/a").status, 404)
def test_fetch_json_parses_and_rejects_non_200(self):
connector = _Probe(transport=_transport(body=b'{"name": "x"}'))
self.assertEqual(connector.fetch_json("https://example.test/a"), {"name": "x"})
waits = []
failing = _Probe(transport=_transport(status=503, body=b"{}"), sleeper=waits.append)
with self.assertRaises(FollowSourceError):
failing.fetch_json("https://example.test/a")
self.assertEqual(waits, [1, 2, 4, 8])
def test_parse_json_rejects_a_non_json_body(self):
connector = _Probe(transport=_transport(body=b"<html/>"))
with self.assertRaises(FollowSourceError):
connector.parse_json(connector.probe("https://example.test/a"))
def test_every_connector_inherits_the_public_probe_api(self):
for provider, factory in CONNECTORS.items():
for name in ("probe", "fetch_json", "parse_json"):
self.assertTrue(
hasattr(factory, name) or hasattr(KemonoConnector, name),
f"{provider} 缺少公开的 {name}",
)
class NoPrivateReachIntoConnectorsTests(unittest.TestCase):
def test_other_modules_never_call_connector_private_methods(self):
"""连接器之外的模块只能用公开 API。
缺少「探测」这类公开入口时，调用方会直接摸 `_get`/`_check_status`/`_json`，
于是基类的任何改动都可能悄悄弄坏发现流程。这条门槛把它挡在源码层面。"""
import pathlib
import peach
root = pathlib.Path(peach.__file__).parent
offenders = []
for path in sorted(root.glob("*.py")):
if path.name == "follow_sources.py":
continue # 基类自己的实现文件
text = path.read_text(encoding="utf-8")
for private in ("connector._get(", "connector._check_status(",
"connector._json(", "connector._post("):
if private in text:
offenders.append(f"{path.name}: {private}")
self.assertEqual(
offenders, [],
"改用 probe()／fetch_json()／parse_json()；缺能力就在基类补公开方法",
)
class ProfileHandleTests(unittest.TestCase):
"""ref 里哪一截是作者本人的手柄，由各站自己的连接器说。
这些规则不写在 `web_follow` 的一串 if/elif 里：「ref 长什么形状」正是连接器
已经在解析、在校验的东西，同一份知识分在两层，改一处就会漂移。
"""
def test_an_official_channel_ref_is_the_author_handle(self):
cases = {
"fanbox": ("ffxivinitiala", "ffxivinitiala"),
"subscribestar": ("subscribestar.adult/initiala", "initiala"),
"patreon": ("initiala", "initiala"),
}
for provider, (ref, expected) in cases.items():
with self.subTest(provider=provider):
self.assertEqual(official_profile_handle(provider, ref), expected)
def test_a_numeric_patreon_user_page_has_no_handle(self):
"""`user/12345` 是内部 id，不是名字；学成别名会造出一个假作者。"""
for ref in ("user/12345", "12345", "/user/12345/"):
self.assertEqual(official_profile_handle("patreon", ref), "")
def test_archive_and_tag_sites_offer_no_handle_at_all(self):
cases = (("kemono", "fanbox/30917150"), ("coomer", "onlyfans/x"),
("pawchive", "fanbox/1"), ("rule34xxx", "lazyprocrastinator"),
("rule34video", "lazyprocrastinator"), ("f95zone", "50685"),
("rule34paheal", "tag"), ("simpcity", "thread"))
for provider, ref in cases:
with self.subTest(provider=provider):
self.assertEqual(official_profile_handle(provider, ref), "")
def test_every_connector_answers_this_question(self):
"""基类给了默认值，所以新增站点不会因为漏实现而炸在检查更新的中途。"""
for provider, factory in CONNECTORS.items():
with self.subTest(provider=provider):
self.assertIsInstance(factory.profile_handle("anything"), str)
self.assertEqual(official_profile_handle("not-registered", "x"), "")
def test_only_the_official_identity_providers_actually_return_one(self):
"""能返回手柄的站必须正好是 `_OFFICIAL_IDENTITY_PROVIDERS` 那一组。
这两处一旦不一致，就会出现「明明解析出了手柄却永远不学」或者反过来
「学了一个不该信的名字」，而两种都不会报错。
"""
answering = {provider for provider in CONNECTORS
if official_profile_handle(provider, "some-name")}
self.assertEqual(answering, set(_OFFICIAL_IDENTITY_PROVIDERS))
if __name__ == "__main__":
unittest.main()