"""微博微豆基础客户端协议测试。"""
from __future__ import annotations
from typing import Any
import pytest
import requests
from core.weibo import (
WeiboApiError,
WeiboAuthError,
WeiboBindError,
WeiboBindQrCode,
WeiboClient,
WeiboExchangeError,
WeiboExchangeUncertainError,
WeiboHomeSnapshot,
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,
url: str = "",
status_code: int = 200,
headers: dict[str, str] | None = None,
):
self.payload = payload
self.text = payload if isinstance(payload, str) else ""
self.url = url
self.status_code = status_code
self.headers = {key.lower(): value for key, value in (headers or {}).items()}
def raise_for_status(self):
if self.status_code >= 400:
raise requests.HTTPError(f"HTTP {self.status_code}")
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
response = result if isinstance(result, FakeResponse) else FakeResponse(result)
if not response.url:
response.url = url
return response
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_generate_bind_qrcode_posts_channel_and_preserves_raw_query():
session = FakeSession(
[
{
"iRet": 0,
"sMsg": "ok",
"jData": {
"qrcode": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB",
"qrcodeToken": "token-1",
},
}
]
)
client = WeiboClient(session=session)
qrcode = client.generate_bind_qrcode(BIND_URL, "qq")
assert isinstance(qrcode, WeiboBindQrCode)
assert qrcode.image_src == (
"data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB"
)
assert qrcode.token == "token-1"
method, url, kwargs = session.calls[0]
assert method == "POST"
assert url == "https://livelinkbind.game.qq.com/api/qrcode/getMiniAppQrcode"
body = kwargs["json"]
raw_url = extract_bind_raw_url(BIND_URL)
assert body["queryString"].startswith(f"{raw_url}&sessionId=")
assert body["accountType"] == "qq"
assert kwargs["headers"]["Gap-Act-Id"] == "16722"
assert kwargs["headers"]["Gap-Plat-Id"] == "weiyouxi"
assert kwargs["headers"]["Gap-User-From"] == "pc"
assert "Cookie" not in kwargs["headers"]
def test_generate_bind_qrcode_supports_jpeg_and_server_rate_limit():
jpeg_session = FakeSession(
[
{
"iRet": 0,
"sMsg": "ok",
"jData": {"qrcode": "/9j/4AAQSkZJRgABAQ", "qrcodeToken": "token-2"},
}
]
)
qrcode = WeiboClient(session=jpeg_session).generate_bind_qrcode(BIND_URL, "wx")
assert qrcode.image_src == "data:image/jpeg;base64,/9j/4AAQSkZJRgABAQ"
rate_limited = FakeSession(
[{"iRet": -1140603, "sMsg": "刷新二维码频率过快，请间隔5秒后重试"}]
)
with pytest.raises(WeiboBindError, match="间隔5秒"):
WeiboClient(session=rate_limited).generate_bind_qrcode(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 = """
立即支付
"""
alipay_form_html = """
"""
alipay_html = """
"""
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"]
assert not any("/aj/pay/modular/prepare" in call[1] for call in session.calls)
def test_create_recharge_order_reports_payment_rejection():
client = WeiboClient("SCF=a; SUB=b", session=FakeSession([]))
def fake_context(
activity: str,
*,
home: WeiboHomeSnapshot | None = None,
) -> tuple[str, dict[str, str], dict[str, Any], dict[str, Any]]:
_ = (activity, home)
return (
"4014000001",
{"vuid": "vuid", "source_itemid": "20601", "F": "bowen_tjg"},
{"tab_type": "8", "cashier_id": "205"},
{
"tab_info": [
{
"is_current": 1,
"tab_type": "8",
"cashier_id": "205",
"menu": [
{
"subitemid": "2060101",
"itemid": 20601,
"appkey": 2000,
"commodity_id": "2060101",
"max_money": 1000,
}
],
}
]
},
)
def fake_bid(
vuid: str,
source_itemid: str,
tab_type: str,
activity: str,
) -> str:
_ = (vuid, source_itemid, tab_type, activity)
return "1000294191"
client._load_recharge_context = fake_context
client._recharge_bid = fake_bid
client._request_json = lambda *_args, **_kwargs: {
"code": 100000,
"data": {
"url": "https://pay.sc.weibo.com/api/merchant/pay/cashier?sign=test",
"orderid": "order-1",
},
}
def rejected_page(*_args, **_kwargs):
client._last_text_url = "https://pay.sc.weibo.com/sorry"
return "404错误"
client._request_text = rejected_page
with pytest.raises(WeiboRechargeError, match=r"支付页拒绝.*/sorry"):
client.create_recharge_order(amount_yuan=1, activity="bowen_tjg")
@pytest.mark.parametrize(
("location", "message"),
[
(
"https://pay.sc.weibo.com/sso/login.php?url=test",
"微博支付未登录",
),
(
"https://weibo.com/sorry?usernotexists",
"支付账号不存在",
),
],
)
def test_recharge_preflight_detects_payment_auth_redirect(location, message):
home = WeiboHomeSnapshot(
uid="4014000001",
nickname="测试微博",
is_bound=False,
role_name="",
can_rebind=True,
rebind_remain_days=0,
weidou_balance=100,
bind_url=BIND_URL,
recharge_url=(
"https://vipclub.weibo.com/vmember/subscribe"
"?vuid=8455793145&source_itemid=20601&F=bowen_tjg"
),
)
session = FakeSession(
[
{"code": 100000, "data": {}},
{
"code": 100000,
"data": {
"cashier_id": 205,
"tab_info": [
{
"cashier_id": "205",
"tab_id": "536",
"tab_type": "8",
"is_current": 1,
}
],
},
},
{
"code": 100000,
"data": {
"tab_info": [
{
"cashier_id": "205",
"tab_id": "536",
"tab_type": "8",
"is_current": 1,
"menu": [
{
"price": 6000,
"weibo_unified_params": {
"seller_id": 6329048632,
"modular_id": "7d5d81dfb1",
"default_cashier": "ali_cashier",
},
}
],
}
]
},
},
FakeResponse(
"",
status_code=302,
headers={"Location": location},
),
]
)
client = WeiboClient("SCF=a; SUB=b", session=session)
with pytest.raises(WeiboRechargeError, match=message):
client.check_payment_access("bowen_tjg", home=home)
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)