live-hub-py/tests/test_weibo_accounts.py

"""微博 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, WeiboWorkbenchAccount
from web.backend.permissions import ROLE_PERMISSIONS
from web.backend.routers import weibo as weibo_router
from web.backend.routers.weibo import (
    add_workbench_accounts,
    assignments_summary,
    batch_assign_accounts,
    batch_tag_accounts,
    copy_account_ck,
    create_bind_qrcodes,
    disable_account,
    enable_account,
    get_account,
    import_cookies,
    list_account_ids,
    list_account_tags,
    list_accounts,
    list_workbench_accounts,
    list_workbench_pool,
    list_workbench_tags,
    refresh_account,
    remove_workbench_accounts,
    update_account,
)
from web.backend.schemas import (
    WeiboAccountBatchTagRequest,
    WeiboAccountDisableRequest,
    WeiboAccountUpdate,
    WeiboBatchAssignRequest,
    WeiboCookieImport,
    WeiboExchangeRequest,
    WeiboPaymentStatusRequest,
    WeiboRechargeRequest,
    WeiboWorkbenchAccountsRequest,
)
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_once_and_updates_existing_only_when_requested(db, current_user):
    first = import_cookies(
        WeiboCookieImport(text="SCF=a; SUB=b", tag="运营"),
        db=db,
        current=current_user,
    )
    duplicate = import_cookies(
        WeiboCookieImport(text="SCF=c; SUB=d", tag="新标签"),
        db=db,
        current=current_user,
    )

    assert first.count == 1
    assert duplicate.count == 0
    assert duplicate.skipped == 1
    assert duplicate.accounts == []
    assert db.query(WeiboAccount).count() == 1
    account = db.query(WeiboAccount).one()
    assert account.uid == "4014000001"
    assert account.cookie == "SCF=a; SUB=b"
    assert account.tag == "运营"

    updated = import_cookies(
        WeiboCookieImport(
            text="SCF=c; SUB=d",
            tag="新标签",
            update_existing=True,
        ),
        db=db,
        current=current_user,
    )

    assert updated.count == 1
    assert updated.skipped == 0
    assert db.query(WeiboAccount).count() == 1
    account = db.query(WeiboAccount).one()
    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 = updated.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_occupied_workbench_account_cannot_be_updated_by_another_user(db, current_user):
    imported_out = import_cookies(
        WeiboCookieImport(text="SCF=a; SUB=owner"),
        db=db,
        current=current_user,
    ).accounts[0]
    imported = db.query(WeiboAccount).filter(WeiboAccount.id == imported_out.id).one()
    original_cookie = imported.cookie
    owner = User(
        username="weibo-owner",
        password_hash="hash",
        role="support",
        is_active=True,
    )
    other = User(
        username="weibo-other",
        password_hash="hash",
        role="support",
        is_active=True,
    )
    db.add_all([owner, other])
    db.commit()
    db.add(
        WeiboWorkbenchAccount(
            user_id=owner.id,
            account_id=imported.id,
        )
    )
    db.commit()

    result = import_cookies(
        WeiboCookieImport(
            text="SCF=z; SUB=other",
            update_existing=True,
        ),
        db=db,
        current=other,
    )

    assert result.count == 0
    assert len(result.errors) == 1
    assert "其他人的工作台占用" in result.errors[0].message
    db.refresh(imported)
    assert imported.cookie == original_cookie


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()
    account = db.query(WeiboAccount).filter(WeiboAccount.id == account_out.id).one()
    account.assigned_to = support.id
    db.commit()

    support_result = copy_account_ck(account_out.id, db=db, current=support)
    assert support_result.ck_text == "SCF=a; SUB=b"

    denied = User(
        username="weibo-copy-denied",
        password_hash="hash",
        role="support",
        custom_permissions=[],
        is_active=True,
    )
    db.add(denied)
    db.commit()
    with pytest.raises(HTTPException) as exc_info:
        copy_account_ck(account_out.id, db=db, current=denied)

    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_support_role_has_weibo_operation_permissions_without_delete():
    support = set(ROLE_PERMISSIONS["support"])

    assert {
        "weibo:view",
        "weibo:workbench",
        "weibo:import",
        "weibo:manage",
        "weibo:account:ck_export",
        "weibo:bind",
        "weibo:recharge",
        "weibo:exchange",
        "weibo:collab:view",
        "weibo:collab:edit",
    } <= support
    assert "weibo:delete" not in support
    assert "weibo:collab:manage" not in support


def test_weibo_workbench_is_isolated_per_user(db, current_user):
    first = WeiboAccount(
        uid="4014000101",
        nickname="工作台账号A",
        cookie="SCF=a; SUB=first",
        tag="A",
    )
    second = WeiboAccount(
        uid="4014000102",
        nickname="工作台账号B",
        cookie="SCF=b; SUB=second",
        tag="B",
    )
    db.add_all([first, second])
    db.commit()
    user_a = User(
        username="weibo-workbench-a",
        password_hash="hash",
        role="support",
        is_active=True,
    )
    user_b = User(
        username="weibo-workbench-b",
        password_hash="hash",
        role="support",
        is_active=True,
    )
    db.add_all([user_a, user_b])
    db.commit()
    first.assigned_to = user_a.id
    second.assigned_to = user_b.id
    db.commit()

    add_workbench_accounts(
        WeiboWorkbenchAccountsRequest(account_ids=[first.id]),
        db=db,
        current=user_a,
    )
    pool_before = list_workbench_pool(page=1, page_size=20, db=db, current=user_a)
    assert pool_before["items"] == []
    with pytest.raises(HTTPException) as exc_info:
        add_workbench_accounts(
            WeiboWorkbenchAccountsRequest(account_ids=[first.id]),
            db=db,
            current=user_b,
        )
    assert exc_info.value.status_code == 403
    add_workbench_accounts(
        WeiboWorkbenchAccountsRequest(account_ids=[second.id]),
        db=db,
        current=user_b,
    )

    page_a = list_workbench_accounts(page=1, page_size=20, db=db, current=user_a)
    page_b = list_workbench_accounts(page=1, page_size=20, db=db, current=user_b)
    assert [item.uid for item in page_a["items"]] == [first.uid]
    assert [item.uid for item in page_b["items"]] == [second.uid]
    assert (
        list_workbench_accounts(tag="A", page=1, page_size=20, db=db, current=user_a)[
            "total"
        ]
        == 1
    )
    assert list_workbench_tags(db=db, current=user_a)["tags"] == ["A"]
    assert list_account_tags(db=db, current=current_user)["tags"] == ["A", "B"]
    assert db.query(WeiboWorkbenchAccount).count() == 2

    pool_a = list_workbench_pool(page=1, page_size=20, db=db, current=user_a)
    assert pool_a["items"] == []

    remove_workbench_accounts(
        WeiboWorkbenchAccountsRequest(account_ids=[first.id]),
        db=db,
        current=user_a,
    )
    assert (
        list_workbench_accounts(page=1, page_size=20, db=db, current=user_a)["total"]
        == 0
    )
    assert (
        list_workbench_accounts(page=1, page_size=20, db=db, current=user_b)["total"]
        == 1
    )
    pool_b = list_workbench_pool(page=1, page_size=20, db=db, current=user_b)
    assert pool_b["items"] == []

    pool_a = list_workbench_pool(page=1, page_size=20, db=db, current=user_a)
    assert [item.uid for item in pool_a["items"]] == [first.uid]


def test_batch_tag_updates_selected_accounts(db, current_user):
    first = WeiboAccount(
        uid="4014000201",
        nickname="批量标签A",
        cookie="SCF=a; SUB=first",
        tag="旧A",
    )
    second = WeiboAccount(
        uid="4014000202",
        nickname="批量标签B",
        cookie="SCF=b; SUB=second",
        tag="旧B",
    )
    db.add_all([first, second])
    db.commit()

    result = batch_tag_accounts(
        WeiboAccountBatchTagRequest(ids=[first.id, second.id], tag="客服组"),
        db=db,
        current=current_user,
    )

    assert result["updated"] == 2
    tags = {
        account.id: account.tag
        for account in db.query(WeiboAccount)
        .filter(WeiboAccount.id.in_([first.id, second.id]))
        .all()
    }
    assert tags == {first.id: "客服组", second.id: "客服组"}


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_weibo_account_scope_for_support_and_operator(db):
    support_a = User(
        username="weibo-scope-a",
        password_hash="hash",
        role="support",
        is_active=True,
    )
    support_b = User(
        username="weibo-scope-b",
        password_hash="hash",
        role="support",
        is_active=True,
    )
    operator = User(
        username="weibo-scope-operator",
        password_hash="hash",
        role="operation",
        is_active=True,
    )
    db.add_all([support_a, support_b, operator])
    db.commit()
    own = WeiboAccount(
        uid="4014000301",
        nickname="自己账号",
        cookie="SCF=a; SUB=own",
        tag="自己标签",
        assigned_to=support_a.id,
    )
    other = WeiboAccount(
        uid="4014000302",
        nickname="他人账号",
        cookie="SCF=b; SUB=other",
        tag="他人标签",
        assigned_to=support_b.id,
    )
    db.add_all([own, other])
    db.commit()

    own_page = list_accounts(page=1, page_size=20, db=db, current=support_a)
    assert own_page["total"] == 1
    own_dump = own_page["items"][0].model_dump()
    assert own_dump["uid"] == own.uid
    assert own_dump["assigned_to"] == support_a.id
    assert own_dump["assigned_username"] == support_a.username
    assert list_account_ids(db=db, current=support_a)["account_ids"] == [own.id]
    assert list_account_tags(db=db, current=support_a)["tags"] == ["自己标签"]

    with pytest.raises(HTTPException) as exc_info:
        get_account(other.id, db=db, current=support_a)
    assert exc_info.value.status_code == 404

    own.tag = "原标签"
    db.commit()
    with pytest.raises(HTTPException) as exc_info:
        batch_tag_accounts(
            WeiboAccountBatchTagRequest(ids=[own.id, other.id], tag="越权"),
            db=db,
            current=support_a,
        )
    assert exc_info.value.status_code == 404
    db.refresh(own)
    db.refresh(other)
    assert own.tag == "原标签"
    assert other.tag == "他人标签"

    operator_page = list_accounts(page=1, page_size=20, db=db, current=operator)
    assert operator_page["total"] == 2
    assert set(list_account_tags(db=db, current=operator)["tags"]) == {
        "他人标签",
        "原标签",
    }

    preview = assignments_summary(db=db, current=operator)
    assert preview["total_count"] == 2
    assert preview["assigned_count"] == 2
    assert preview["unassigned_count"] == 0
    counts = {item["id"]: item["assigned_count"] for item in preview["support_users"]}
    assert counts == {support_a.id: 1, support_b.id: 1}

    with pytest.raises(HTTPException) as exc_info:
        batch_assign_accounts(
            WeiboBatchAssignRequest(account_ids=[own.id], assigned_to=support_b.id),
            db=db,
            current=support_a,
        )
    assert exc_info.value.status_code == 403


def test_weibo_batch_assign_cleans_previous_workbench(db):
    operator = User(
        username="weibo-assign-operator",
        password_hash="hash",
        role="operation",
        is_active=True,
    )
    old_support = User(
        username="weibo-assign-old",
        password_hash="hash",
        role="support",
        is_active=True,
    )
    new_support = User(
        username="weibo-assign-new",
        password_hash="hash",
        role="support",
        is_active=True,
    )
    db.add_all([operator, old_support, new_support])
    db.commit()
    account = WeiboAccount(
        uid="4014000311",
        nickname="待改分配",
        cookie="SCF=a; SUB=b",
        assigned_to=old_support.id,
    )
    db.add(account)
    db.commit()
    add_workbench_accounts(
        WeiboWorkbenchAccountsRequest(account_ids=[account.id]),
        db=db,
        current=old_support,
    )

    result = batch_assign_accounts(
        WeiboBatchAssignRequest(
            account_ids=[account.id],
            assigned_to=new_support.id,
        ),
        db=db,
        current=operator,
    )

    assert result["updated"] == 1
    db.refresh(account)
    assert account.assigned_to == new_support.id
    assert (
        db.query(WeiboWorkbenchAccount)
        .filter(WeiboWorkbenchAccount.account_id == account.id)
        .count()
        == 0
    )
    add_workbench_accounts(
        WeiboWorkbenchAccountsRequest(account_ids=[account.id]),
        db=db,
        current=new_support,
    )
    audit = (
        db.query(AuditLog)
        .filter(AuditLog.action == "weibo:account:assign")
        .order_by(AuditLog.id.desc())
        .first()
    )
    assert audit is not None
    assert str(account.id) in audit.detail
    assert "SCF=a" not in audit.detail


def test_support_cannot_update_another_users_account_by_reimport(db):
    support_a = User(
        username="weibo-reimport-a",
        password_hash="hash",
        role="support",
        is_active=True,
    )
    support_b = User(
        username="weibo-reimport-b",
        password_hash="hash",
        role="support",
        is_active=True,
    )
    db.add_all([support_a, support_b])
    db.commit()
    account = WeiboAccount(
        uid="4014000321",
        nickname="他人 CK",
        cookie="SCF=old; SUB=owner",
        assigned_to=support_a.id,
    )
    db.add(account)
    db.commit()
    StubWeiboClient.snapshot = WeiboHomeSnapshot(
        uid=account.uid,
        nickname="他人 CK",
        is_bound=False,
        role_name="",
        can_rebind=True,
        rebind_remain_days=0,
        weidou_balance=100,
        bind_url=BIND_URL,
    )

    denied = import_cookies(
        WeiboCookieImport(
            text="SCF=new; SUB=attacker",
            update_existing=True,
        ),
        db=db,
        current=support_b,
    )

    assert denied.count == 0
    assert len(denied.errors) == 1
    assert "已分配给其他客服" in denied.errors[0].message
    db.refresh(account)
    assert account.cookie == "SCF=old; SUB=owner"
    assert account.assigned_to == support_a.id

    StubWeiboClient.snapshot = WeiboHomeSnapshot(
        uid="4014000322",
        nickname="新账号",
        is_bound=False,
        role_name="",
        can_rebind=True,
        rebind_remain_days=0,
        weidou_balance=100,
        bind_url=BIND_URL,
    )
    imported = import_cookies(
        WeiboCookieImport(text="SCF=new; SUB=own"),
        db=db,
        current=support_b,
    )
    assert imported.count == 1
    new_account = db.query(WeiboAccount).filter(WeiboAccount.uid == "4014000322").one()
    assert new_account.assigned_to == support_b.id


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