549 lines
18 KiB
Python
549 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""B 站只读客户端测试（transport 边界 Mock，不联网）:
|
|
uv run python tests/test_bilibili_client.py
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.parse
|
|
from dataclasses import dataclass, field
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
from reaction_board.bilibili.client import ( # noqa: E402
|
|
BilibiliAPIError,
|
|
BilibiliClient,
|
|
BilibiliHTTPError,
|
|
BilibiliProtocolError,
|
|
HTTPResponse,
|
|
Pacer,
|
|
TransportError,
|
|
)
|
|
from fixtures import ( # noqa: E402
|
|
FOLDERS_PAYLOAD,
|
|
IDS_PAYLOAD,
|
|
INVALID_INFO,
|
|
NAV_PAYLOAD,
|
|
RELATION_PAYLOAD,
|
|
VALID_INFO,
|
|
VIEW_PAYLOAD,
|
|
VIEW_UNAVAILABLE,
|
|
info_for,
|
|
view_for,
|
|
)
|
|
|
|
|
|
def http_json(payload, status=200):
|
|
return HTTPResponse(
|
|
status=status,
|
|
headers={"content-type": "application/json"},
|
|
body=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class QueueTransport:
|
|
responses: list
|
|
calls: list = field(default_factory=list)
|
|
|
|
def request(self, method, url, headers, data=None, timeout=30):
|
|
self.calls.append({
|
|
"method": method, "url": url, "headers": dict(headers),
|
|
"data": data, "timeout": timeout,
|
|
})
|
|
item = self.responses.pop(0)
|
|
if isinstance(item, BaseException):
|
|
raise item
|
|
return item
|
|
|
|
|
|
@dataclass
|
|
class RouteTransport:
|
|
handler: object
|
|
calls: list = field(default_factory=list)
|
|
|
|
def request(self, method, url, headers, data=None, timeout=30):
|
|
self.calls.append({
|
|
"method": method, "url": url, "headers": dict(headers),
|
|
"data": data, "timeout": timeout,
|
|
})
|
|
return self.handler(url, headers)
|
|
|
|
|
|
def make_client(transport, *, cookie="SESSDATA=secret; buvid3=device", retries=3):
|
|
sleeps = []
|
|
client = BilibiliClient(
|
|
cookie=cookie,
|
|
csrf="csrf",
|
|
mid=28802351,
|
|
transport=transport,
|
|
pacer=Pacer(0, 0, sleep=sleeps.append),
|
|
sleeper=sleeps.append,
|
|
retries=retries,
|
|
backoff=1.0,
|
|
)
|
|
return client, sleeps
|
|
|
|
|
|
def test_fav_ids_parse_and_anonymous_header():
|
|
transport = QueueTransport([http_json(IDS_PAYLOAD)])
|
|
client, _ = make_client(transport)
|
|
ids = client.fav_ids(3700023351)
|
|
assert len(ids) == 4
|
|
assert ids[0].aid == 117530286889638
|
|
assert ids[0].type == 2
|
|
assert ids[0].bvid == "BV1xvxi6LEZD"
|
|
call = transport.calls[0]
|
|
assert "media_id=3700023351" in call["url"]
|
|
assert "SESSDATA" not in call["headers"]["cookie"]
|
|
assert "buvid3=device" in call["headers"]["cookie"]
|
|
assert call["headers"]["origin"] == "https://space.bilibili.com"
|
|
assert call["headers"]["priority"] == "u=1, i"
|
|
assert "Chrome/155.0.0.0" in call["headers"]["user-agent"]
|
|
assert "fid=3700023351" in call["headers"]["referer"]
|
|
|
|
|
|
def test_anonymous_failure_falls_back_to_full_cookie():
|
|
transport = QueueTransport([
|
|
http_json({"code": -101, "message": "账号未登录", "data": None}),
|
|
http_json(IDS_PAYLOAD),
|
|
])
|
|
client, _ = make_client(transport)
|
|
ids = client.fav_ids(3700023351)
|
|
assert len(ids) == 4
|
|
assert len(transport.calls) == 2
|
|
assert "SESSDATA" not in transport.calls[0]["headers"]["cookie"]
|
|
assert "SESSDATA=secret" in transport.calls[1]["headers"]["cookie"]
|
|
|
|
|
|
def test_business_error_is_not_retried():
|
|
error = {"code": -404, "message": "啥都木有", "data": None}
|
|
transport = QueueTransport([http_json(error), http_json(error)])
|
|
client, sleeps = make_client(transport)
|
|
try:
|
|
client.fav_ids(3700023351)
|
|
except BilibiliAPIError as exc:
|
|
assert exc.code == -404
|
|
assert "啥都木有" in str(exc)
|
|
else:
|
|
raise AssertionError("业务错误应抛出")
|
|
assert len(transport.calls) == 2 # 匿名 + 完整 Cookie 各一次，不重试
|
|
assert sleeps == []
|
|
|
|
|
|
def test_http_500_retries_with_backoff():
|
|
transport = QueueTransport([
|
|
http_json({"code": 0}, status=500),
|
|
http_json({"code": 0}, status=500),
|
|
http_json({"code": 0}, status=500),
|
|
])
|
|
client, sleeps = make_client(transport, retries=3)
|
|
try:
|
|
client.get_json("https://api.bilibili.com/x/test", auth=True)
|
|
except BilibiliHTTPError as exc:
|
|
assert exc.status == 500
|
|
else:
|
|
raise AssertionError("HTTP 500 应抛出")
|
|
assert len(transport.calls) == 3
|
|
assert sleeps == [1.0, 2.0]
|
|
|
|
|
|
def test_network_error_retry_then_success():
|
|
transport = QueueTransport([TransportError("reset"), http_json(IDS_PAYLOAD)])
|
|
client, sleeps = make_client(transport)
|
|
ids = client.fav_ids(3700023351)
|
|
assert len(ids) == 4
|
|
assert len(transport.calls) == 2
|
|
assert sleeps == [1.0]
|
|
|
|
|
|
def test_invalid_json_is_protocol_error():
|
|
bad = HTTPResponse(200, {}, b"not json")
|
|
transport = QueueTransport([bad, bad])
|
|
client, _ = make_client(transport)
|
|
try:
|
|
client.fav_ids(3700023351)
|
|
except BilibiliProtocolError as exc:
|
|
assert "JSON" in str(exc)
|
|
else:
|
|
raise AssertionError("非法 JSON 应抛出协议错误")
|
|
|
|
|
|
def test_infos_chunking_and_silent_missing():
|
|
aids = list(range(1000, 1450))
|
|
missing = 1003
|
|
|
|
def handler(url, headers):
|
|
query = urllib.parse.parse_qs(urllib.parse.urlparse(url).query)
|
|
resources = query["resources"][0]
|
|
requested = [int(part.split(":")[0]) for part in resources.split(",")]
|
|
data = [
|
|
info_for(aid, title=f"视频{aid}", up_mid=100, up_name="UP")
|
|
for aid in requested if aid != missing
|
|
]
|
|
return http_json({"code": 0, "message": "0", "ttl": 1, "data": data})
|
|
|
|
transport = RouteTransport(handler)
|
|
client, _ = make_client(transport)
|
|
result = client.fav_infos(3700023351, aids, folder_mid=28802351, chunk_size=200)
|
|
assert len(transport.calls) == 3
|
|
sizes = []
|
|
for call in transport.calls:
|
|
query = urllib.parse.parse_qs(urllib.parse.urlparse(call["url"]).query)
|
|
sizes.append(len(query["resources"][0].split(",")))
|
|
assert sizes == [200, 200, 50]
|
|
assert result.requested == tuple(aids)
|
|
assert result.missing == (missing,)
|
|
assert len(result.items) == 449
|
|
assert result.items[1000].title == "视频1000"
|
|
assert result.items[1000].invalid is False
|
|
|
|
|
|
def test_infos_invalid_attr_and_empty():
|
|
transport = QueueTransport([
|
|
http_json({"code": 0, "message": "0", "ttl": 1, "data": [INVALID_INFO]})
|
|
])
|
|
client, _ = make_client(transport)
|
|
result = client.fav_infos(3700023351, [INVALID_INFO["id"]])
|
|
assert result.items[INVALID_INFO["id"]].invalid is True
|
|
assert result.missing == ()
|
|
|
|
# 空输入不发起请求
|
|
empty = client.fav_infos(3700023351, [])
|
|
assert empty.requested == () and empty.items == {}
|
|
assert len(transport.calls) == 1
|
|
|
|
|
|
def test_infos_parses_page_count():
|
|
payload = {"code": 0, "message": "0", "ttl": 1, "data": [VALID_INFO, INVALID_INFO]}
|
|
client, _ = make_client(QueueTransport([http_json(payload)]))
|
|
result = client.fav_infos(
|
|
3700023351, [VALID_INFO["id"], INVALID_INFO["id"]]
|
|
)
|
|
assert result.items[VALID_INFO["id"]].pages == 2
|
|
assert result.items[INVALID_INFO["id"]].pages == 1
|
|
|
|
# page 字段缺失或非法时保持 None，不把 0 当分 P 数
|
|
broken = json.loads(json.dumps(VALID_INFO))
|
|
broken["id"] = 999
|
|
broken["page"] = 0
|
|
client, _ = make_client(QueueTransport([http_json({
|
|
"code": 0, "message": "0", "ttl": 1, "data": [broken],
|
|
})]))
|
|
assert client.fav_infos(3700023351, [999]).items[999].pages is None
|
|
|
|
|
|
def test_infos_unexpected_aid_is_protocol_error():
|
|
transport = QueueTransport([http_json({
|
|
"code": 0, "message": "0", "ttl": 1, "data": [VALID_INFO],
|
|
})])
|
|
client, _ = make_client(transport)
|
|
try:
|
|
client.fav_infos(3700023351, [1])
|
|
except BilibiliProtocolError as exc:
|
|
assert "未请求" in str(exc)
|
|
else:
|
|
raise AssertionError("返回未请求 aid 应报错")
|
|
|
|
|
|
def test_fav_folders_uses_full_cookie():
|
|
transport = QueueTransport([http_json(FOLDERS_PAYLOAD)])
|
|
client, _ = make_client(transport)
|
|
folders = client.fav_folders()
|
|
assert [(f.media_id, f.media_count) for f in folders] == [
|
|
(4451053851, 167), (3700023351, 664)
|
|
]
|
|
call = transport.calls[0]
|
|
assert "up_mid=28802351" in call["url"]
|
|
assert "SESSDATA=secret" in call["headers"]["cookie"]
|
|
|
|
|
|
def test_fav_infos_chunk_size_validation():
|
|
client, _ = make_client(QueueTransport([]))
|
|
for size in (0, 401):
|
|
try:
|
|
client.fav_infos(1, [1], chunk_size=size)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError(f"chunk_size={size} 应失败")
|
|
|
|
|
|
def test_pacer_delays_after_first_request():
|
|
sleeps = []
|
|
|
|
class Rng:
|
|
def uniform(self, a, b):
|
|
assert (a, b) == (1.5, 2.5)
|
|
return 2.0
|
|
|
|
pacer = Pacer(1.5, 2.5, sleep=sleeps.append, rng=Rng())
|
|
assert pacer.wait() == 0.0
|
|
assert pacer.wait() == 2.0
|
|
assert pacer.wait() == 2.0
|
|
assert sleeps == [2.0, 2.0]
|
|
try:
|
|
Pacer(2.0, 1.0)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError("非法间隔应失败")
|
|
|
|
|
|
def test_timeout_and_method_forwarded():
|
|
transport = QueueTransport([http_json(IDS_PAYLOAD)])
|
|
client, _ = make_client(transport)
|
|
client.fav_ids(1)
|
|
call = transport.calls[0]
|
|
assert call["method"] == "GET"
|
|
assert call["data"] is None
|
|
assert call["timeout"] == 30
|
|
|
|
|
|
def test_view_seven_metrics():
|
|
transport = QueueTransport([http_json(VIEW_PAYLOAD)])
|
|
client, _ = make_client(transport)
|
|
stat = client.view(117530286889638)
|
|
assert stat.aid == 117530286889638
|
|
assert stat.bvid == "BV1xvxi6LEZD"
|
|
assert stat.metrics() == {
|
|
"view": 819, "like": 44, "favorite": 43, "coin": 9,
|
|
"share": 1, "reply": 16, "danmaku": 7,
|
|
}
|
|
assert "aid=117530286889638" in transport.calls[0]["url"]
|
|
|
|
|
|
def test_view_unavailable_and_malformed():
|
|
transport = QueueTransport([
|
|
http_json(VIEW_UNAVAILABLE), http_json(VIEW_UNAVAILABLE),
|
|
])
|
|
client, _ = make_client(transport)
|
|
try:
|
|
client.view(1)
|
|
except BilibiliAPIError as exc:
|
|
assert exc.code == 62012
|
|
else:
|
|
raise AssertionError("不可访问视频应抛业务错误")
|
|
|
|
payload = view_for(1)
|
|
payload["data"]["stat"].pop("coin")
|
|
bad = http_json(payload)
|
|
transport = QueueTransport([bad, bad])
|
|
client, _ = make_client(transport)
|
|
try:
|
|
client.view(1)
|
|
except BilibiliProtocolError as exc:
|
|
assert "coin" in str(exc)
|
|
else:
|
|
raise AssertionError("缺少 stat 字段应报协议错误")
|
|
|
|
|
|
def test_view_detail_by_aid_and_bvid():
|
|
transport = QueueTransport([http_json(VIEW_PAYLOAD)])
|
|
client, _ = make_client(transport)
|
|
detail = client.view_detail(aid=117530286889638)
|
|
assert detail.aid == 117530286889638
|
|
assert detail.bvid == "BV1xvxi6LEZD"
|
|
assert detail.up_mid == 434442072
|
|
assert detail.up_name == "_-抱抱-_"
|
|
assert detail.pubdate == 1793370433
|
|
assert detail.duration == 9504
|
|
assert detail.pages == 2
|
|
assert detail.cover and detail.cover.startswith("http")
|
|
assert detail.metrics()["view"] == 819
|
|
assert detail.up_face and detail.up_face.startswith("http")
|
|
assert "aid=117530286889638" in transport.calls[0]["url"]
|
|
|
|
transport = QueueTransport([http_json(VIEW_PAYLOAD)])
|
|
client, _ = make_client(transport)
|
|
detail = client.view_detail(bvid="BV1xvxi6LEZD")
|
|
assert detail.aid == 117530286889638
|
|
assert "bvid=BV1xvxi6LEZD" in transport.calls[0]["url"]
|
|
|
|
|
|
def test_view_detail_fallbacks_and_required_arg():
|
|
payload = view_for(1)
|
|
for key in ("owner", "duration", "pages", "videos", "pic"):
|
|
payload["data"].pop(key, None)
|
|
transport = QueueTransport([http_json(payload)])
|
|
client, _ = make_client(transport)
|
|
detail = client.view_detail(aid=1)
|
|
assert detail.up_mid is None and detail.up_name is None
|
|
assert detail.duration == 0
|
|
assert detail.pages is None
|
|
assert detail.cover is None
|
|
assert detail.up_face is None
|
|
|
|
client, _ = make_client(QueueTransport([]))
|
|
for kwargs in ({}, {"aid": 1, "bvid": "BV1xvxi6LEZD"}):
|
|
try:
|
|
client.view_detail(**kwargs)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError("必须且只能提供 aid 或 bvid")
|
|
|
|
|
|
def test_view_detail_unavailable():
|
|
transport = QueueTransport([
|
|
http_json(VIEW_UNAVAILABLE), http_json(VIEW_UNAVAILABLE),
|
|
])
|
|
client, _ = make_client(transport)
|
|
try:
|
|
client.view_detail(bvid="BV1xvxi6LEZD")
|
|
except BilibiliAPIError as exc:
|
|
assert exc.code == 62012
|
|
else:
|
|
raise AssertionError("不可访问视频应抛业务错误")
|
|
|
|
|
|
def test_resolve_short_link():
|
|
redirect = HTTPResponse(
|
|
302,
|
|
{"location": "https://www.bilibili.com/video/BV1x8x166EzY"},
|
|
b"",
|
|
)
|
|
transport = QueueTransport([redirect])
|
|
client, _ = make_client(transport)
|
|
final = client.resolve_short_link("https://b23.tv/abc")
|
|
assert final == "https://www.bilibili.com/video/BV1x8x166EzY"
|
|
assert transport.calls[0]["method"] == "GET"
|
|
assert "b23.tv" in transport.calls[0]["url"]
|
|
|
|
# 多跳重定向，大小写不同的 Location 头
|
|
transport = QueueTransport([
|
|
HTTPResponse(301, {"Location": "https://b23.tv/next"}, b""),
|
|
HTTPResponse(302, {"location": "https://www.bilibili.com/video/BV1x8x166EzY"}, b""),
|
|
])
|
|
client, _ = make_client(transport)
|
|
assert client.resolve_short_link("https://b23.tv/start").endswith("BV1x8x166EzY")
|
|
|
|
# 非重定向直接返回最终 URL
|
|
transport = QueueTransport([HTTPResponse(200, {}, b"<html>")])
|
|
client, _ = make_client(transport)
|
|
assert client.resolve_short_link("https://b23.tv/x") == "https://b23.tv/x"
|
|
|
|
# HTTP 错误应抛出
|
|
transport = QueueTransport([HTTPResponse(404, {}, b"not found")])
|
|
client, _ = make_client(transport)
|
|
try:
|
|
client.resolve_short_link("https://b23.tv/x")
|
|
except BilibiliHTTPError as exc:
|
|
assert exc.status == 404
|
|
else:
|
|
raise AssertionError("短链 HTTP 404 应抛出")
|
|
|
|
|
|
def test_relation_stat():
|
|
transport = QueueTransport([http_json(RELATION_PAYLOAD)])
|
|
client, _ = make_client(transport)
|
|
fan = client.relation_stat(37754047)
|
|
assert fan.mid == 37754047
|
|
assert fan.follower == 2767268
|
|
assert fan.following == 95
|
|
assert "vmid=37754047" in transport.calls[0]["url"]
|
|
|
|
|
|
def test_check_login():
|
|
transport = QueueTransport([http_json(NAV_PAYLOAD)])
|
|
client, _ = make_client(transport)
|
|
state = client.check_login()
|
|
assert state.is_login is True
|
|
assert state.mid == 28802351
|
|
assert state.uname == "冷萃编译器"
|
|
assert "SESSDATA=secret" in transport.calls[0]["headers"]["cookie"]
|
|
|
|
transport = QueueTransport([http_json({
|
|
"code": 0, "message": "0", "ttl": 1,
|
|
"data": {"isLogin": False, "mid": 0, "uname": ""},
|
|
})])
|
|
client, _ = make_client(transport)
|
|
state = client.check_login()
|
|
assert state.is_login is False and state.mid == 0
|
|
|
|
transport = QueueTransport([http_json({"code": -101, "message": "账号未登录", "data": None})])
|
|
client, _ = make_client(transport)
|
|
state = client.check_login()
|
|
assert state.is_login is False
|
|
assert state.code == -101
|
|
assert "账号未登录" in state.message
|
|
|
|
|
|
def test_check_login_without_cookie_makes_no_request():
|
|
transport = QueueTransport([])
|
|
client, _ = make_client(transport, cookie="")
|
|
state = client.check_login()
|
|
assert state.is_login is False
|
|
assert state.message == "未配置 cookie"
|
|
assert transport.calls == []
|
|
|
|
|
|
def test_missing_data_is_protocol_error():
|
|
for payload in (
|
|
{"code": 0, "message": "OK"},
|
|
{"code": 0, "message": "OK", "data": None},
|
|
):
|
|
transport = QueueTransport([http_json(payload)])
|
|
client, _ = make_client(transport)
|
|
try:
|
|
client.fav_ids(1)
|
|
except BilibiliProtocolError as exc:
|
|
assert "data" in str(exc)
|
|
else:
|
|
raise AssertionError(f"ids 缺少 data 应报协议错误: {payload}")
|
|
|
|
transport = QueueTransport([http_json(payload)])
|
|
client, _ = make_client(transport)
|
|
try:
|
|
client.fav_infos(1, [123])
|
|
except BilibiliProtocolError as exc:
|
|
assert "data" in str(exc)
|
|
else:
|
|
raise AssertionError(f"infos 缺少 data 应报协议错误: {payload}")
|
|
|
|
|
|
def test_empty_data_list_is_valid():
|
|
transport = QueueTransport([http_json({"code": 0, "message": "OK", "data": []})])
|
|
client, _ = make_client(transport)
|
|
assert client.fav_ids(4413147139) == []
|
|
|
|
transport = QueueTransport([http_json({"code": 0, "message": "OK", "data": []})])
|
|
client, _ = make_client(transport)
|
|
result = client.fav_infos(1, [123, 456])
|
|
assert result.items == {}
|
|
assert result.missing == (123, 456)
|
|
|
|
|
|
def main():
|
|
test_fav_ids_parse_and_anonymous_header()
|
|
test_anonymous_failure_falls_back_to_full_cookie()
|
|
test_business_error_is_not_retried()
|
|
test_http_500_retries_with_backoff()
|
|
test_network_error_retry_then_success()
|
|
test_invalid_json_is_protocol_error()
|
|
test_infos_chunking_and_silent_missing()
|
|
test_infos_invalid_attr_and_empty()
|
|
test_infos_unexpected_aid_is_protocol_error()
|
|
test_fav_folders_uses_full_cookie()
|
|
test_fav_infos_chunk_size_validation()
|
|
test_pacer_delays_after_first_request()
|
|
test_timeout_and_method_forwarded()
|
|
test_view_seven_metrics()
|
|
test_view_unavailable_and_malformed()
|
|
test_view_detail_by_aid_and_bvid()
|
|
test_view_detail_fallbacks_and_required_arg()
|
|
test_view_detail_unavailable()
|
|
test_resolve_short_link()
|
|
test_relation_stat()
|
|
test_check_login()
|
|
test_check_login_without_cookie_makes_no_request()
|
|
test_missing_data_is_protocol_error()
|
|
test_empty_data_list_is_valid()
|
|
print("OK: B 站只读客户端契约与重试测试全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|