live-hub-py/tests/test_weibo_collab.py

"""微博外部单协作核心流程测试。"""

from __future__ import annotations

from sqlalchemy.orm import Session

from web.backend.models import AuditLog, User, WeiboAccount, WeiboExternalOrderMetric
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.check_balance_groups(db, user, [first], threshold=100)
    collab.set_status(db, user, first, collab.STATUS_EXCHANGE_PENDING)
    collab.set_status(db, user, second, collab.STATUS_EXCHANGE_PENDING)

    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_EXCHANGE_PENDING
    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"])