531 lines
16 KiB
Python
531 lines
16 KiB
Python
"""微博微豆基础客户端协议测试。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
from core.weibo import (
|
|
WeiboApiError,
|
|
WeiboAuthError,
|
|
WeiboBindError,
|
|
WeiboClient,
|
|
WeiboExchangeError,
|
|
WeiboExchangeUncertainError,
|
|
WeiboRechargeError,
|
|
extract_bind_raw_url,
|
|
normalize_weibo_cookie,
|
|
)
|
|
|
|
BIND_URL = (
|
|
"sinaweibo://wbdiversion?username=gh_test"
|
|
"&path=%2Fpages%2FgameAccountBind%2Findex%3FactId%3D16722"
|
|
"%26code%3Dabc%252Bdef%253D%26gameId%3Dcjm"
|
|
"%26livePlatId%3Dweiyouxi%26nonce%3Dabc123"
|
|
"%26sig%3D0123456789abcdef%26t%3D1794046086%26v%3D2.0"
|
|
)
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, payload):
|
|
self.payload = payload
|
|
self.text = payload if isinstance(payload, str) else ""
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
if not isinstance(self.payload, dict):
|
|
raise TypeError("not json")
|
|
return self.payload
|
|
|
|
|
|
class FakeSession(requests.Session):
|
|
def __init__(self, responses):
|
|
super().__init__()
|
|
self.responses = list(responses)
|
|
self.calls = []
|
|
|
|
def request(self, method, url, **_kwargs):
|
|
kwargs = _kwargs
|
|
self.calls.append((method, url, kwargs))
|
|
result = self.responses.pop(0)
|
|
if isinstance(result, Exception):
|
|
raise result
|
|
return FakeResponse(result)
|
|
|
|
|
|
def test_normalize_cookie_accepts_prefix_and_multiline():
|
|
raw = "Cookie: SCF=a;\n SUB=b;SUBP=c\n"
|
|
assert normalize_weibo_cookie(raw) == "SCF=a; SUB=b; SUBP=c"
|
|
|
|
|
|
def test_extract_bind_raw_url_preserves_inner_encoding():
|
|
raw = extract_bind_raw_url(BIND_URL)
|
|
assert raw.startswith("actId=16722&code=abc%2Bdef%3D")
|
|
assert "faceUrl" not in raw
|
|
# 内层 code 的多重编码不能被再次解开
|
|
assert "abc+def=" not in raw
|
|
|
|
|
|
def test_extract_bind_raw_url_rejects_wrong_path():
|
|
with pytest.raises(WeiboBindError, match="绑定小程序路径"):
|
|
extract_bind_raw_url("sinaweibo://wbdiversion?path=%2Fwrong")
|
|
|
|
|
|
def test_fetch_home_returns_basic_snapshot():
|
|
session = FakeSession(
|
|
[
|
|
{
|
|
"code": 100000,
|
|
"msg": "success",
|
|
"data": {
|
|
"user_info": {"uid": "4014000001", "nickname": "测试微博"},
|
|
"game_role": {
|
|
"is_bound": False,
|
|
"role_name": "",
|
|
"can_rebind": True,
|
|
"rebind_remain_days": 0,
|
|
"bind_url": BIND_URL,
|
|
},
|
|
"weidou_balance": {"balance": 100},
|
|
"task_list": [
|
|
{
|
|
"button_type": "cashier",
|
|
"action_url": (
|
|
"https://vipclub.weibo.com/vmember/subscribe"
|
|
"?vuid=8455793145&source_itemid=20601&F=bowen_tjg"
|
|
),
|
|
}
|
|
],
|
|
},
|
|
}
|
|
]
|
|
)
|
|
client = WeiboClient("SCF=a; SUB=b", session=session)
|
|
|
|
snapshot = client.fetch_home("bowen_tjg")
|
|
|
|
assert snapshot.uid == "4014000001"
|
|
assert snapshot.nickname == "测试微博"
|
|
assert snapshot.is_bound is False
|
|
assert snapshot.can_rebind is True
|
|
assert snapshot.weidou_balance == 100
|
|
assert snapshot.activity == "bowen_tjg"
|
|
assert "source_itemid=20601" in snapshot.recharge_url
|
|
method, url, kwargs = session.calls[0]
|
|
assert method == "GET"
|
|
assert "F=bowen_tjg" in url
|
|
assert "Cookie" not in kwargs["headers"]
|
|
assert session.cookies.get("SUB", domain=".weibo.com") == "b"
|
|
|
|
|
|
def test_fetch_home_marks_auth_error():
|
|
session = FakeSession([{"code": 400002, "msg": "未登陆"}])
|
|
client = WeiboClient("SUB=expired", session=session)
|
|
|
|
with pytest.raises(WeiboAuthError, match="未登陆"):
|
|
client.fetch_home()
|
|
|
|
|
|
def test_fetch_home_keeps_trace_id_for_other_api_error():
|
|
session = FakeSession(
|
|
[{"code": 100003, "msg": "活动异常", "trace_id": "trace-001"}]
|
|
)
|
|
client = WeiboClient("SUB=test", session=session)
|
|
|
|
with pytest.raises(WeiboApiError, match="trace-001"):
|
|
client.fetch_home()
|
|
|
|
|
|
def test_generate_bind_scheme_posts_account_type_without_weibo_cookie():
|
|
session = FakeSession(
|
|
[
|
|
{
|
|
"iRet": 0,
|
|
"sMsg": "ok",
|
|
"jData": {
|
|
"urlScheme": "weixin://dl/business/?appid=wx-test&query=scheme_id%3D1"
|
|
},
|
|
"apiName": "GenMiniappScheme",
|
|
}
|
|
]
|
|
)
|
|
client = WeiboClient(session=session)
|
|
|
|
scheme = client.generate_bind_scheme(BIND_URL, "wx")
|
|
|
|
assert scheme.startswith("weixin://dl/business/")
|
|
method, url, kwargs = session.calls[0]
|
|
assert method == "POST"
|
|
assert "c=Sdk&a=GenMiniappScheme&actId=16722" in url
|
|
assert kwargs["json"] == {"accountType": "wx"}
|
|
assert "Cookie" not in kwargs["headers"]
|
|
|
|
|
|
def test_generate_bind_scheme_rejects_missing_url_scheme():
|
|
session = FakeSession([{"iRet": 0, "sMsg": "ok", "jData": {}}])
|
|
client = WeiboClient(session=session)
|
|
|
|
with pytest.raises(WeiboBindError, match="urlScheme"):
|
|
client.generate_bind_scheme(BIND_URL, "qq")
|
|
|
|
|
|
def test_mall_exchange_and_point_queries():
|
|
session = FakeSession(
|
|
[
|
|
{
|
|
"code": 100000,
|
|
"data": {
|
|
"list": [{"id": 37, "goods_name": "会员1天卡", "price": 100}],
|
|
"weidou_balance": 100,
|
|
},
|
|
},
|
|
{
|
|
"code": 100000,
|
|
"data": {
|
|
"page": 1,
|
|
"pageSize": 20,
|
|
"totalCount": 0,
|
|
"list": [],
|
|
},
|
|
},
|
|
{
|
|
"code": 100000,
|
|
"data": {
|
|
"page_num": 1,
|
|
"page_size": 20,
|
|
"total": 1,
|
|
"list": [{"id": 1, "source_name": "V+助威", "point_num": 100}],
|
|
},
|
|
},
|
|
]
|
|
)
|
|
client = WeiboClient("SUB=valid", session=session)
|
|
|
|
mall = client.list_mall("bowen_tjg")
|
|
records = client.list_exchange_records("bowen_tjg")
|
|
points = client.list_point_history("bowen_tjg")
|
|
|
|
assert mall["weidou_balance"] == 100
|
|
assert records["totalCount"] == 0
|
|
assert points["total"] == 1
|
|
assert session.calls[0][2]["params"]["F"] == "bowen_tjg"
|
|
|
|
|
|
def test_create_recharge_order_full_manual_chain():
|
|
home = {
|
|
"code": 100000,
|
|
"data": {
|
|
"user_info": {"uid": "4014000001", "nickname": "测试微博"},
|
|
"game_role": {
|
|
"is_bound": True,
|
|
"role_name": "测试角色",
|
|
"can_rebind": False,
|
|
"rebind_remain_days": 7,
|
|
"bind_url": BIND_URL,
|
|
},
|
|
"weidou_balance": {"balance": 0},
|
|
"task_list": [
|
|
{
|
|
"button_type": "cashier",
|
|
"action_url": (
|
|
"https://vipclub.weibo.com/vmember/subscribe"
|
|
"?vuid=8455793145&source_itemid=20601&F=bowen_tjg"
|
|
),
|
|
}
|
|
],
|
|
},
|
|
}
|
|
userinfo = {
|
|
"code": 100000,
|
|
"data": {
|
|
"cashier_id": 205,
|
|
"tab_info": [
|
|
{
|
|
"cashier_id": "205",
|
|
"tab_id": "536",
|
|
"tab_type": "8",
|
|
"is_current": 1,
|
|
}
|
|
],
|
|
},
|
|
}
|
|
cashierinfo = {
|
|
"code": 100000,
|
|
"data": {
|
|
"tab_info": [
|
|
{
|
|
"cashier_id": "205",
|
|
"tab_id": "536",
|
|
"tab_type": "8",
|
|
"is_current": 1,
|
|
"menu": [
|
|
{
|
|
"commodity_id": "2060101",
|
|
"subitemid": "2060101",
|
|
"itemid": 20601,
|
|
"appkey": 2000,
|
|
"max_money": 1000,
|
|
"discount_info": {"desc": "1元=100微豆"},
|
|
}
|
|
],
|
|
}
|
|
]
|
|
},
|
|
}
|
|
privilege = {
|
|
"code": 100000,
|
|
"data": {
|
|
"privilege": {
|
|
"url": "https://vipclub.weibo.com/appreciate/description?bid=1000294191"
|
|
}
|
|
},
|
|
}
|
|
createorder = {
|
|
"code": 100000,
|
|
"msg": "下单成功",
|
|
"data": {
|
|
"url": "https://pay.sc.weibo.com/api/merchant/pay/cashier?sign=test",
|
|
"orderid": "26111000000000000001",
|
|
},
|
|
}
|
|
cashier_html = """
|
|
<input type="hidden" value="1220351218153064346" node-type="pay_id" name="pay_id">
|
|
<img node-type="QRCode" src="https://qr.weibo.cn/inf/gen?sign=test">
|
|
<a href="https://pay.sc.weibo.com/pay/pc/comb?pay_id=1220351218153064346">立即支付</a>
|
|
"""
|
|
alipay_form_html = """
|
|
<form action="https://mapi.alipay.com/gateway.do" method="GET">
|
|
<input name="out_trade_no" value="1020351218229058766">
|
|
<input name="total_fee" value="1.00">
|
|
</form>
|
|
"""
|
|
alipay_html = """
|
|
<input name="qrCode" type="hidden" value="https://qr.alipay.com/upx-test" id="J_qrCode" />
|
|
"""
|
|
session = FakeSession(
|
|
[
|
|
home,
|
|
{"code": 100000, "data": {}},
|
|
userinfo,
|
|
cashierinfo,
|
|
privilege,
|
|
createorder,
|
|
cashier_html,
|
|
alipay_form_html,
|
|
alipay_html,
|
|
]
|
|
)
|
|
client = WeiboClient("SCF=a; SUB=b", session=session)
|
|
|
|
order = client.create_recharge_order(amount_yuan=1, activity="bowen_tjg")
|
|
|
|
assert order.order_id == "26111000000000000001"
|
|
assert order.pay_id == "1220351218153064346"
|
|
assert order.single_reward == 100
|
|
assert order.alipay_qr_code == "https://qr.alipay.com/upx-test"
|
|
assert order.desktop_pay_url.startswith("https://pay.sc.weibo.com/")
|
|
create_call = next(
|
|
call for call in session.calls if call[1].endswith("/paycomponent/createorder")
|
|
)
|
|
assert create_call[2]["data"]["single_reward"] == "100"
|
|
assert '"bid":"1000294191"' in create_call[2]["data"]["ext"]
|
|
assert create_call[2]["params"] == create_call[2]["data"]
|
|
|
|
|
|
def test_create_recharge_order_rejects_missing_dynamic_menu():
|
|
home = {
|
|
"code": 100000,
|
|
"data": {
|
|
"user_info": {"uid": "1", "nickname": "测试"},
|
|
"game_role": {
|
|
"is_bound": False,
|
|
"role_name": "",
|
|
"can_rebind": True,
|
|
"rebind_remain_days": 0,
|
|
"bind_url": BIND_URL,
|
|
},
|
|
"weidou_balance": {"balance": 0},
|
|
"task_list": [
|
|
{
|
|
"button_type": "cashier",
|
|
"action_url": (
|
|
"https://vipclub.weibo.com/vmember/subscribe"
|
|
"?vuid=8455793145&source_itemid=20601&F=bowen_tjg"
|
|
),
|
|
}
|
|
],
|
|
},
|
|
}
|
|
userinfo = {
|
|
"code": 100000,
|
|
"data": {
|
|
"cashier_id": 205,
|
|
"tab_info": [
|
|
{
|
|
"cashier_id": "205",
|
|
"tab_id": "536",
|
|
"tab_type": "8",
|
|
"is_current": 1,
|
|
}
|
|
],
|
|
},
|
|
}
|
|
cashierinfo = {
|
|
"code": 100000,
|
|
"data": {
|
|
"tab_info": [
|
|
{
|
|
"cashier_id": "205",
|
|
"tab_id": "536",
|
|
"tab_type": "8",
|
|
"is_current": 1,
|
|
"menu": [],
|
|
}
|
|
]
|
|
},
|
|
}
|
|
session = FakeSession([home, {"code": 100000, "data": {}}, userinfo, cashierinfo])
|
|
client = WeiboClient("SUB=valid", session=session)
|
|
|
|
with pytest.raises(WeiboRechargeError, match="自定义微豆充值档位"):
|
|
client.create_recharge_order(amount_yuan=1)
|
|
|
|
|
|
def test_get_payment_status():
|
|
session = FakeSession(
|
|
[
|
|
{
|
|
"code": "100000",
|
|
"msg": "操作成功",
|
|
"data": {
|
|
"pay_id": "pay-1",
|
|
"out_pay_id": "out-1",
|
|
"status": "PAY_STATUS_PENDING",
|
|
"sub_status": "PAY_SUB_STATUS_INITIAL",
|
|
},
|
|
}
|
|
]
|
|
)
|
|
client = WeiboClient("SUB=valid", session=session)
|
|
|
|
status = client.get_payment_status("pay-1")
|
|
|
|
assert status.status == "PAY_STATUS_PENDING"
|
|
assert session.calls[0][2]["data"]["pay_id"] == "pay-1"
|
|
|
|
|
|
def test_exchange_goods_executes_once_and_verifies_balance_and_stock():
|
|
before = {
|
|
"code": 100000,
|
|
"data": {
|
|
"weidou_balance": 200,
|
|
"list": [
|
|
{
|
|
"id": 37,
|
|
"goods_name": "微博会员1天卡",
|
|
"price": 100,
|
|
"remain_count": 10,
|
|
"is_sold_out": False,
|
|
}
|
|
],
|
|
},
|
|
}
|
|
after = {
|
|
"code": 100000,
|
|
"data": {
|
|
"weidou_balance": 100,
|
|
"list": [
|
|
{
|
|
"id": 37,
|
|
"goods_name": "微博会员1天卡",
|
|
"price": 100,
|
|
"remain_count": 9,
|
|
"is_sold_out": False,
|
|
}
|
|
],
|
|
},
|
|
}
|
|
session = FakeSession(
|
|
[
|
|
before,
|
|
{"code": 100000, "msg": "success", "data": None},
|
|
after,
|
|
]
|
|
)
|
|
client = WeiboClient("SUB=valid", session=session)
|
|
|
|
result = client.exchange_goods(
|
|
37,
|
|
expected_price=100,
|
|
activity="bowen_tjg",
|
|
)
|
|
|
|
assert result.goods_name == "微博会员1天卡"
|
|
assert result.balance_before == 200
|
|
assert result.balance_after == 100
|
|
assert result.remain_before == 10
|
|
assert result.remain_after == 9
|
|
assert result.verified is True
|
|
exchange_calls = [
|
|
call for call in session.calls if call[1].endswith("/game/exchangegoods")
|
|
]
|
|
assert len(exchange_calls) == 1
|
|
assert exchange_calls[0][2]["params"]["goods_id"] == 37
|
|
|
|
|
|
def test_exchange_goods_rejects_insufficient_balance_before_write():
|
|
session = FakeSession(
|
|
[
|
|
{
|
|
"code": 100000,
|
|
"data": {
|
|
"weidou_balance": 99,
|
|
"list": [
|
|
{
|
|
"id": 37,
|
|
"goods_name": "微博会员1天卡",
|
|
"price": 100,
|
|
"remain_count": 10,
|
|
"is_sold_out": False,
|
|
}
|
|
],
|
|
},
|
|
}
|
|
]
|
|
)
|
|
client = WeiboClient("SUB=valid", session=session)
|
|
|
|
with pytest.raises(WeiboExchangeError, match="余额不足"):
|
|
client.exchange_goods(37, expected_price=100)
|
|
|
|
assert not any(call[1].endswith("/game/exchangegoods") for call in session.calls)
|
|
|
|
|
|
def test_exchange_goods_marks_transport_failure_as_uncertain():
|
|
session = FakeSession(
|
|
[
|
|
{
|
|
"code": 100000,
|
|
"data": {
|
|
"weidou_balance": 200,
|
|
"list": [
|
|
{
|
|
"id": 37,
|
|
"goods_name": "微博会员1天卡",
|
|
"price": 100,
|
|
"remain_count": 10,
|
|
"is_sold_out": False,
|
|
}
|
|
],
|
|
},
|
|
},
|
|
requests.ConnectionError("connection reset"),
|
|
]
|
|
)
|
|
client = WeiboClient("SUB=valid", session=session)
|
|
|
|
with pytest.raises(WeiboExchangeUncertainError, match="结果未知"):
|
|
client.exchange_goods(37, expected_price=100)
|