132 lines
4.8 KiB
Python
132 lines
4.8 KiB
Python
"""core/huya/anon_device.py 的协议装配与解析测试（mock 网络，不打真实请求）。"""
|
||
|
||
from unittest.mock import Mock
|
||
|
||
import pytest
|
||
import requests
|
||
|
||
from core.huya.anon_device import (
|
||
ANON_LOGIN_URL,
|
||
AnonymousDeviceError,
|
||
fetch_anonymous_device_state,
|
||
)
|
||
|
||
|
||
def _fake_response(status=200, headers=None, text="", json_body=None):
|
||
resp = Mock(spec=requests.Response)
|
||
resp.status_code = status
|
||
resp.headers = headers or {}
|
||
resp.text = text
|
||
resp.json = Mock(return_value=json_body) if json_body is not None else Mock(
|
||
side_effect=ValueError("bad json")
|
||
)
|
||
return resp
|
||
|
||
|
||
def _fake_anon_body(uid="1471255709907", biztoken="A" * 344):
|
||
return {
|
||
"uri": 10014,
|
||
"returnCode": 0,
|
||
"message": "success",
|
||
"data": {"uid": uid, "biztoken": biztoken},
|
||
}
|
||
|
||
|
||
def test_fetches_four_fields_from_middle_and_anon(monkeypatch):
|
||
session = Mock(spec=requests.Session)
|
||
deviceid = "w_1147246591246680065"
|
||
session.get.return_value = _fake_response(
|
||
headers={"Set-Cookie": f"udb_deviceid={deviceid}; Max-Age=315360000; Domain=huya.com"}
|
||
)
|
||
session.post.return_value = _fake_response(json_body=_fake_anon_body())
|
||
monkeypatch.setattr("core.huya.anon_device._new_session", lambda **kw: session)
|
||
|
||
state = fetch_anonymous_device_state(session=session)
|
||
|
||
assert state["udb_deviceid"] == deviceid
|
||
assert state["udb_anouid"] == "1471255709907"
|
||
assert state["udb_anobiztoken"] == "A" * 344
|
||
assert len(state["udb_guiddata"]) == 32
|
||
|
||
# middle 请求装配：URL 含 8 位随机数与 32hex guiddata
|
||
get_url = session.get.call_args.args[0]
|
||
parts = get_url.split("/")
|
||
assert len(parts[-1]) == 32
|
||
assert len(parts[-3]) == 8
|
||
assert parts[-3].isdigit()
|
||
assert state["udb_guiddata"] == parts[-1]
|
||
|
||
# anonymousLogin 装配：必须带 middle Referer；body 含随机 context/csid
|
||
post_kwargs = session.post.call_args.kwargs
|
||
assert session.post.call_args.args[0] == ANON_LOGIN_URL
|
||
assert post_kwargs["headers"]["Referer"] == get_url
|
||
import json as _json
|
||
|
||
payload = _json.loads(post_kwargs["data"])
|
||
assert payload["context"].startswith("WB-")
|
||
assert payload["sdid"].startswith("csid_")
|
||
|
||
|
||
def test_values_are_random_across_calls(monkeypatch):
|
||
session = Mock(spec=requests.Session)
|
||
session.get.return_value = _fake_response(
|
||
headers={"Set-Cookie": "udb_deviceid=w_1147246591246680065; Max-Age=1"}
|
||
)
|
||
session.post.return_value = _fake_response(json_body=_fake_anon_body(uid="1"))
|
||
monkeypatch.setattr("core.huya.anon_device._new_session", lambda **kw: session)
|
||
|
||
import json as _json
|
||
|
||
a = fetch_anonymous_device_state(session=session)
|
||
b = fetch_anonymous_device_state(session=session)
|
||
|
||
# 客户端随机部分：guiddata(middle URL) 与 anonymousLogin 的 context/csid
|
||
assert a["udb_guiddata"] != b["udb_guiddata"]
|
||
post_a = _json.loads(session.post.call_args_list[0].kwargs["data"])
|
||
post_b = _json.loads(session.post.call_args_list[1].kwargs["data"])
|
||
assert post_a["context"] != post_b["context"]
|
||
assert post_a["sdid"] != post_b["sdid"]
|
||
|
||
|
||
def test_bad_middle_status_raises(monkeypatch):
|
||
session = Mock(spec=requests.Session)
|
||
session.get.return_value = _fake_response(status=403)
|
||
monkeypatch.setattr("core.huya.anon_device._new_session", lambda **kw: session)
|
||
|
||
with pytest.raises(AnonymousDeviceError):
|
||
fetch_anonymous_device_state(session=session)
|
||
|
||
|
||
def test_missing_deviceid_in_set_cookie_raises(monkeypatch):
|
||
session = Mock(spec=requests.Session)
|
||
session.get.return_value = _fake_response(headers={"Set-Cookie": "PHPSESSID=x"})
|
||
monkeypatch.setattr("core.huya.anon_device._new_session", lambda **kw: session)
|
||
|
||
with pytest.raises(AnonymousDeviceError, match="udb_deviceid"):
|
||
fetch_anonymous_device_state(session=session)
|
||
|
||
|
||
def test_anon_error_return_code_raises(monkeypatch):
|
||
session = Mock(spec=requests.Session)
|
||
session.get.return_value = _fake_response(
|
||
headers={"Set-Cookie": "udb_deviceid=w_1147246591246680065; Max-Age=1"}
|
||
)
|
||
session.post.return_value = _fake_response(
|
||
json_body={"uri": 10014, "returnCode": 108, "message": "blocked"}
|
||
)
|
||
monkeypatch.setattr("core.huya.anon_device._new_session", lambda **kw: session)
|
||
|
||
with pytest.raises(AnonymousDeviceError, match="returnCode=108"):
|
||
fetch_anonymous_device_state(session=session)
|
||
|
||
|
||
def test_non_json_anon_body_raises(monkeypatch):
|
||
session = Mock(spec=requests.Session)
|
||
session.get.return_value = _fake_response(
|
||
headers={"Set-Cookie": "udb_deviceid=w_1147246591246680065; Max-Age=1"}
|
||
)
|
||
session.post.return_value = _fake_response(text="error!")
|
||
monkeypatch.setattr("core.huya.anon_device._new_session", lambda **kw: session)
|
||
|
||
with pytest.raises(AnonymousDeviceError, match="非 JSON"):
|
||
fetch_anonymous_device_state(session=session) |