680 lines
20 KiB
Python
680 lines
20 KiB
Python
"""微博 CK 导入与绑定二维码接口契约。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import ClassVar
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import text
|
|
|
|
from core.weibo import (
|
|
WeiboAuthError,
|
|
WeiboBindError,
|
|
WeiboBindQrCode,
|
|
WeiboExchangeResult,
|
|
WeiboHomeSnapshot,
|
|
WeiboPaymentStatus,
|
|
WeiboRechargeError,
|
|
WeiboRechargeOrder,
|
|
)
|
|
from web.backend.models import AuditLog, User, WeiboAccount
|
|
from web.backend.routers import weibo as weibo_router
|
|
from web.backend.routers.weibo import (
|
|
copy_account_ck,
|
|
create_bind_qrcodes,
|
|
disable_account,
|
|
enable_account,
|
|
get_account,
|
|
import_cookies,
|
|
list_accounts,
|
|
refresh_account,
|
|
update_account,
|
|
)
|
|
from web.backend.schemas import (
|
|
WeiboAccountDisableRequest,
|
|
WeiboAccountUpdate,
|
|
WeiboCookieImport,
|
|
WeiboExchangeRequest,
|
|
WeiboPaymentStatusRequest,
|
|
WeiboRechargeRequest,
|
|
)
|
|
from web.backend.services import weibo_service
|
|
|
|
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 StubWeiboClient:
|
|
snapshot = WeiboHomeSnapshot(
|
|
uid="4014000001",
|
|
nickname="测试微博",
|
|
is_bound=False,
|
|
role_name="",
|
|
can_rebind=True,
|
|
rebind_remain_days=0,
|
|
weidou_balance=100,
|
|
bind_url=BIND_URL,
|
|
)
|
|
scheme_errors: ClassVar[dict[str, str]] = {}
|
|
payment_error: ClassVar[WeiboRechargeError | None] = None
|
|
calls: ClassVar[list[tuple[str, str]]] = []
|
|
|
|
def __init__(self, cookie: str = "", **_kwargs):
|
|
self.cookie = cookie
|
|
|
|
def fetch_home(self, activity: str = ""):
|
|
self.calls.append(("home", activity))
|
|
if "expired" in self.cookie:
|
|
raise WeiboAuthError("未登陆")
|
|
return self.snapshot
|
|
|
|
def check_payment_access(self, activity: str = "", *, home=None):
|
|
_ = home
|
|
self.calls.append(("payment", activity))
|
|
if self.payment_error is not None:
|
|
raise self.payment_error
|
|
|
|
def generate_bind_scheme(self, bind_url: str, account_type: str):
|
|
self.calls.append(("scheme", account_type))
|
|
error = self.scheme_errors.get(account_type)
|
|
if error:
|
|
raise WeiboBindError(error)
|
|
prefix = (
|
|
"weixin://dl/business/?appid=wx-test"
|
|
if account_type == "wx"
|
|
else "mqqapi://miniapp/open?_mappid=qq-test"
|
|
)
|
|
return prefix
|
|
|
|
def generate_bind_qrcode(self, bind_url: str, account_type: str):
|
|
self.calls.append(("qrcode", account_type))
|
|
error = self.scheme_errors.get(account_type)
|
|
if error:
|
|
raise WeiboBindError(error)
|
|
mime = "png" if account_type == "wx" else "jpeg"
|
|
return WeiboBindQrCode(
|
|
image_src=f"data:image/{mime};base64,test-{account_type}",
|
|
token=f"token-{account_type}",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def current_user(db):
|
|
user = User(username="weibo-operator", password_hash="hash", role="super_admin")
|
|
db.add(user)
|
|
db.commit()
|
|
return user
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def stub_client(monkeypatch):
|
|
StubWeiboClient.calls = []
|
|
StubWeiboClient.scheme_errors = {}
|
|
StubWeiboClient.payment_error = None
|
|
StubWeiboClient.snapshot = WeiboHomeSnapshot(
|
|
uid="4014000001",
|
|
nickname="测试微博",
|
|
is_bound=False,
|
|
role_name="",
|
|
can_rebind=True,
|
|
rebind_remain_days=0,
|
|
weidou_balance=100,
|
|
bind_url=BIND_URL,
|
|
)
|
|
monkeypatch.setattr(weibo_service, "WeiboClient", StubWeiboClient)
|
|
|
|
|
|
def test_import_creates_encrypted_account_and_is_idempotent(db, current_user):
|
|
first = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag="运营"),
|
|
db=db,
|
|
current=current_user,
|
|
)
|
|
second = import_cookies(
|
|
WeiboCookieImport(text="SCF=c; SUB=d", tag="新标签"),
|
|
db=db,
|
|
current=current_user,
|
|
)
|
|
|
|
assert first.count == 1
|
|
assert second.count == 1
|
|
assert db.query(WeiboAccount).count() == 1
|
|
account = db.query(WeiboAccount).one()
|
|
assert account.uid == "4014000001"
|
|
assert account.cookie == "SCF=c; SUB=d"
|
|
assert account.tag == "新标签"
|
|
assert account.ck_check_status == "valid"
|
|
raw_cookie = db.execute(text("SELECT cookie FROM weibo_accounts")).scalar_one()
|
|
assert raw_cookie.startswith("enc:v1:")
|
|
assert "SUB=d" not in raw_cookie
|
|
|
|
dumped = second.accounts[0].model_dump()
|
|
assert dumped["uid"] == "4014000001"
|
|
assert dumped["is_bound"] is False
|
|
assert dumped["weidou_balance"] == 100
|
|
assert "cookie" not in dumped
|
|
|
|
|
|
def test_import_reports_invalid_ck_per_line_and_keeps_valid_account(db, current_user):
|
|
result = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=expired\nSCF=b; SUB=valid", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
)
|
|
|
|
assert result.count == 1
|
|
assert len(result.errors) == 1
|
|
assert result.errors[0].line == 1
|
|
assert "未登陆" in result.errors[0].message
|
|
assert db.query(WeiboAccount).count() == 1
|
|
|
|
|
|
def test_import_reports_payment_access_failure(db, current_user):
|
|
StubWeiboClient.payment_error = WeiboRechargeError("触发微博账号风控")
|
|
|
|
result = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
)
|
|
|
|
assert result.count == 0
|
|
assert len(result.errors) == 1
|
|
assert "触发微博账号风控" in result.errors[0].message
|
|
assert db.query(WeiboAccount).count() == 0
|
|
|
|
|
|
def test_refresh_marks_payment_access_failure(db, current_user):
|
|
account_out = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
).accounts[0]
|
|
account = db.query(WeiboAccount).filter(WeiboAccount.id == account_out.id).one()
|
|
StubWeiboClient.payment_error = WeiboRechargeError("触发微博账号风控")
|
|
|
|
with pytest.raises(WeiboRechargeError, match="触发微博账号风控"):
|
|
weibo_service.refresh_weibo_account(db, account)
|
|
|
|
assert account.ck_check_status == "error"
|
|
assert "触发微博账号风控" in account.ck_check_message
|
|
|
|
|
|
def test_refresh_marks_payment_login_expiry_as_invalid(db, current_user):
|
|
account_out = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
).accounts[0]
|
|
account = db.query(WeiboAccount).filter(WeiboAccount.id == account_out.id).one()
|
|
StubWeiboClient.payment_error = WeiboAuthError("支付域 retcode=6102")
|
|
|
|
with pytest.raises(WeiboAuthError, match="6102"):
|
|
weibo_service.refresh_weibo_account(db, account)
|
|
|
|
assert account.ck_check_status == "invalid"
|
|
assert "6102" in account.ck_check_message
|
|
|
|
|
|
def test_copy_account_ck_returns_plaintext_and_audits_without_ck(db, current_user):
|
|
account_out = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
).accounts[0]
|
|
|
|
result = copy_account_ck(account_out.id, db=db, current=current_user)
|
|
|
|
dumped = result.model_dump()
|
|
assert dumped["account_id"] == account_out.id
|
|
assert dumped["ck_text"] == "SCF=a; SUB=b"
|
|
audit = (
|
|
db.query(AuditLog).filter(AuditLog.action == "weibo:account:ck_export").one()
|
|
)
|
|
assert dumped["uid"] in audit.detail
|
|
assert "SUB=b" not in audit.detail
|
|
|
|
|
|
def test_copy_account_ck_uses_export_permission_defaults(db, current_user):
|
|
account_out = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
).accounts[0]
|
|
support = User(
|
|
username="weibo-copy-support",
|
|
password_hash="hash",
|
|
role="support",
|
|
is_active=True,
|
|
)
|
|
db.add(support)
|
|
db.commit()
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
copy_account_ck(account_out.id, db=db, current=support)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
|
|
operator = User(
|
|
username="weibo-copy-operator",
|
|
password_hash="hash",
|
|
role="operation",
|
|
is_active=True,
|
|
)
|
|
db.add(operator)
|
|
db.commit()
|
|
result = copy_account_ck(account_out.id, db=db, current=operator)
|
|
assert result.ck_text == "SCF=a; SUB=b"
|
|
|
|
|
|
def test_list_accounts_returns_parallel_page_shape(db, current_user):
|
|
assert (
|
|
import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
).count
|
|
== 1
|
|
)
|
|
|
|
page = list_accounts(page=1, page_size=20, db=db, current=current_user)
|
|
|
|
assert page["total"] == 1
|
|
dumped = page["items"][0].model_dump()
|
|
assert set(dumped) >= {
|
|
"id",
|
|
"uid",
|
|
"nickname",
|
|
"tag",
|
|
"remark",
|
|
"is_enabled",
|
|
"is_bound",
|
|
"can_rebind",
|
|
"rebind_remain_days",
|
|
"weidou_balance",
|
|
"ck_check_status",
|
|
}
|
|
|
|
|
|
def test_account_crud_and_lifecycle(db, current_user):
|
|
account_out = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag="初始"),
|
|
db=db,
|
|
current=current_user,
|
|
).accounts[0]
|
|
|
|
detail = get_account(
|
|
account_out.id,
|
|
db=db,
|
|
current=current_user,
|
|
).model_dump()
|
|
assert detail["is_enabled"] is True
|
|
assert detail["remark"] == ""
|
|
|
|
updated = update_account(
|
|
account_out.id,
|
|
WeiboAccountUpdate(tag="重点", remark="客户 A"),
|
|
db=db,
|
|
current=current_user,
|
|
).model_dump()
|
|
assert updated["tag"] == "重点"
|
|
assert updated["remark"] == "客户 A"
|
|
|
|
disabled = disable_account(
|
|
account_out.id,
|
|
WeiboAccountDisableRequest(reason="暂停使用"),
|
|
db=db,
|
|
current=current_user,
|
|
).model_dump()
|
|
assert disabled["is_enabled"] is False
|
|
assert disabled["disabled_reason"] == "暂停使用"
|
|
|
|
disabled_page = list_accounts(
|
|
enabled=False,
|
|
page=1,
|
|
page_size=20,
|
|
db=db,
|
|
current=current_user,
|
|
)
|
|
assert disabled_page["total"] == 1
|
|
assert disabled_page["items"][0].model_dump()["is_enabled"] is False
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
refresh_account(account_out.id, db=db, current=current_user)
|
|
assert exc_info.value.status_code == 409
|
|
|
|
disabled_actions = [
|
|
lambda: create_bind_qrcodes(account_out.id, db=db, current=current_user),
|
|
lambda: weibo_router.list_mall_goods(
|
|
account_out.id,
|
|
db=db,
|
|
current=current_user,
|
|
),
|
|
lambda: weibo_router.exchange_goods(
|
|
account_out.id,
|
|
WeiboExchangeRequest(goods_id=37, expected_price=100, confirm=True),
|
|
db=db,
|
|
current=current_user,
|
|
),
|
|
lambda: weibo_router.create_recharge_order(
|
|
account_out.id,
|
|
WeiboRechargeRequest(amount_yuan=1),
|
|
db=db,
|
|
current=current_user,
|
|
),
|
|
]
|
|
for action in disabled_actions:
|
|
with pytest.raises(HTTPException) as disabled_exc:
|
|
action()
|
|
assert disabled_exc.value.status_code == 409
|
|
|
|
import_cookies(
|
|
WeiboCookieImport(text="SCF=c; SUB=d", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
)
|
|
account = db.query(WeiboAccount).filter(WeiboAccount.id == account_out.id).one()
|
|
assert account.is_enabled is False
|
|
assert account.tag == "重点"
|
|
|
|
enabled = enable_account(
|
|
account_out.id,
|
|
db=db,
|
|
current=current_user,
|
|
).model_dump()
|
|
assert enabled["is_enabled"] is True
|
|
assert enabled["disabled_reason"] == ""
|
|
|
|
|
|
def test_bind_qrcodes_returns_both_channels_and_partial_error(db, current_user):
|
|
account = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
).accounts[0]
|
|
StubWeiboClient.scheme_errors = {"qq": "QQ Scheme 暂时不可用"}
|
|
|
|
result = create_bind_qrcodes(account.id, db=db, current=current_user)
|
|
|
|
dumped = result.model_dump()
|
|
assert dumped["account_id"] == account.id
|
|
assert dumped["is_bound"] is False
|
|
assert dumped["can_rebind"] is True
|
|
by_channel = {item["channel"]: item for item in dumped["schemes"]}
|
|
assert by_channel["wx"]["scheme"].startswith("weixin://")
|
|
assert by_channel["wx"]["error"] == ""
|
|
assert by_channel["qq"]["scheme"] == ""
|
|
assert by_channel["qq"]["error"] == "QQ Scheme 暂时不可用"
|
|
|
|
|
|
def test_bind_qrcodes_selected_channel_returns_scannable_image(db, current_user):
|
|
account = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
).accounts[0]
|
|
|
|
result = create_bind_qrcodes(
|
|
account.id,
|
|
channel="qq",
|
|
db=db,
|
|
current=current_user,
|
|
)
|
|
|
|
dumped = result.model_dump()
|
|
assert len(dumped["schemes"]) == 1
|
|
qq = dumped["schemes"][0]
|
|
assert qq["channel"] == "qq"
|
|
assert qq["qrcode_image"] == "data:image/jpeg;base64,test-qq"
|
|
assert qq["qrcode_token"] == "token-qq"
|
|
assert qq["scheme"].startswith("mqqapi://miniapp/open")
|
|
assert [call for call in StubWeiboClient.calls if call[0] == "qrcode"] == [
|
|
("qrcode", "qq")
|
|
]
|
|
|
|
|
|
def test_bind_qrcodes_rejects_bound_account_without_rebind(db, current_user):
|
|
account = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
).accounts[0]
|
|
StubWeiboClient.snapshot = WeiboHomeSnapshot(
|
|
uid=account.uid,
|
|
nickname=account.nickname,
|
|
is_bound=True,
|
|
role_name="测试角色",
|
|
can_rebind=False,
|
|
rebind_remain_days=7,
|
|
weidou_balance=100,
|
|
bind_url=BIND_URL,
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
create_bind_qrcodes(account.id, db=db, current=current_user)
|
|
|
|
assert exc_info.value.status_code == 409
|
|
assert "7 天" in str(exc_info.value.detail)
|
|
|
|
|
|
def test_manual_mall_exchange_and_point_routes_serialize(monkeypatch, db, current_user):
|
|
account = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
).accounts[0]
|
|
monkeypatch.setattr(
|
|
weibo_router,
|
|
"query_weibo_mall",
|
|
lambda *_args, **_kwargs: {
|
|
"weidou_balance": 120,
|
|
"list": [
|
|
{
|
|
"id": 37,
|
|
"goods_name": "会员1天卡",
|
|
"price": 100,
|
|
"remain_count": 99,
|
|
"is_sold_out": False,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
weibo_router,
|
|
"query_weibo_exchange_records",
|
|
lambda *_args, **_kwargs: {
|
|
"page": 1,
|
|
"pageSize": 20,
|
|
"totalCount": 1,
|
|
"list": [
|
|
{
|
|
"goods_name": "会员1天卡",
|
|
"game_account": "测试角色",
|
|
"cost_weidou": 100,
|
|
"exchange_time": "2026-11-07 18:00:00",
|
|
"reward_status": 2,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
weibo_router,
|
|
"query_weibo_point_history",
|
|
lambda *_args, **_kwargs: {
|
|
"page_num": 1,
|
|
"page_size": 20,
|
|
"total": 1,
|
|
"list": [
|
|
{
|
|
"id": 1,
|
|
"source_name": "V+助威",
|
|
"point_num": 100,
|
|
"source_id": "order-1",
|
|
}
|
|
],
|
|
},
|
|
)
|
|
|
|
mall = weibo_router.list_mall_goods(
|
|
account.id,
|
|
db=db,
|
|
current=current_user,
|
|
).model_dump()
|
|
exchange = weibo_router.list_exchange_records(
|
|
account.id,
|
|
db=db,
|
|
current=current_user,
|
|
).model_dump()
|
|
points = weibo_router.list_point_history(
|
|
account.id,
|
|
db=db,
|
|
current=current_user,
|
|
).model_dump()
|
|
|
|
assert mall["goods"][0]["goods_name"] == "会员1天卡"
|
|
assert mall["weidou_balance"] == 120
|
|
exchange_item = exchange["items"][0]
|
|
assert set(exchange_item) == {
|
|
"goods_name",
|
|
"game_account",
|
|
"cost_weidou",
|
|
"exchange_time",
|
|
"reward_status",
|
|
"raw",
|
|
}
|
|
assert exchange_item["game_account"] == "测试角色"
|
|
assert exchange_item["cost_weidou"] == 100
|
|
assert exchange_item["exchange_time"] == "2026-11-07 18:00:00"
|
|
assert exchange_item["reward_status"] == 2
|
|
assert "id" not in exchange_item
|
|
assert exchange["total"] == 1
|
|
assert points["items"][0]["source_id"] == "order-1"
|
|
|
|
|
|
def test_manual_recharge_order_and_status_routes_serialize(
|
|
monkeypatch, db, current_user
|
|
):
|
|
account = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
).accounts[0]
|
|
monkeypatch.setattr(
|
|
weibo_router,
|
|
"create_weibo_recharge_order",
|
|
lambda *_args, **_kwargs: WeiboRechargeOrder(
|
|
order_id="order-1",
|
|
pay_id="pay-1",
|
|
amount_yuan=1,
|
|
single_reward=100,
|
|
alipay_qr_code="https://qr.alipay.com/upx-test",
|
|
desktop_pay_url="https://pay.sc.weibo.com/pay/pc/comb?pay_id=pay-1",
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
weibo_router,
|
|
"query_weibo_payment_status",
|
|
lambda *_args, **_kwargs: WeiboPaymentStatus(
|
|
pay_id="pay-1",
|
|
out_pay_id="order-1",
|
|
status="PAY_STATUS_PENDING",
|
|
sub_status="PAY_SUB_STATUS_INITIAL",
|
|
),
|
|
)
|
|
|
|
order = weibo_router.create_recharge_order(
|
|
account.id,
|
|
WeiboRechargeRequest(amount_yuan=1),
|
|
db=db,
|
|
current=current_user,
|
|
).model_dump()
|
|
status = weibo_router.get_recharge_status(
|
|
account.id,
|
|
WeiboPaymentStatusRequest(pay_id="pay-1"),
|
|
db=db,
|
|
current=current_user,
|
|
).model_dump()
|
|
|
|
assert order["single_reward"] == 100
|
|
assert order["alipay_qr_code"] == "https://qr.alipay.com/upx-test"
|
|
assert status["status"] == "PAY_STATUS_PENDING"
|
|
|
|
|
|
def test_recharge_failure_keeps_valid_ck_status(monkeypatch, db, current_user):
|
|
account_out = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
).accounts[0]
|
|
account = db.query(WeiboAccount).filter(WeiboAccount.id == account_out.id).one()
|
|
|
|
def reject_recharge(*_args, **_kwargs):
|
|
raise WeiboRechargeError("微博支付页拒绝了当前账号")
|
|
|
|
monkeypatch.setattr(
|
|
StubWeiboClient,
|
|
"create_recharge_order",
|
|
reject_recharge,
|
|
raising=False,
|
|
)
|
|
|
|
with pytest.raises(WeiboRechargeError, match="支付页拒绝"):
|
|
weibo_service.create_weibo_recharge_order(
|
|
db,
|
|
account,
|
|
amount_yuan=1,
|
|
)
|
|
|
|
assert account.ck_check_status == "valid"
|
|
assert account.ck_check_message == ""
|
|
|
|
|
|
def test_manual_exchange_requires_confirm_and_serializes_result(
|
|
monkeypatch, db, current_user
|
|
):
|
|
account = import_cookies(
|
|
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
|
|
db=db,
|
|
current=current_user,
|
|
).accounts[0]
|
|
monkeypatch.setattr(
|
|
weibo_router,
|
|
"exchange_weibo_goods",
|
|
lambda *_args, **_kwargs: WeiboExchangeResult(
|
|
goods_id=37,
|
|
goods_name="微博会员1天卡",
|
|
price=100,
|
|
balance_before=200,
|
|
balance_after=100,
|
|
remain_before=10,
|
|
remain_after=9,
|
|
verified=True,
|
|
),
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
weibo_router.exchange_goods(
|
|
account.id,
|
|
WeiboExchangeRequest(goods_id=37, expected_price=100, confirm=False),
|
|
db=db,
|
|
current=current_user,
|
|
)
|
|
assert exc_info.value.status_code == 400
|
|
|
|
result = weibo_router.exchange_goods(
|
|
account.id,
|
|
WeiboExchangeRequest(goods_id=37, expected_price=100, confirm=True),
|
|
db=db,
|
|
current=current_user,
|
|
).model_dump()
|
|
|
|
assert result["goods_id"] == 37
|
|
assert result["balance_after"] == 100
|
|
assert result["verified"] is True
|