sansenhoshiandClaude Code
a31aab00e8
- 旧平铺模块（data/data_utils/database_op/image_builder*/get_bf6_data/ gametools_bf6/bf6_data/img_utils）拆分为 handlers（绑定/查询）、 services（fetchers/processing/query）、utils（渲染/参数）分层； 新增 repository.py / config.py / paths.py - 字体/图片/模板资源迁入 res/，新增 BF2042 卡面素材（bg/object_icon 等） - browser_ext/：BF2042 扩展捕获桥（WS 直连兜底通道）源码入库 - 新增 hexi/web/src/plugins/bf_bot 前端管理页 - 新增 tests：test_bf_bot_helpers / repository / tracker_api Co-Authored-By: Claude Code <noreply@anthropic.com>
430 lines
14 KiB
Python
430 lines
14 KiB
Python
"""BF2042 直连通道(tracker_api)与通道调度(tracker_edge)单测。
|
|
|
|
两个模块都按文件裸加载(tests/test_bf_bot_repository.py 同法,不触发插件包
|
|
__init__ 的 NoneBot 初始化);**全部用例零真实网络**——令牌端点、API 请求
|
|
分别用假 ``_do_refresh`` / 假 ``cr.Session`` / 假子模块替换,令牌文件一律
|
|
指到 tmp_path,绝不碰 hexi/data 下的真实令牌与绑定库。
|
|
"""
|
|
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
_PLUGIN_DIR = (
|
|
Path(__file__).resolve().parents[1] / "hexi" / "plugins" / "nonebot_plugin_bf_bot"
|
|
)
|
|
|
|
|
|
def _load(name: str, path: Path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def _make_pkg(name: str, path: Path):
|
|
pkg = types.ModuleType(name)
|
|
pkg.__path__ = [str(path)] # type: ignore[attr-defined]
|
|
sys.modules[name] = pkg
|
|
return pkg
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# tracker_api — 令牌存储 / 引导提取 / 刷新链 / HTTP 错误映射
|
|
# --------------------------------------------------------------------------- #
|
|
_API_PATH = _PLUGIN_DIR / "services" / "fetchers" / "tracker_api.py"
|
|
|
|
|
|
@pytest.fixture
|
|
def api_mod(tmp_path, monkeypatch):
|
|
"""裸加载 tracker_api;令牌路径指到 tmp_path 下的文件。"""
|
|
module = _load("bf_tracker_api_under_test", _API_PATH)
|
|
token_path = tmp_path / "trn_tokens.json"
|
|
monkeypatch.setattr(module, "_token_path", lambda: token_path)
|
|
return module
|
|
|
|
|
|
def test_decode_tokens_blob_latin1():
|
|
module = _load("bf_tracker_api_decode1", _API_PATH)
|
|
payload = {"access_token": "a1", "refresh_token": "r1"}
|
|
raw = (
|
|
b"\x00_https://tracker.gg\x00\x01auth.tokens"
|
|
+ b"\x01"
|
|
+ json.dumps(payload).encode("latin-1")
|
|
+ b"\x00garbage"
|
|
)
|
|
assert module._decode_tokens_blob(raw) == payload
|
|
|
|
|
|
def test_decode_tokens_blob_utf16_and_garbage():
|
|
module = _load("bf_tracker_api_decode2", _API_PATH)
|
|
payload = {"access_token": "a2", "refresh_token": "r2"}
|
|
raw = (
|
|
"auth.tokens".encode("utf-16-le")
|
|
+ b"\x00"
|
|
+ json.dumps(payload).encode("utf-16-le")
|
|
)
|
|
assert module._decode_tokens_blob(raw) == payload
|
|
assert module._decode_tokens_blob(b"\x01no keys here") is None
|
|
|
|
|
|
def test_extract_tokens_from_dir(tmp_path):
|
|
module = _load("bf_tracker_api_extract", _API_PATH)
|
|
|
|
assert module._extract_tokens_from_dir(tmp_path / "missing") is None
|
|
|
|
garbage = tmp_path / "garbage"
|
|
garbage.mkdir()
|
|
(garbage / "000003.log").write_bytes(b"nothing to see")
|
|
assert module._extract_tokens_from_dir(garbage) is None
|
|
|
|
hit = tmp_path / "hit"
|
|
hit.mkdir()
|
|
(hit / "000003.log").write_bytes(
|
|
b"auth.tokens\x01"
|
|
+ json.dumps({"access_token": "a", "refresh_token": "r"}).encode()
|
|
)
|
|
found = module._extract_tokens_from_dir(hit)
|
|
assert found and found["refresh_token"] == "r"
|
|
|
|
|
|
def test_token_fresh(api_mod):
|
|
assert api_mod._token_fresh({"access_token": "x", "expires_at": 2000.0}, now=1000.0)
|
|
assert not api_mod._token_fresh(
|
|
{"access_token": "x", "expires_at": 500.0}, now=1000.0
|
|
)
|
|
assert not api_mod._token_fresh({"access_token": "x"})
|
|
assert not api_mod._token_fresh({"expires_at": 9999.0})
|
|
|
|
|
|
def test_normalize_token_payload_keeps_prev_refresh(api_mod):
|
|
fresh = api_mod._normalize_token_payload(
|
|
{"access_token": "a", "expires_in": 3600}, "old"
|
|
)
|
|
assert fresh["access_token"] == "a"
|
|
assert fresh["refresh_token"] == "old" # 未轮换时沿用旧值
|
|
assert fresh["expires_at"] > api_mod.time.time() + 3000
|
|
|
|
|
|
def test_ensure_access_token_fresh_skips_refresh(api_mod, monkeypatch):
|
|
api_mod._token_path().write_text(
|
|
json.dumps({"access_token": "good", "refresh_token": "r", "expires_at": 9e9}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def _boom(*_a, **_k):
|
|
raise AssertionError("fresh 令牌不应触发刷新")
|
|
|
|
monkeypatch.setattr(api_mod, "_do_refresh", _boom)
|
|
assert api_mod.ensure_access_token() == "good"
|
|
|
|
|
|
def test_ensure_access_token_refreshes_and_persists(api_mod, monkeypatch):
|
|
api_mod._token_path().write_text(
|
|
json.dumps({"access_token": "stale", "refresh_token": "r1", "expires_at": 1.0}),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(api_mod, "_bootstrap_tokens", lambda: None)
|
|
seen: list[str] = []
|
|
|
|
def _fake_refresh(refresh_token: str) -> dict:
|
|
seen.append(refresh_token)
|
|
return {
|
|
"access_token": "new",
|
|
"refresh_token": "r2",
|
|
"expires_at": api_mod.time.time() + 3000,
|
|
}
|
|
|
|
monkeypatch.setattr(api_mod, "_do_refresh", _fake_refresh)
|
|
assert api_mod.ensure_access_token() == "new"
|
|
assert seen == ["r1"]
|
|
saved = json.loads(api_mod._token_path().read_text(encoding="utf-8"))
|
|
assert saved["refresh_token"] == "r2" # 轮换后的令牌已落盘
|
|
assert saved["access_token"] == "new"
|
|
|
|
|
|
def test_ensure_access_token_bootstrap_when_file_token_rejected(api_mod, monkeypatch):
|
|
api_mod._token_path().write_text(
|
|
json.dumps(
|
|
{"access_token": "stale", "refresh_token": "dead", "expires_at": 1.0}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(
|
|
api_mod, "_bootstrap_tokens", lambda: {"refresh_token": "from-browser"}
|
|
)
|
|
|
|
def _fake_refresh(refresh_token: str) -> dict:
|
|
if refresh_token == "dead":
|
|
raise api_mod.TrnAuthError("invalid_grant")
|
|
return {
|
|
"access_token": "ok",
|
|
"refresh_token": "rt3",
|
|
"expires_at": api_mod.time.time() + 3000,
|
|
}
|
|
|
|
monkeypatch.setattr(api_mod, "_do_refresh", _fake_refresh)
|
|
assert api_mod.ensure_access_token() == "ok"
|
|
|
|
|
|
def test_ensure_access_token_raises_without_any_source(api_mod, monkeypatch):
|
|
monkeypatch.setattr(api_mod, "_bootstrap_tokens", lambda: None)
|
|
with pytest.raises(api_mod.TrnAuthError):
|
|
api_mod.ensure_access_token()
|
|
|
|
|
|
class _FakeResp:
|
|
def __init__(self, status: int, payload=None, text: str = ""):
|
|
self.status_code = status
|
|
self._payload = payload
|
|
self.text = text
|
|
|
|
def json(self):
|
|
if self._payload is None:
|
|
raise ValueError("not json")
|
|
return self._payload
|
|
|
|
|
|
class _FakeSession:
|
|
"""按 URL 关键词路由的假会话;profile 无关键词时兜底到 'profile'。"""
|
|
|
|
def __init__(self, routes: dict):
|
|
self._routes = routes
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_exc):
|
|
return False
|
|
|
|
def get(self, url: str, headers=None, timeout=None):
|
|
for key, resp in self._routes.items():
|
|
if key == "profile":
|
|
continue
|
|
if len(key) > len("segments/") and key in url:
|
|
return resp
|
|
return self._routes["profile"]
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_session(api_mod, monkeypatch):
|
|
def _install(routes: dict):
|
|
fake_cr = types.SimpleNamespace(Session=lambda **kw: _FakeSession(routes))
|
|
monkeypatch.setattr(api_mod, "cr", fake_cr)
|
|
monkeypatch.setattr(api_mod, "ensure_access_token", lambda force=False: "tok")
|
|
|
|
return _install
|
|
|
|
|
|
def test_fetch_sync_maps_404_to_not_found(api_mod, fake_session):
|
|
fake_session(
|
|
{"profile": _FakeResp(404, {"errors": [{"message": "We could not find..."}]})}
|
|
)
|
|
out = api_mod._fetch_sync("Ghost", "origin")
|
|
assert out["error"] == "not_found"
|
|
assert "could not find" in out["detail"]
|
|
assert out["profile"] is None
|
|
|
|
|
|
def test_fetch_sync_maps_403_to_auth(api_mod, fake_session):
|
|
fake_session({"profile": _FakeResp(403, None, text="<html>blocked</html>")})
|
|
out = api_mod._fetch_sync("X", "origin")
|
|
assert out["error"] == "auth"
|
|
|
|
|
|
def test_fetch_sync_success_fills_all_segments(api_mod, fake_session):
|
|
fake_session(
|
|
{
|
|
"profile": _FakeResp(
|
|
200, {"data": {"platformInfo": {"platformUserHandle": "P"}}}
|
|
),
|
|
"segments/weapon": _FakeResp(200, {"data": [{"type": "weapon"}]}),
|
|
"segments/vehicle": _FakeResp(200, {"data": [{"type": "vehicle"}]}),
|
|
"segments/gadget": _FakeResp(200, {"data": [{"type": "gadget"}]}),
|
|
"segments/soldier": _FakeResp(200, {"data": [{"type": "soldier"}]}),
|
|
}
|
|
)
|
|
out = api_mod._fetch_sync("P", "origin")
|
|
assert out["error"] is None
|
|
assert out["profile"]["data"]["platformInfo"]["platformUserHandle"] == "P"
|
|
assert out["weapon"] == [{"type": "weapon"}]
|
|
assert out["gadget"] == [{"type": "gadget"}]
|
|
assert out["soldier"] == [{"type": "soldier"}]
|
|
|
|
|
|
def test_fetch_sync_segment_failure_degrades_not_aborts(api_mod, fake_session):
|
|
fake_session(
|
|
{
|
|
"profile": _FakeResp(200, {"data": {"platformInfo": {}}}),
|
|
"segments/weapon": _FakeResp(500, None, text="err"),
|
|
}
|
|
)
|
|
out = api_mod._fetch_sync("P", "origin")
|
|
assert out["error"] is None # profile 成功即算成功
|
|
assert out["weapon"] == [] # 失败的段退化为空,不拖垮整次取数
|
|
|
|
|
|
async def test_fetch_bf2042_direct_retries_once_on_auth(api_mod, monkeypatch):
|
|
calls: list[bool] = []
|
|
|
|
def _fake_fetch_sync(name, platform, *, force_refresh=False):
|
|
calls.append(force_refresh)
|
|
if not force_refresh:
|
|
return {**api_mod._empty_result(), "error": "auth", "detail": "HTTP 403"}
|
|
return {**api_mod._empty_result(), "profile": {"data": {}}}
|
|
|
|
monkeypatch.setattr(api_mod, "_fetch_sync", _fake_fetch_sync)
|
|
out = await api_mod.fetch_bf2042_direct("P", "origin")
|
|
assert calls == [False, True] # 鉴权失败 → 强制刷新重试一次
|
|
assert out["error"] is None
|
|
|
|
|
|
async def test_fetch_bf2042_direct_no_retry_on_success(api_mod, monkeypatch):
|
|
calls: list[bool] = []
|
|
|
|
def _fake_fetch_sync(name, platform, *, force_refresh=False):
|
|
calls.append(force_refresh)
|
|
return {**api_mod._empty_result(), "profile": {"data": {}}}
|
|
|
|
monkeypatch.setattr(api_mod, "_fetch_sync", _fake_fetch_sync)
|
|
out = await api_mod.fetch_bf2042_direct("P", "origin")
|
|
assert calls == [False]
|
|
assert out["error"] is None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# tracker_edge.fetch_bf2042_segments — 直连优先 / 桥兜底调度
|
|
# --------------------------------------------------------------------------- #
|
|
_DISPATCH_PKG = "bf_dispatch_testpkg"
|
|
|
|
|
|
@pytest.fixture
|
|
def dispatch(monkeypatch):
|
|
"""包骨架 + 真 tracker_edge + 假 tracker_api / tracker_bridge 子模块。"""
|
|
_make_pkg(_DISPATCH_PKG, _PLUGIN_DIR)
|
|
_make_pkg(f"{_DISPATCH_PKG}.services", _PLUGIN_DIR / "services")
|
|
fetchers_pkg = _make_pkg(
|
|
f"{_DISPATCH_PKG}.services.fetchers", _PLUGIN_DIR / "services" / "fetchers"
|
|
)
|
|
edge = _load(
|
|
f"{_DISPATCH_PKG}.services.fetchers.tracker_edge",
|
|
_PLUGIN_DIR / "services" / "fetchers" / "tracker_edge.py",
|
|
)
|
|
api = types.ModuleType(f"{_DISPATCH_PKG}.services.fetchers.tracker_api")
|
|
bridge = types.ModuleType(f"{_DISPATCH_PKG}.services.fetchers.tracker_bridge")
|
|
sys.modules[api.__name__] = api
|
|
sys.modules[bridge.__name__] = bridge
|
|
# 直接挂到父包上,让 edge 里 `from . import tracker_api` 稳定命中
|
|
fetchers_pkg.tracker_api = api
|
|
fetchers_pkg.tracker_bridge = bridge
|
|
|
|
# 关掉节流(测试里不等 3s/20s)
|
|
monkeypatch.setattr(edge, "_DIRECT_MIN_INTERVAL_S", 0.0)
|
|
monkeypatch.setattr(edge, "_MIN_INTERVAL_S", 0.0)
|
|
monkeypatch.setattr(edge, "_last_direct_monotonic", 0.0)
|
|
monkeypatch.setattr(edge, "_last_fetch_monotonic", 0.0)
|
|
return edge, api, bridge
|
|
|
|
|
|
def _direct_ok(**over):
|
|
base = {
|
|
"profile": {"data": {"platformInfo": {}}},
|
|
"weapon": [{"type": "weapon"}],
|
|
"vehicle": [],
|
|
"gadget": [{"type": "gadget"}],
|
|
"soldier": [],
|
|
"error": None,
|
|
"detail": "",
|
|
}
|
|
base.update(over)
|
|
return base
|
|
|
|
|
|
async def test_dispatch_direct_ok_never_touches_bridge(dispatch):
|
|
edge, api, bridge = dispatch
|
|
|
|
async def _direct(name, platform):
|
|
return _direct_ok()
|
|
|
|
async def _bridge(**_kw):
|
|
raise AssertionError("直连成功时不应回退扩展桥")
|
|
|
|
api.fetch_bf2042_direct = _direct
|
|
bridge.fetch_via_extension = _bridge
|
|
out = await edge.fetch_bf2042_segments("P", "origin")
|
|
assert out["error"] is None
|
|
assert out["weapon"] == [{"type": "weapon"}]
|
|
assert out["gadget"] == [{"type": "gadget"}]
|
|
|
|
|
|
async def test_dispatch_not_found_is_final(dispatch):
|
|
edge, api, bridge = dispatch
|
|
|
|
async def _direct(name, platform):
|
|
return _direct_ok(
|
|
profile=None,
|
|
weapon=[],
|
|
gadget=[],
|
|
error="not_found",
|
|
detail="no such player",
|
|
)
|
|
|
|
async def _bridge(**_kw):
|
|
raise AssertionError("not_found 是确定性结果,不应回退扩展桥")
|
|
|
|
api.fetch_bf2042_direct = _direct
|
|
bridge.fetch_via_extension = _bridge
|
|
out = await edge.fetch_bf2042_segments("Ghost", "origin")
|
|
assert out["error"] == "not_found"
|
|
assert out["detail"] == "no such player"
|
|
|
|
|
|
async def test_dispatch_falls_back_to_bridge_on_direct_error(dispatch):
|
|
edge, api, bridge = dispatch
|
|
calls = []
|
|
|
|
async def _direct(name, platform):
|
|
return _direct_ok(
|
|
profile=None, weapon=[], gadget=[], error="error", detail="HTTP 500"
|
|
)
|
|
|
|
async def _bridge(**kw):
|
|
calls.append(kw)
|
|
return (
|
|
{"profile": {"data": {"platformInfo": {}}}, "weapon": {"data": [1]}},
|
|
None,
|
|
"",
|
|
)
|
|
|
|
api.fetch_bf2042_direct = _direct
|
|
bridge.fetch_via_extension = _bridge
|
|
out = await edge.fetch_bf2042_segments("P", "origin")
|
|
assert len(calls) == 1
|
|
assert out["error"] is None
|
|
assert out["weapon"] == [1]
|
|
|
|
|
|
async def test_dispatch_merges_details_when_both_fail(dispatch):
|
|
edge, api, bridge = dispatch
|
|
|
|
async def _direct(name, platform):
|
|
return _direct_ok(
|
|
profile=None, weapon=[], gadget=[], error="auth", detail="令牌失效"
|
|
)
|
|
|
|
async def _bridge(**kw):
|
|
return {}, "timeout", "等待页面数据超时"
|
|
|
|
api.fetch_bf2042_direct = _direct
|
|
bridge.fetch_via_extension = _bridge
|
|
out = await edge.fetch_bf2042_segments("P", "origin")
|
|
assert out["error"] == "timeout"
|
|
assert "等待页面数据超时" in out["detail"]
|
|
assert "令牌失效" in out["detail"]
|