bot-ci
8e7ead1669
Load business config as JSONC (comments + trailing commas) via an in-repo stripper then stdlib json — no new deps. Default path prefers config.jsonc, falls back to legacy config.json. Auth write-back targets only sibling secrets.json so JSONC comments are never wiped; secrets overlay merges on load. Examples, compose, Docker, docs, favsort and tools all share the same loader. Phase A only — no DB settings migration.
249 lines
9.8 KiB
Python
249 lines
9.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""双账号 auth 纯函数与写回隔离测试（不联网）: uv run python tests/test_auth.py"""
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from contextlib import redirect_stdout
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
from reaction_board.bilibili.auth import ( # noqa: E402
|
|
cookie_header,
|
|
cookie_value,
|
|
cookies_from_set_cookie,
|
|
make_headers,
|
|
merge_cookie,
|
|
parse_cookie,
|
|
sessdata_expiry,
|
|
)
|
|
from reaction_board.bilibili.auth_flow import ( # noqa: E402
|
|
AuthService,
|
|
_QR_BLOCKS,
|
|
format_qr_ascii,
|
|
render_qr_ascii,
|
|
write_qr_png,
|
|
)
|
|
import qrcodegen # noqa: E402
|
|
|
|
READ_OLD = "buvid3=READ-B3; {{SECRET_i5bghhpt}}readcsrf; DedeUserID=11; SESSDATA=old%2C1700000000%2Ca"
|
|
WRITE_OLD = "buvid3=WRITE-B3; bili_jct=writecsrf; DedeUserID=22; SESSDATA=old%2C1700000000%2Cb"
|
|
NEW = {"SESSDATA": "new%2C1800000000%2Cxyz", "bili_jct": "newcsrf", "DedeUserID": "99"}
|
|
|
|
|
|
def raw_config():
|
|
return {
|
|
"bilibili": {
|
|
"read": {"mid": 11, "csrf": "readcsrf", "cookie": READ_OLD},
|
|
"write": {"mid": 22, "csrf": "writecsrf", "cookie": WRITE_OLD},
|
|
},
|
|
"folders": [],
|
|
}
|
|
|
|
|
|
def test_cookie_helpers():
|
|
assert parse_cookie(READ_OLD)[0] == ("buvid3", "READ-B3")
|
|
merged = merge_cookie(WRITE_OLD, NEW)
|
|
assert cookie_value(merged, "SESSDATA") == "new%2C1800000000%2Cxyz"
|
|
assert cookie_value(merged, "bili_jct") == "newcsrf"
|
|
assert cookie_value(merged, "buvid3") == "WRITE-B3"
|
|
assert sessdata_expiry(merged) == 1800000000
|
|
headers = ["SESSDATA=x; Path=/", "bili_jct=y; Domain=.bilibili.com"]
|
|
assert cookies_from_set_cookie(headers) == {"SESSDATA": "x", "bili_jct": "y"}
|
|
|
|
|
|
def test_finish_updates_only_selected_account_and_rolls_back():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
# 主配置为 JSONC（含注释）；auth 只写 secrets.json
|
|
config_path = Path(tmp) / "config.jsonc"
|
|
secrets_path = Path(tmp) / "secrets.json"
|
|
original = raw_config()
|
|
config_body = (
|
|
"// keep me\n"
|
|
+ json.dumps({"folders": [], "telegram": {"enabled": False}}, ensure_ascii=False)
|
|
)
|
|
config_path.write_text(config_body, encoding="utf-8")
|
|
secrets_path.write_text(json.dumps(original, ensure_ascii=False), encoding="utf-8")
|
|
service = AuthService(config_path, account="write")
|
|
assert service.secrets_path == secrets_path
|
|
|
|
service._get_json = lambda credentials, url, auth: { # type: ignore[method-assign]
|
|
"code": 0, "data": {"isLogin": True, "mid": 99, "uname": "writer"}
|
|
}
|
|
assert service.finish(service._load_raw(), NEW) == 0
|
|
saved = json.loads(secrets_path.read_text(encoding="utf-8"))
|
|
assert saved["bilibili"]["read"] == original["bilibili"]["read"]
|
|
assert saved["bilibili"]["write"]["mid"] == 99
|
|
assert saved["bilibili"]["write"]["csrf"] == "newcsrf"
|
|
assert cookie_value(saved["bilibili"]["write"]["cookie"], "buvid3") == "WRITE-B3"
|
|
assert secrets_path.with_name("secrets.json.bak").exists()
|
|
assert config_path.read_text(encoding="utf-8") == config_body
|
|
assert "// keep me" in config_path.read_text(encoding="utf-8")
|
|
|
|
before = secrets_path.read_text(encoding="utf-8")
|
|
service._get_json = lambda credentials, url, auth: { # type: ignore[method-assign]
|
|
"code": -101, "data": {"isLogin": False}
|
|
}
|
|
assert service.finish(service._load_raw(), NEW) == 4
|
|
assert secrets_path.read_text(encoding="utf-8") == before
|
|
assert config_path.read_text(encoding="utf-8") == config_body
|
|
|
|
|
|
def test_finish_seeds_secrets_from_legacy_config_without_touching_it():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "config.json"
|
|
original = raw_config()
|
|
path.write_text(json.dumps(original, ensure_ascii=False), encoding="utf-8")
|
|
service = AuthService(path, account="write")
|
|
service._get_json = lambda credentials, url, auth: { # type: ignore[method-assign]
|
|
"code": 0, "data": {"isLogin": True, "mid": 99, "uname": "writer"}
|
|
}
|
|
before = path.read_text(encoding="utf-8")
|
|
assert service.finish(service._load_raw(), NEW) == 0
|
|
assert path.read_text(encoding="utf-8") == before
|
|
secrets = json.loads((Path(tmp) / "secrets.json").read_text(encoding="utf-8"))
|
|
assert secrets["bilibili"]["write"]["mid"] == 99
|
|
assert secrets["bilibili"]["read"] == original["bilibili"]["read"]
|
|
|
|
|
|
def test_qr_png():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "qr.png"
|
|
write_qr_png("https://example.com/login?key=abc", path)
|
|
assert path.read_bytes().startswith(b"\x89PNG\r\n\x1a\n")
|
|
|
|
|
|
def test_render_qr_ascii_roundtrip_and_quiet_zone():
|
|
text = "https://passport.bilibili.com/h5/cloud/login?qrcode_key=abc"
|
|
for border in (0, 1, 2, 4):
|
|
art = render_qr_ascii(text, border=border)
|
|
qr = qrcodegen.QrCode.encode_text(text, qrcodegen.QrCode.Ecc.MEDIUM)
|
|
size = qr.get_size()
|
|
dim = size + border * 2
|
|
lines = art.split("\n")
|
|
assert len(lines) == (dim + 1) // 2
|
|
assert all(len(line) == dim for line in lines)
|
|
inverse = {glyph: pair for pair, glyph in _QR_BLOCKS.items()}
|
|
recovered = []
|
|
for row, line in enumerate(lines):
|
|
tops, bottoms = [], []
|
|
for glyph in line:
|
|
top, bottom = inverse[glyph]
|
|
tops.append(top)
|
|
bottoms.append(bottom)
|
|
recovered.append(tops)
|
|
if row * 2 + 1 < dim:
|
|
recovered.append(bottoms)
|
|
assert len(recovered) == dim
|
|
for y in range(dim):
|
|
for x in range(dim):
|
|
expected = qr.get_module(x - border, y - border)
|
|
assert recovered[y][x] is expected
|
|
if border >= 1:
|
|
assert all(line[0] == " " and line[-1] == " " for line in lines)
|
|
if border >= 2:
|
|
assert set(lines[0]) == {" "}
|
|
|
|
try:
|
|
render_qr_ascii(text, border=-1)
|
|
except ValueError as exc:
|
|
assert "border" in str(exc)
|
|
else:
|
|
raise AssertionError("负 border 应报错")
|
|
|
|
colored = format_qr_ascii(text, color=True)
|
|
plain = format_qr_ascii(text, color=False)
|
|
assert plain == render_qr_ascii(text)
|
|
assert "\033[48;5;231m" in colored
|
|
assert "\033[0m" in colored
|
|
assert plain.split("\n")[0] in colored
|
|
|
|
|
|
def test_run_prints_ascii_qr_without_network():
|
|
login_url = "https://passport.bilibili.com/h5/cloud/login?qrcode_key=unit-test"
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "config.json"
|
|
path.write_text(json.dumps(raw_config(), ensure_ascii=False), encoding="utf-8")
|
|
service = AuthService(path, account="read")
|
|
|
|
def fake_request(credentials, url, *, auth):
|
|
if "qrcode/generate" in url:
|
|
return {
|
|
"code": 0,
|
|
"message": "OK",
|
|
"data": {"url": login_url, "qrcode_key": "abcdefghijklmnop"},
|
|
}, []
|
|
if "qrcode/poll" in url:
|
|
return {"code": 0, "data": {"code": 86038, "message": "二维码已过期"}}, []
|
|
raise AssertionError(url)
|
|
|
|
service._request = fake_request # type: ignore[method-assign]
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
assert service.run(timeout=9, no_browser=True) == 3
|
|
output = buf.getvalue()
|
|
ascii_qr = render_qr_ascii(login_url)
|
|
assert ascii_qr in output
|
|
assert "请使用哔哩哔哩 App 扫描下方二维码" in output
|
|
assert login_url in output
|
|
assert "二维码已过期" in output
|
|
assert "bili-fav-read-abcdefgh.png" in output
|
|
|
|
|
|
def test_make_headers_matches_raw_request():
|
|
headers = make_headers({"cookie": "SESSDATA=x; buvid3=y"}, auth=False)
|
|
assert headers["origin"] == "https://space.bilibili.com"
|
|
assert headers["referer"] == "https://space.bilibili.com/"
|
|
assert headers["priority"] == "u=1, i"
|
|
assert headers["user-agent"].endswith("Chrome/155.0.0.0 Safari/537.36")
|
|
assert headers["sec-ch-ua"].startswith('"Google Chrome";v="155"')
|
|
assert "SESSDATA" not in headers["cookie"]
|
|
assert "buvid3=y" in headers["cookie"]
|
|
|
|
full = make_headers({"cookie": "SESSDATA=x"}, auth=True)
|
|
assert full["cookie"] == "SESSDATA=x"
|
|
|
|
custom = make_headers(
|
|
{"cookie": ""}, referer="https://space.bilibili.com/1/favlist?fid=2"
|
|
)
|
|
assert custom["referer"] == "https://space.bilibili.com/1/favlist?fid=2"
|
|
assert custom["cookie"] == ""
|
|
|
|
|
|
def test_make_headers_drops_placeholder_and_non_latin1_cookies():
|
|
placeholder = "只读账号的完整 Cookie（需含 SESSDATA、bili_jct、DedeUserID 等）"
|
|
headers = make_headers({"cookie": placeholder}, auth=False)
|
|
for value in headers.values():
|
|
value.encode("latin-1")
|
|
assert headers["cookie"] == ""
|
|
assert cookie_header(placeholder, auth=True) == ""
|
|
|
|
mixed = make_headers(
|
|
{"cookie": "buvid3=abc; 写入账号的完整 Cookie; SESSDATA=x; note=中文"},
|
|
auth=True,
|
|
)
|
|
mixed["cookie"].encode("latin-1")
|
|
assert mixed["cookie"] == "buvid3=abc; SESSDATA=x"
|
|
missing = make_headers({}, auth=False)
|
|
assert missing["cookie"] == ""
|
|
|
|
|
|
def main():
|
|
test_cookie_helpers()
|
|
test_finish_updates_only_selected_account_and_rolls_back()
|
|
test_finish_seeds_secrets_from_legacy_config_without_touching_it()
|
|
test_qr_png()
|
|
test_render_qr_ascii_roundtrip_and_quiet_zone()
|
|
test_run_prints_ascii_qr_without_network()
|
|
test_make_headers_matches_raw_request()
|
|
test_make_headers_drops_placeholder_and_non_latin1_cookies()
|
|
print("OK: 双账号 auth Cookie 合并、secrets 写回隔离、终端二维码与回滚测试全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|