"""微博微豆基础客户端协议测试。""" 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, WeiboTransportError, 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 requests.exceptions.JSONDecodeError("not json", self.text, 0) 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_fetch_home_treats_sina_login_redirect_as_auth_error(): session = FakeSession( [ ( "" ) ] ) client = WeiboClient("SUB=expired", session=session) with pytest.raises(WeiboAuthError, match="重新登录微博"): client.fetch_home() def test_fetch_home_reports_non_json_endpoint_path(): session = FakeSession(["gateway error"]) client = WeiboClient("SUB=test", session=session) with pytest.raises(WeiboTransportError, match="/aj/mall/home"): 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)