292 lines
8.4 KiB
Python
292 lines
8.4 KiB
Python
"""微博外部单协作核心流程测试。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from core.weibo import WeiboAuthError
|
|
from web.backend.models import AuditLog, User, WeiboAccount, WeiboExternalOrderMetric
|
|
from web.backend.routers import weibo_collab as weibo_collab_router
|
|
from web.backend.schemas import WeiboCollabCkExportOut
|
|
from web.backend.services import weibo_collab_service as collab
|
|
|
|
|
|
def make_account(db: Session, uid: str = "4014000001") -> WeiboAccount:
|
|
account = WeiboAccount(
|
|
uid=uid,
|
|
nickname="测试微博",
|
|
cookie="SCF=a; SUB=b",
|
|
tag="",
|
|
remark="",
|
|
activity="bowen_tjg",
|
|
is_enabled=True,
|
|
ck_check_status="valid",
|
|
is_bound=True,
|
|
game_role_name="角色A",
|
|
can_rebind=False,
|
|
rebind_remain_days=7,
|
|
weidou_balance=0,
|
|
)
|
|
db.add(account)
|
|
db.commit()
|
|
db.refresh(account)
|
|
return account
|
|
|
|
|
|
def make_user(db: Session) -> User:
|
|
user = User(username="weibo-collab", password_hash="hash", role="super_admin")
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user
|
|
|
|
|
|
def test_create_orders_groups_and_balance_check(monkeypatch, db):
|
|
user = make_user(db)
|
|
account = make_account(db)
|
|
|
|
class FakeClient:
|
|
def __init__(self, *_args, **_kwargs):
|
|
pass
|
|
|
|
def list_mall(self, *_args, **_kwargs):
|
|
return {"weidou_balance": 400, "list": []}
|
|
|
|
monkeypatch.setattr(collab, "WeiboClient", FakeClient)
|
|
|
|
first = collab.create_order(
|
|
db,
|
|
user,
|
|
order_no="WB-001",
|
|
shop="店铺A",
|
|
account_id=account.id,
|
|
goods_name="商品A",
|
|
required_weidou=100,
|
|
home_images=["proof/a.png"],
|
|
require_verified=False,
|
|
)
|
|
second = collab.create_order(
|
|
db,
|
|
user,
|
|
order_no="WB-002",
|
|
shop="店铺A",
|
|
account_id=account.id,
|
|
goods_name="商品B",
|
|
required_weidou=200,
|
|
require_verified=False,
|
|
)
|
|
|
|
assert first.status == collab.STATUS_TO_RECHARGE
|
|
assert second.status == collab.STATUS_TO_RECHARGE
|
|
assert first.game_role_name == "角色A"
|
|
|
|
result = collab.check_balance_groups(
|
|
db,
|
|
user,
|
|
[first],
|
|
threshold=100,
|
|
activity="bowen_tjg",
|
|
)
|
|
|
|
assert result["updated"] == 2
|
|
assert result["groups"][0]["balance"] == 400
|
|
assert result["groups"][0]["required_total"] == 300
|
|
assert result["groups"][0]["surplus"] == 100
|
|
db.refresh(first)
|
|
db.refresh(second)
|
|
assert first.status == collab.STATUS_RECHARGED
|
|
assert second.status == collab.STATUS_RECHARGED
|
|
assert first.recharged_at is not None
|
|
assert db.query(WeiboExternalOrderMetric).count() == 2
|
|
|
|
|
|
def test_exchange_record_unique_match_and_archive(monkeypatch, db):
|
|
user = make_user(db)
|
|
account = make_account(db)
|
|
first = collab.create_order(
|
|
db,
|
|
user,
|
|
order_no="WB-101",
|
|
shop="店铺A",
|
|
account_id=account.id,
|
|
goods_name="商品A",
|
|
required_weidou=100,
|
|
require_verified=False,
|
|
)
|
|
second = collab.create_order(
|
|
db,
|
|
user,
|
|
order_no="WB-102",
|
|
shop="店铺A",
|
|
account_id=account.id,
|
|
goods_name="商品B",
|
|
required_weidou=200,
|
|
require_verified=False,
|
|
)
|
|
|
|
class FakeClient:
|
|
def __init__(self, *_args, **_kwargs):
|
|
pass
|
|
|
|
def list_mall(self, *_args, **_kwargs):
|
|
return {"weidou_balance": 500, "list": []}
|
|
|
|
def list_exchange_records(self, *_args, **_kwargs):
|
|
return {
|
|
"list": [
|
|
{
|
|
"goods_name": "商品A",
|
|
"game_account": "角色A",
|
|
"cost_weidou": 100,
|
|
"exchange_time": "2099-01-01 00:00:00",
|
|
"reward_status": 2,
|
|
}
|
|
]
|
|
}
|
|
|
|
monkeypatch.setattr(collab, "WeiboClient", FakeClient)
|
|
collab.set_status(db, user, first, collab.STATUS_RECHARGED)
|
|
collab.set_status(db, user, second, collab.STATUS_RECHARGED)
|
|
|
|
result = collab.check_exchange_records(db, user, [first, second])
|
|
|
|
assert result["success"] == 1
|
|
assert result["unknown"] == 1
|
|
db.refresh(first)
|
|
db.refresh(second)
|
|
assert first.status == collab.STATUS_DONE
|
|
assert first.exchange_verdict == collab.VERDICT_SUCCESS
|
|
assert second.status == collab.STATUS_RECHARGED
|
|
assert second.exchange_verdict == collab.VERDICT_UNKNOWN
|
|
|
|
archived = collab.archive_orders(db, user, [first.id], reason="客户验收")
|
|
assert archived["archived"] == 1
|
|
db.refresh(first)
|
|
assert first.archived_at is not None
|
|
assert first.archive_reason == "客户验收"
|
|
|
|
timeline = collab.order_timeline(db, first.id)
|
|
assert timeline
|
|
assert any(item["action"] == "weibo:collab:archive" for item in timeline)
|
|
|
|
|
|
def test_export_cks_returns_plaintext_and_audits_without_ck(db):
|
|
user = make_user(db)
|
|
account = make_account(db)
|
|
order = collab.create_order(
|
|
db,
|
|
user,
|
|
order_no="WB-CK-001",
|
|
account_id=account.id,
|
|
goods_name="商品A",
|
|
required_weidou=100,
|
|
require_verified=False,
|
|
)
|
|
|
|
result = collab.export_cks(db, user, [order.id])
|
|
|
|
dumped = WeiboCollabCkExportOut.model_validate(result).model_dump()
|
|
assert dumped["total"] == 1
|
|
assert dumped["exported"] == 1
|
|
assert dumped["items"][0]["account_id"] == account.id
|
|
assert dumped["items"][0]["ck_text"] == "SCF=a; SUB=b"
|
|
audit = (
|
|
db.query(AuditLog).filter(AuditLog.action == "weibo:account:ck_export").one()
|
|
)
|
|
assert f'"order_id":{order.id}' in audit.detail
|
|
assert "SCF=a; SUB=b" not in audit.detail
|
|
|
|
|
|
def test_export_cks_exposes_skipped_orders(db):
|
|
user = make_user(db)
|
|
account = make_account(db)
|
|
valid = collab.create_order(
|
|
db,
|
|
user,
|
|
order_no="WB-CK-101",
|
|
account_id=account.id,
|
|
goods_name="商品A",
|
|
required_weidou=100,
|
|
require_verified=False,
|
|
)
|
|
unbound = collab.create_order(
|
|
db,
|
|
user,
|
|
order_no="WB-CK-102",
|
|
goods_name="商品B",
|
|
required_weidou=100,
|
|
require_verified=False,
|
|
)
|
|
|
|
result = collab.export_cks(db, user, [valid.id, unbound.id, 999999])
|
|
|
|
dumped = WeiboCollabCkExportOut.model_validate(result).model_dump()
|
|
assert dumped["total"] == 3
|
|
assert dumped["exported"] == 1
|
|
assert {item["id"] for item in dumped["skipped"]} == {unbound.id, 999999}
|
|
assert any("还没关联微博账号" in item["reason"] for item in dumped["skipped"])
|
|
|
|
|
|
def test_list_orders_filters_invalid_ck_and_summary_excludes_error(db):
|
|
user = make_user(db)
|
|
valid_account = make_account(db, uid="4014000101")
|
|
invalid_account = make_account(db, uid="4014000102")
|
|
error_account = make_account(db, uid="4014000103")
|
|
invalid_account.ck_check_status = "invalid"
|
|
error_account.ck_check_status = "error"
|
|
db.commit()
|
|
|
|
for index, account in enumerate(
|
|
(valid_account, invalid_account, error_account), start=1
|
|
):
|
|
collab.create_order(
|
|
db,
|
|
user,
|
|
order_no=f"WB-CK-FILTER-{index}",
|
|
account_id=account.id,
|
|
goods_name="商品A",
|
|
required_weidou=100,
|
|
require_verified=False,
|
|
)
|
|
|
|
invalid = collab.list_orders(
|
|
db,
|
|
page=1,
|
|
page_size=20,
|
|
ck_status="invalid",
|
|
)
|
|
|
|
assert invalid["total"] == 1
|
|
assert invalid["items"][0]["account_id"] == invalid_account.id
|
|
assert collab.summary(db)["ck_invalid"] == 1
|
|
|
|
|
|
def test_verify_batch_persists_ck_invalid_after_row_rollback(monkeypatch, db):
|
|
user = make_user(db)
|
|
account = make_account(db)
|
|
order = collab.create_order(
|
|
db,
|
|
user,
|
|
order_no="WB-CK-VERIFY-INVALID",
|
|
account_id=account.id,
|
|
goods_name="商品A",
|
|
required_weidou=100,
|
|
require_verified=False,
|
|
)
|
|
|
|
def fail_refresh(*_args, **_kwargs):
|
|
raise WeiboAuthError("CK 已过期")
|
|
|
|
monkeypatch.setattr(collab, "refresh_weibo_account", fail_refresh)
|
|
|
|
result = weibo_collab_router.verify_external_orders_batch(
|
|
[order.id],
|
|
db=db,
|
|
current=user,
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["results"][0]["ck_status"] == "invalid"
|
|
db.refresh(account)
|
|
assert account.ck_check_status == "invalid"
|