live-hub-py/tests/test_huya_collab_orders.py

"""虎牙外部 CK 协作台账：CRUD / 两套状态机 / 审计 / 图片 / 批量导入。

设计见 ``docs/设计-虎牙外部CK协作台账.md``。

用例里的 CK 都是**合成**的（形态对齐真机：分号后无空格、重复 ``guid``、
超长 ``udb_biztoken``/``udb_cred``）——真实凭据绝不写进仓库，
一旦提交就永久留在 git 历史里。
"""

from __future__ import annotations

import json
from types import SimpleNamespace

import pytest

from core.huya.taf_protocol import TafOutputStream, TafType
from tests.support.db import get_test_engine, make_test_session
from web.backend.huya_defaults import (
    HUYA_DEFAULT_BIND_ACT_ID,
    HUYA_DEFAULT_OUTER_ACT_ID,
)
from web.backend.models import (
    AuditLog,
    HuyaAccount,
    HuyaConfig,
    HuyaExternalOrder,
    HuyaWorkbenchAccount,
    User,
)
from web.backend.permissions import ROLE_PERMISSIONS
from web.backend.routers.huya import collab as collab_api
from web.backend.schemas import HuyaExternalOrderListOut, HuyaExternalOrderUpdate
from web.backend.services import huya_collab_service as collab
from web.backend.services.huya_collab_events import collab_events


def fake_ck(uid: str, *, web: bool = True) -> str:
    """合成一份形态对齐真机的虎牙 CK。

    ``web=True`` 模拟浏览器里导出的网页 CK（带 guid/_qimei_uuid42，且真机里
    ``guid`` 会重复出现）；``web=False`` 模拟 App 原始凭据（没有网页设备字段）。
    """
    parts = [
        f"udb_uid={uid}",
        f"udb_passport=hy_{uid}",
        "udb_version=1.0",
        "huya_ua=webh5&0.0.1&websocket&&h5_/30596253",
        f"hdid={'a' * 40}",
        f"username=hy_{uid}",
        f"yyuid={uid}",
        "udb_passdata=3",
    ]
    if web:
        # 真机 CK 里 guid 确实重复出现两次，归一化后应只剩一个。
        parts += [f"guid={'b' * 32}", f"guid={'b' * 32}"]
        parts.append("_qimei_uuid42=qc_123456")
    parts += [f"udb_biztoken={'T' * 200}", f"udb_cred={'C' * 152}", "udb_status=1"]
    return ";".join(parts)  # 注意：真机是分号后无空格


class TestCollabBase:
    def setup_method(self):
        self.engine = get_test_engine()
        self.db = make_test_session(self.engine)
        self.user = User(username="collab_op", password_hash="h", role="operation")
        self.other = User(username="collab_support", password_hash="h", role="support")
        self.db.add_all([self.user, self.other])
        self.db.commit()

    def teardown_method(self):
        self.db.close()

    def _create(
        self, order_no: str = "2627501604891357", **kwargs
    ) -> HuyaExternalOrder:
        return collab.create_order(self.db, self.user, order_no=order_no, **kwargs)

    def _audits(self, order: HuyaExternalOrder) -> list[AuditLog]:
        return (
            self.db.query(AuditLog)
            .filter(AuditLog.target == f"huya_external_order:{order.id}")
            .order_by(AuditLog.id.asc())
            .all()
        )

    def _bind_ready(self, order: HuyaExternalOrder) -> HuyaExternalOrder:
        """把一行补到「待充值」：CK 有效 + 绑定账号/角色 + 礼包 + 主页截图。

        2026-10-19 新口径：资料齐全且 CK 有效后由服务层自动推进到 `to_recharge`，
        不再存在 `pending` / `bound` 两个人工状态。
        """
        if not order.gift_name:
            collab.update_order(
                self.db,
                self.user,
                order,
                {"gift_name": "不朽伊人大礼包", "required_points": 8000},
            )
        collab.set_images(
            self.db,
            self.user,
            order,
            "home_images",
            ["collab/2026/10/16/home01.png"],
        )
        collab.set_ck(self.db, self.user, order, fake_ck("1199647239697"))
        collab.verify_ck(
            self.db,
            self.user,
            order,
            client=FakeProbeClient(
                FakeBindResp(
                    nick="微信用户", role="吃鸡199706", plat="苹果(iOS)", area="微信"
                )
            ),
        )
        return order


class TestCollabCrud(TestCollabBase):
    def test_create_parses_ck_and_writes_audit(self):
        order = self._create(ck_text=fake_ck("1199647239697"), shop="佳佳")

        assert order.uid == "1199647239697"
        assert order.yyuid == "1199647239697"
        assert order.hy_username == "hy_1199647239697"
        assert order.status == collab.STATUS_INCOMPLETE
        assert order.ck_status == "unverified"
        assert order.version == 0
        assert order.created_by == self.user.id

        audits = self._audits(order)
        assert [row.action for row in audits] == ["huya:collab:create"]

    def test_create_without_ck_leaves_uid_empty(self):
        order = self._create(ck_text="")
        assert order.uid is None

    def test_create_without_owner_defaults_to_the_creator(self):
        """负责人不填 = 当前登录用户（新建时不用选）。"""
        order = self._create(order_no="OWNER1")
        assert order.owner_id == self.user.id

    def test_create_can_still_assign_someone_else(self):
        order = self._create(order_no="OWNER2", owner_id=self.other.id)
        assert order.owner_id == self.other.id

    def test_multiple_rows_without_ck_are_allowed(self):
        """uid 唯一但可空：多个 NULL 不冲突，所以「先建单、后贴 CK」走得通。"""
        self._create(order_no="A1")
        self._create(order_no="A2")
        assert self.db.query(HuyaExternalOrder).count() == 2

    def test_create_rejects_duplicate_order_no(self):
        self._create(order_no="DUP1")
        with pytest.raises(collab.CollabError) as excinfo:
            self._create(order_no="DUP1")
        assert excinfo.value.status_code == 409
        assert excinfo.value.code == "order_no_conflict"

    def test_same_uid_can_serve_multiple_orders(self):
        """2026-10-16 用户拍板**放开** uid 唯一性：同一个虎牙号可以挂多单。

        原来这里是 409（``uid_conflict``）—— 但真机上"同一个买家买两件"常常继续用同一份 CK，
        挡了就变成第二单建不了。现在只把「这个号还挂在哪些单上」记进创建审计。
        """
        first = self._create(order_no="A1", ck_text=fake_ck("1001"))
        second = self._create(order_no="A2", ck_text=fake_ck("1001"))

        assert first.uid == second.uid == "1001"
        assert self.db.query(HuyaExternalOrder).count() == 2
        assert "A1" in str(self._audits(second)[-1].detail)  # 复用痕迹留在审计里

    def test_uid_used_orders_lists_other_rows(self):
        self._create(order_no="A3", ck_text=fake_ck("2001"))
        second = self._create(order_no="A4", ck_text=fake_ck("2001"))

        assert [row.order_no for row in collab.uid_used_orders(self.db, "2001")] == [
            "A3",
            "A4",
        ]
        assert [
            row.order_no
            for row in collab.uid_used_orders(self.db, "2001", exclude_id=second.id)
        ] == ["A3"]
        assert collab.uid_used_orders(self.db, "") == []
        assert collab.uids_in_use(self.db, "2001")[0]["order_no"] == "A3"

    def test_web_ck_does_not_fill_raw_cookie(self):
        """网页 CK（有 guid）不塞进 raw_cookie —— 它是 App 原始凭据的专用位。"""
        order = self._create(ck_text=fake_ck("1001", web=True))
        assert order.cookie
        assert order.raw_cookie == ""

    def test_app_ck_also_kept_as_raw_cookie_for_rebuild(self):
        order = self._create(ck_text=fake_ck("1002", web=False))
        assert order.cookie
        assert order.raw_cookie == order.cookie

    def test_update_returns_field_diff_and_bumps_version(self):
        order = self._create(shop="佳佳")
        changes = collab.update_order(
            self.db, self.user, order, {"shop": "小鱼儿", "remark": "熟人单"}
        )

        assert set(changes) == {"shop", "remark"}
        assert changes["shop"] == {"old": "佳佳", "new": "小鱼儿"}
        assert order.version == 1
        assert order.updated_by == self.user.id

        actions = [row.action for row in self._audits(order)]
        assert actions[-1] == "huya:collab:update"

    def test_update_only_touches_submitted_fields(self):
        order = self._create(shop="佳佳", gift_name="不朽伊人大礼包")
        collab.update_order(self.db, self.user, order, {"remark": "只改备注"})
        assert order.shop == "佳佳"
        assert order.gift_name == "不朽伊人大礼包"

    def test_update_rejects_stale_version(self):
        """乐观锁：别人先改过就 409，绝不静默覆盖（这正是 Excel 最痛的点）。"""
        order = self._create(shop="佳佳")
        collab.update_order(self.db, self.user, order, {"shop": "小鱼儿"})

        with pytest.raises(collab.CollabError) as excinfo:
            collab.update_order(
                self.db,
                self.other,
                order,
                {"remark": "并发写"},
                expected_version=0,
            )
        assert excinfo.value.status_code == 409
        assert excinfo.value.code == "version_conflict"
        assert order.remark == ""

    def test_update_accepts_matching_version(self):
        order = self._create(shop="佳佳")
        collab.update_order(self.db, self.user, order, {"shop": "小鱼儿"})
        collab.update_order(
            self.db, self.other, order, {"remark": "跟版本走"}, expected_version=1
        )
        assert order.remark == "跟版本走"
        assert order.version == 2

    def test_update_rejects_order_no_taken_by_another_row(self):
        self._create(order_no="A1")
        order = self._create(order_no="A2")
        with pytest.raises(collab.CollabError) as excinfo:
            collab.update_order(self.db, self.user, order, {"order_no": "A1"})
        assert excinfo.value.code == "order_no_conflict"

    def test_rejects_overlong_field_instead_of_silently_truncating(self):
        with pytest.raises(collab.CollabError) as excinfo:
            self._create(order_no="A1", remark="x" * 600)
        assert excinfo.value.code == "field_too_long"

    def test_delete_keeps_audit_row(self):
        order = self._create(order_no="DEL1")
        order_id = order.id
        snapshot = collab.delete_order(self.db, self.user, order)

        assert snapshot["order_no"] == "DEL1"
        assert self.db.query(HuyaExternalOrder).count() == 0
        actions = [
            row.action
            for row in self.db.query(AuditLog)
            .filter(AuditLog.target == f"huya_external_order:{order_id}")
            .all()
        ]
        assert "huya:collab:delete" in actions


class TestCollabStatusMachine(TestCollabBase):
    def test_forward_jump_is_allowed_but_backward_needs_rollback(self):
        """人工切换允许**往前跳步**，回退必须显式 rollback。"""
        order = self._bind_ready(self._create())
        assert order.status == collab.STATUS_TO_RECHARGE

        # 待充值 → 已上脚本（跳过「已充值」）是允许的
        collab.set_status(self.db, self.user, order, "scripted")
        assert order.status == "scripted"
        # 跳步时中间阶段的时间戳补上，别让时间线断档
        assert order.bound_at is not None
        assert order.recharged_at is not None
        assert order.scripted_at is not None

        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_status(self.db, self.user, order, "to_recharge")
        assert excinfo.value.code == "status_not_forward"
        assert order.status == "scripted"

    def test_to_recharge_requires_complete_customer_info_and_valid_ck(self):
        order = self._create()

        # 空单：直接往前会被资料完整性挡住，错误信息必须逐项说清
        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_status(self.db, self.user, order, collab.STATUS_TO_RECHARGE)
        message = excinfo.value.detail
        assert "外部 CK" in message
        assert "绑定账号" in message
        assert "绑定角色" in message
        assert "礼包名" in message
        assert "主页截图" in message

        # 只贴 CK + 读到角色，但没礼包 / 主页截图 → 仍然停在待补资料
        collab.set_ck(self.db, self.user, order, fake_ck("1199647001234"))
        collab.verify_ck(
            self.db,
            self.user,
            order,
            client=FakeProbeClient(FakeBindResp(nick="微信用户", role="吃鸡199706")),
        )
        assert order.status == collab.STATUS_INCOMPLETE
        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_status(self.db, self.user, order, collab.STATUS_TO_RECHARGE)
        assert "礼包名" in excinfo.value.detail
        assert "主页截图" in excinfo.value.detail
        assert "绑定图" not in excinfo.value.detail  # 绑定图已停用，不该再被要求

        # 资料补齐后由 set_images 自动推进（不要求用户再点一次状态）
        collab.update_order(
            self.db,
            self.user,
            order,
            {"gift_name": "不朽伊人大礼包", "required_points": 8000},
        )
        assert order.status == collab.STATUS_INCOMPLETE
        collab.set_images(
            self.db, self.user, order, "home_images", ["collab/2026/10/16/home01.png"]
        )
        assert order.status == collab.STATUS_TO_RECHARGE
        assert order.bound_at is not None

    def test_auto_advance_waits_for_valid_ck(self):
        order = self._create(
            order_no="AUTO-CK",
            gift_name="不朽伊人大礼包",
            required_points=8000,
            home_images=["collab/2026/10/16/home01.png"],
        )
        collab.set_ck(self.db, self.user, order, fake_ck("1199647001234"))
        # 只贴 CK 不验证：资料看似齐全，但 CK 仍是 unverified，不能进待充值。
        # 绑定快照通常由探针给；这里直接塞进 ORM 对象，单独验证「CK 必须有效」这条。
        order.bind_game_account = "微信用户"
        order.bind_game_role = "吃鸡199706"
        self.db.commit()
        assert order.ck_status == "unverified"
        assert order.status == collab.STATUS_INCOMPLETE
        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_status(self.db, self.user, order, collab.STATUS_TO_RECHARGE)
        assert "CK 验证有效" in excinfo.value.detail

        collab.verify_ck(
            self.db,
            self.user,
            order,
            client=FakeProbeClient(FakeBindResp(nick="微信用户", role="吃鸡199706")),
        )
        assert order.status == collab.STATUS_TO_RECHARGE

    def test_forward_flow_sets_timestamps(self):
        order = self._bind_ready(self._create())
        assert order.status == collab.STATUS_TO_RECHARGE

        collab.set_status(
            self.db, self.user, order, collab.STATUS_RECHARGED, allow_recharge=True
        )
        collab.set_status(self.db, self.user, order, "scripted")
        collab.set_status(self.db, self.user, order, "done")

        assert order.status == "done"
        assert order.bound_at is not None
        assert order.recharged_at is not None
        assert order.scripted_at is not None
        assert order.done_at is not None

    def test_every_status_change_writes_audit(self):
        """设计文档 §4.3 防漂移：状态写入必须留痕，漏一条就是 bug。"""
        order = self._bind_ready(self._create())
        before = len(self._audits(order))
        collab.set_status(
            self.db, self.user, order, collab.STATUS_RECHARGED, allow_recharge=True
        )
        for target in ("scripted", "done"):
            collab.set_status(self.db, self.user, order, target)
        after = self._audits(order)
        assert len(after) - before == 3
        status_audits = [row for row in after if row.action == "huya:collab:status"]
        assert len(status_audits) == 3

    def test_rollback_requires_flag_and_clears_later_timestamps(self):
        order = self._bind_ready(self._create())
        collab.set_status(
            self.db, self.user, order, collab.STATUS_RECHARGED, allow_recharge=True
        )
        for target in ("scripted", "done"):
            collab.set_status(self.db, self.user, order, target)
        assert order.scripted_at is not None

        # 不给 rollback 标记 → 拒
        with pytest.raises(collab.CollabError):
            collab.set_status(
                self.db,
                self.user,
                order,
                collab.STATUS_RECHARGED,
                allow_recharge=True,
            )

        collab.set_status(
            self.db,
            self.user,
            order,
            collab.STATUS_RECHARGED,
            rollback=True,
            allow_recharge=True,
        )
        assert order.status == collab.STATUS_RECHARGED
        # 回退后后面阶段的时间戳必须清掉，否则状态与时间戳自相矛盾
        assert order.scripted_at is None
        assert order.done_at is None
        assert order.recharged_at is not None

    def test_void_requires_reason_and_blocks_returning(self):
        order = self._create()
        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_status(self.db, self.user, order, "void")
        assert excinfo.value.code == "field_required"

        collab.set_status(self.db, self.user, order, "void", reason="客户退款")
        assert order.status == "void"
        assert order.void_reason == "客户退款"

        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_status(self.db, self.user, order, "incomplete", rollback=True)
        assert excinfo.value.code == "status_from_void"

    def test_same_status_is_noop(self):
        order = self._bind_ready(self._create())
        before = len(self._audits(order))
        assert (
            collab.set_status(self.db, self.user, order, collab.STATUS_TO_RECHARGE)
            == {}
        )
        # 幂等点击不该刷出审计噪声
        assert len(self._audits(order)) == before

    def test_direct_recharged_and_points_error_require_recharge_flow(self):
        order = self._bind_ready(self._create())
        for target in (collab.STATUS_RECHARGED, collab.STATUS_POINTS_ERROR):
            with pytest.raises(collab.CollabError) as excinfo:
                collab.set_status(self.db, self.user, order, target)
            assert excinfo.value.code == "recharge_check_required"
        assert order.status == collab.STATUS_TO_RECHARGE

    def test_done_has_no_next_status(self):
        order = self._create()
        collab.set_status(self.db, self.user, order, "void", reason="x")
        assert collab.NEXT_STATUS.get("done") is None


class TestCollabCkLifecycle(TestCollabBase):
    def test_valid_sets_checked_at_but_disabled_does_not(self):
        order = self._create(ck_text=fake_ck("1001"))

        collab.set_ck_status(self.db, self.user, order, "valid")
        assert order.ck_status == "valid"
        assert order.ck_checked_at is not None
        checked_at = order.ck_checked_at
        assert order.ck_check_message == ""

        collab.set_ck_status(self.db, self.user, order, "disabled", message="不再使用")
        assert order.ck_status == "disabled"
        # 人工停用不是「刚跑过探针」，时间戳不该被刷
        assert order.ck_checked_at == checked_at

    def test_invalid_records_reason(self):
        order = self._create(ck_text=fake_ck("1001"))
        collab.set_ck_status(
            self.db, self.user, order, "invalid", message="登录态已过期"
        )
        assert order.ck_status == "invalid"
        assert order.ck_check_message == "登录态已过期"

    def test_ck_status_rejects_unknown_value(self):
        order = self._create()
        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_ck_status(self.db, self.user, order, "whatever")
        assert excinfo.value.code == "ck_status_unknown"

    def test_pasting_new_ck_resets_verification(self):
        order = self._create(ck_text=fake_ck("1001"))
        collab.set_ck_status(self.db, self.user, order, "valid")
        collab.set_ck(self.db, self.user, order, fake_ck("1002"))
        assert order.uid == "1002"
        assert order.ck_status == "unverified"
        assert order.ck_checked_at is None

    def test_ck_audit_never_leaks_the_credential(self):
        """审计 detail 里绝不能出现凭据本身（脱敏规则一旦变动就是泄漏）。"""
        ck = fake_ck("1001")
        order = self._create(ck_text=ck)

        create_detail = self._audits(order)[0].detail
        assert "udb_cred" not in create_detail
        assert ck not in create_detail

        # 「换 CK」这条最有风险：逐条审计都查一遍
        collab.set_ck(self.db, self.user, order, fake_ck("1003"))
        for entry in self._audits(order):
            assert "udb_cred" not in entry.detail
            assert "cookie" not in entry.detail
            assert ck not in entry.detail

        paste_entry = next(
            entry for entry in self._audits(order) if entry.action == "huya:collab:ck"
        )
        # uid 与「内容变了」要留痕，否则查不出换过 CK
        assert "1003" in paste_entry.detail
        assert "credential_changed" in paste_entry.detail

    def test_rebuild_without_raw_cookie_is_rejected(self):
        order = self._create(ck_text=fake_ck("1001", web=True))
        with pytest.raises(collab.CollabError) as excinfo:
            collab.rebuild_ck(self.db, self.user, order)
        assert excinfo.value.code == "no_raw_cookie"

    def test_rebuild_from_raw_cookie_replaces_running_cookie(self):
        order = self._create(ck_text=fake_ck("1002", web=False))
        raw = order.raw_cookie
        collab.rebuild_ck(self.db, self.user, order)
        assert order.cookie  # 重建后的运行态
        assert order.raw_cookie == raw  # 原始凭据要原样留着，可反复重建
        assert order.ck_status == "unverified"

    def test_ck_parse_failure_is_readable(self):
        with pytest.raises(collab.CollabError) as excinfo:
            self._create(ck_text="这不是一份 CK")
        assert excinfo.value.code == "ck_invalid"


class TestCollabImages(TestCollabBase):
    def test_set_images_normalizes_and_dedupes(self):
        order = self._create()
        result = collab.set_images(
            self.db,
            self.user,
            order,
            "bind_images",
            ["collab/2026/10/16/a.png", "collab/2026/10/16/a.png", " "],
        )
        assert result["images"] == ["collab/2026/10/16/a.png"]

    def test_images_reject_path_traversal(self):
        order = self._create()
        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_images(
                self.db, self.user, order, "bind_images", ["../etc/passwd"]
            )
        assert excinfo.value.code == "image_key_invalid"

    def test_images_reject_unknown_slot_and_too_many(self):
        order = self._create()
        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_images(self.db, self.user, order, "other_images", [])
        assert excinfo.value.code == "image_slot_unknown"

        too_many = [f"collab/2026/10/16/{index}.png" for index in range(21)]
        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_images(self.db, self.user, order, "bind_images", too_many)
        assert excinfo.value.code == "images_too_many"

    def test_image_change_is_audited_with_added_and_removed(self):
        order = self._create()
        collab.set_images(
            self.db, self.user, order, "home_images", ["collab/2026/10/16/a.png"]
        )
        collab.set_images(
            self.db, self.user, order, "home_images", ["collab/2026/10/16/b.png"]
        )
        entry = self._audits(order)[-1]
        assert entry.action == "huya:collab:image"
        assert "b.png" in entry.detail and "a.png" in entry.detail

    def test_serialize_builds_urls_from_keys_not_stored_urls(self):
        order = self._create()
        collab.set_images(
            self.db, self.user, order, "bind_images", ["collab/2026/10/16/a.png"]
        )
        payload = collab.serialize_order(order)

        assert payload["bind_images"] == ["collab/2026/10/16/a.png"]
        assert payload["bind_image_urls"] == [
            "/api/files/object?key=collab%2F2026%2F10%2F16%2Fa.png"
        ]
        assert payload["bind_image_thumb_urls"] == [
            "/api/files/object?key=collab%2F2026%2F10%2F16%2Fa.png&variant=thumb"
        ]

    def test_serialize_never_exposes_cookie(self):
        order = self._create(ck_text=fake_ck("1001"))
        payload = collab.serialize_order(order)
        assert payload["has_cookie"] is True
        assert "cookie" not in payload
        assert "raw_cookie" not in payload


class TestCollabImport(TestCollabBase):
    def test_import_with_header_creates_rows(self):
        rows = [
            "店铺\t订单编号\t负责人\t绑定斗鱼号\t礼包\t备注",
            "佳佳\t2627501604891357\tcollab_op\t用户4241534208\t不朽伊人大礼包\t熟人",
            "小鱼儿\t2627501761915541\tcollab_support\t愚人节记得来骗我756\t激萌罗小黑\t",
        ]
        text = "\n".join(rows)
        result = collab.import_orders(
            self.db,
            self.user,
            text,
            owner_by_name={"collab_op": self.user.id, "collab_support": self.other.id},
        )

        assert result["created"] == 2
        assert result["errors"] == []
        orders = (
            self.db.query(HuyaExternalOrder)
            .filter(HuyaExternalOrder.order_no == "2627501604891357")
            .all()
        )
        assert orders[0].shop == "佳佳"
        assert orders[0].owner_id == self.user.id
        assert orders[0].gift_name == "不朽伊人大礼包"

    def test_import_without_header_uses_fixed_column_order(self):
        text = "佳佳\t2627501604891357\tcollab_op\t用户4241534208\t礼包A\t备注B"
        result = collab.import_orders(
            self.db, self.user, text, owner_by_name={"collab_op": self.user.id}
        )
        assert result["created"] == 1
        order = self.db.query(HuyaExternalOrder).one()
        assert order.shop == "佳佳"
        assert order.bind_douyu_account == "用户4241534208"
        assert order.gift_name == "礼包A"

    def test_import_unknown_owner_only_fails_that_row(self):
        """单行失败不能带崩整批（AGENTS §2：单项失败要如实暴露，但不整体失败）。"""
        rows = [
            "佳佳\tA1\t不认识的人\t用户1\t礼包\t",
            "佳佳\tA2\tcollab_op\t用户2\t礼包\t",
        ]
        text = "\n".join(rows)
        result = collab.import_orders(
            self.db, self.user, text, owner_by_name={"collab_op": self.user.id}
        )
        assert result["created"] == 1
        assert len(result["errors"]) == 1
        assert result["errors"][0]["line"] == 1
        assert "不认识的人" in result["errors"][0]["message"]
        assert self.db.query(HuyaExternalOrder).one().order_no == "A2"

    def test_import_single_bad_row_does_not_poison_the_batch(self):
        """超长字段会让那一行报错；用 SAVEPOINT 隔离后，后面几行必须照常写入。"""
        rows = [
            f"佳佳\tA1\tcollab_op\t用户1\t{'礼包' * 200}\t",
            "佳佳\tA2\tcollab_op\t用户2\t礼包\t",
            "佳佳\tA3\tcollab_op\t用户3\t礼包\t",
        ]
        text = "\n".join(rows)
        result = collab.import_orders(
            self.db, self.user, text, owner_by_name={"collab_op": self.user.id}
        )
        assert result["created"] == 2
        assert len(result["errors"]) == 1
        order_nos = {row.order_no for row in self.db.query(HuyaExternalOrder).all()}
        assert order_nos == {"A2", "A3"}

    def test_import_survives_a_row_that_fails_at_the_database_level(self):
        """「写库时才炸」的错误（这里用不存在的 owner_id 触发外键失败）必须被
        SAVEPOINT 隔离：不加 SAVEPOINT 时，第一次约束冲突会把会话打成失败态，
        后面每一行都会跟着报错 —— 单行失败就带崩了整批。
        """
        rows = [
            "佳佳\tA1\tghost_owner\t用户1\t礼包\t",
            "佳佳\tA2\tcollab_op\t用户2\t礼包\t",
        ]
        text = "\n".join(rows)
        result = collab.import_orders(
            self.db,
            self.user,
            text,
            owner_by_name={"ghost_owner": 999999, "collab_op": self.user.id},
        )
        assert result["created"] == 1
        assert len(result["errors"]) == 1
        assert self.db.query(HuyaExternalOrder).one().order_no == "A2"

    def test_import_duplicate_in_same_paste_keeps_first_only(self):
        rows = [
            "佳佳\tA1\tcollab_op\t用户1\t礼包\t",
            "小鱼儿\tA1\tcollab_op\t用户2\t礼包\t",
        ]
        text = "\n".join(rows)
        result = collab.import_orders(
            self.db, self.user, text, owner_by_name={"collab_op": self.user.id}
        )
        assert result["created"] == 1
        assert len(result["errors"]) == 1
        assert self.db.query(HuyaExternalOrder).one().shop == "佳佳"

    def test_import_updates_existing_row(self):
        order = self._create(order_no="A1", shop="佳佳")
        result = collab.import_orders(
            self.db,
            self.user,
            "小鱼儿\tA1\tcollab_op\t用户9\t礼包\t",
            owner_by_name={"collab_op": self.user.id},
        )
        assert result["updated"] == 1
        self.db.refresh(order)
        assert order.shop == "小鱼儿"
        assert order.bind_douyu_account == "用户9"

    def test_import_skips_rows_without_order_no(self):
        result = collab.import_orders(
            self.db, self.user, "佳佳\t\tcollab_op\t用户1\t\t"
        )
        assert result["skipped"] == 1
        assert result["created"] == 0

    def test_import_without_owner_defaults_to_the_importer(self):
        """导入时负责人留空 = 导入人自己（与手建同一套默认）。"""
        result = collab.import_orders(self.db, self.user, "佳佳\tO1\t\t用户1\t礼包\t")
        assert result["created"] == 1
        assert self.db.query(HuyaExternalOrder).one().owner_id == self.user.id


class FakeBindResp:
    """合成一份 ``check_user_bind_game_account`` 响应（字段名对齐真机响应结构）。"""

    def __init__(
        self,
        *,
        status: int = 200,
        msg: str = "",
        nick: str = "",
        role: str = "",
        role_id: str = "",
        open_id: str = "",
        plat: str = "",
        area: str = "",
        game_name: str = "和平精英",
    ):
        self.status = status
        self.msg = msg
        self.gameName = game_name
        self.accountData = SimpleNamespace(
            gameAccount=SimpleNamespace(nick=nick, openId=open_id),
            gameRole=SimpleNamespace(
                roleName=role, roleId=role_id, platName=plat, areaName=area
            ),
        )


class FakeGiftResp:
    """合成一份 ``getActPrizeList`` 响应（礼包列表）。"""

    def __init__(self, *, status: int = 200, msg: str = "", goods: list | None = None):
        self.status = status
        self.msg = msg
        self._goods = (
            goods
            if goods is not None
            else [
                {
                    "product_id": "hy-1001",
                    "name": "不朽伊人大礼包",
                    "price": 200,
                    "remain_text": "39%",
                    "category_name": "礼包",
                },
                {"product_id": "", "name": "缺 product_id 的脏数据"},
                {"product_id": "hy-1002", "name": ""},
            ]
        )

    def to_dict(self) -> dict:
        return {"status": self.status, "msg": self.msg, "goods": self._goods}


def points_center_blob(
    *,
    name: str = "不朽伊人大礼包",
    product_id: str = "hy-1001",
    points: int = 200,
    stock: int = 39,
    category: str = "礼包",
) -> bytes:
    """构造 queryAllocation 的 ``tag5`` 最小商品列表（TAF LIST）。"""
    os = TafOutputStream()
    os.write_head(0, TafType.LIST)
    os.write_int32(0, 1)
    # 分类
    os.write_head(0, TafType.STRUCT_BEGIN)
    os.write_string(1, category)
    os.write_head(6, TafType.LIST)
    os.write_int32(0, 1)
    # 商品
    os.write_head(0, TafType.STRUCT_BEGIN)
    os.write_head(1, TafType.STRUCT_BEGIN)
    # 商品基础信息 tag0
    os.write_head(0, TafType.STRUCT_BEGIN)
    os.write_string(0, product_id)
    os.write_string(2, name)
    os.write_string(4, category)
    os.write_string(5, "")
    os.write_int64(9, points * 10)
    os.write_int64(13, stock)
    # 积分 tag75.tag0.tag2
    os.write_head(75, TafType.STRUCT_BEGIN)
    os.write_head(0, TafType.STRUCT_BEGIN)
    os.write_int64(2, points)
    os.write_struct_end()
    os.write_struct_end()
    os.write_struct_end()  # tag0
    os.write_struct_end()  # tag1
    os.write_struct_end()  # 商品
    os.write_struct_end()  # 分类
    return os.get_bytes()


class FakePointsCenterResp:
    """合成一份 ``PlayMallUI.queryAllocation`` 响应。"""

    def __init__(
        self,
        *,
        code: int = 200,
        message: str = "成功",
        payload: dict | None = None,
        goods: list | None = None,
    ):
        if payload is None:
            if goods is None:
                goods = [
                    {
                        "product_id": "hy-1001",
                        "name": "不朽伊人大礼包",
                        "points": 200,
                        "stock": 39,
                        "category": "礼包",
                    }
                ]
            blobs = [points_center_blob(**item) for item in goods]
            payload = {"tag2": [{"tag5": blob} for blob in blobs]}
        self.code = code
        self.message = message
        self.payload = payload

    def to_dict(self) -> dict:
        return {"code": self.code, "message": self.message, "payload": self.payload}


class FakeProbeClient:
    """离线探针客户端：按脚本依次给出 ``None`` / 异常 / 响应，并记录调用参数。"""

    def __init__(
        self, *outcomes, gifts: list | None = None, points_gifts: list | None = None
    ):
        self.outcomes = list(outcomes)
        self.gift_outcomes = list(gifts or [])
        self.point_outcomes = list(points_gifts or [])
        self.calls: list[dict] = []
        self.gift_calls: list[dict] = []
        self.points_calls: list[dict] = []

    def check_user_bind_game_account(self, **kwargs):
        self.calls.append(kwargs)
        if not self.outcomes:
            raise AssertionError("探针被调用的次数超出用例预期")
        outcome = self.outcomes.pop(0)
        if isinstance(outcome, Exception):
            raise outcome
        return outcome

    def get_act_prize_list(self, **kwargs):
        self.gift_calls.append(kwargs)
        outcome = self.gift_outcomes.pop(0) if self.gift_outcomes else FakeGiftResp()
        if isinstance(outcome, Exception):
            raise outcome
        return outcome

    def play_mall_query_allocation(self, **kwargs):
        self.points_calls.append(kwargs)
        outcome = (
            self.point_outcomes.pop(0)
            if self.point_outcomes
            else FakePointsCenterResp()
        )
        if isinstance(outcome, Exception):
            raise outcome
        return outcome


class TestCollabCkProbe(TestCollabBase):
    """S5：只读探针。判据（真机对照 2026-10-16）= 响应非 None 且 status == 200。"""

    def _row_with_ck(self, order_no: str = "P1", uid: str = "1199647239697"):
        return self._create(order_no=order_no, ck_text=fake_ck(uid))

    def test_verify_success_marks_valid_and_writes_audit(self):
        order = self._row_with_ck()
        client = FakeProbeClient(
            FakeBindResp(
                nick="随遇而安",
                role="吃鸡199706",
                role_id="1234567",
                plat="手Q",
                area="苹果(iOS)",
            )
        )

        result = collab.verify_ck(self.db, self.user, order, client=client)

        assert result["ok"] is True
        assert order.ck_status == "valid"
        assert order.ck_checked_at is not None
        assert "随遇而安" in order.ck_check_message
        assert "吃鸡199706" in order.ck_check_message
        assert order.version == 1
        assert "huya:collab:ck_status" in [row.action for row in self._audits(order)]

    def test_verify_uses_this_rows_ck_and_the_default_activity(self):
        """探针必须拿**本行的** CK 去问默认活动，不能借用别的行/别的活动。"""
        order = self._row_with_ck(uid="1199647239697")
        client = FakeProbeClient(FakeBindResp(nick="随遇而安"))

        collab.verify_ck(self.db, self.user, order, client=client)

        call = client.calls[0]
        assert call["uid"] == 1199647239697
        assert isinstance(call["uid"], int)
        assert call["cookie"] == order.cookie
        assert (
            call["b_act_id"]
            == collab.PROBE_BIND_ACT_ID
            == int(HUYA_DEFAULT_BIND_ACT_ID)
        )
        assert call["is_use_outer_act_id"] == 1
        assert (
            call["outer_act_id"]
            == collab.PROBE_OUTER_ACT_ID
            == HUYA_DEFAULT_OUTER_ACT_ID
        )

    def test_verify_marks_invalid_when_status_is_not_200(self):
        order = self._row_with_ck()
        client = FakeProbeClient(FakeBindResp(status=401, msg="登录态失效"))

        result = collab.verify_ck(self.db, self.user, order, client=client)

        assert result["ok"] is False
        assert order.ck_status == "invalid"
        assert order.ck_checked_at is not None
        assert "401" in order.ck_check_message
        assert "登录态失效" in order.ck_check_message

    def test_verify_treats_empty_response_as_invalid(self):
        """裸 cookie / 别人的 uid 在真机上都返回 None —— 必须是「失效」。"""
        order = self._row_with_ck()

        result = collab.verify_ck(
            self.db, self.user, order, client=FakeProbeClient(None)
        )

        assert result["ok"] is False
        assert order.ck_status == "invalid"
        assert "没有返回数据" in order.ck_check_message

    def test_verify_does_not_accept_payload_without_200(self):
        """别把判据改成「有数据就算成功」：status=0 却带数据也必须判失效。"""
        order = self._row_with_ck()
        client = FakeProbeClient(
            FakeBindResp(status=0, nick="随遇而安", role="吃鸡199706")
        )

        result = collab.verify_ck(self.db, self.user, order, client=client)

        assert result["ok"] is False
        assert order.ck_status == "invalid"
        # 失效响应里的字段一个都不许落到台账上（绑定快照也不许被清空/覆盖）
        assert order.bind_game_account == ""
        assert order.bind_game_role == ""
        assert order.bind_game_channel == ""
        assert order.bind_snapshot_at is None

    def test_verify_survives_probe_exception(self):
        order = self._row_with_ck()

        result = collab.verify_ck(
            self.db, self.user, order, client=FakeProbeClient(RuntimeError("boom"))
        )

        assert result["ok"] is False
        assert order.ck_status == "invalid"
        assert "RuntimeError" in order.ck_check_message

    def test_verify_writes_binding_snapshot(self):
        """绑定账号 + **绑定角色**（用户强调这条重要）+ 平台区服都落到快照列。"""
        order = self._row_with_ck()
        client = FakeProbeClient(
            FakeBindResp(
                nick="随遇而安",
                role="吃鸡199706",
                plat="手Q",
                area="苹果(iOS)",
            )
        )

        result = collab.verify_ck(self.db, self.user, order, client=client)

        assert order.bind_game_account == "随遇而安"
        assert order.bind_game_role == "吃鸡199706"
        assert order.bind_game_channel == "手Q · 苹果(iOS)"
        assert order.bind_snapshot_at is not None
        assert result["binding"] == {
            "bind_game_account": "随遇而安",
            "bind_game_role": "吃鸡199706",
            "bind_game_channel": "手Q · 苹果(iOS)",
        }
        # 人工填的列一个都不碰；nickname 是「虎牙号昵称」，这个探针给不出
        assert order.bind_douyu_account == ""
        assert order.nickname == ""
        assert "huya:collab:probe_snapshot" in [
            row.action for row in self._audits(order)
        ]

    def test_verify_overwrites_previous_binding_snapshot(self):
        """快照列是探针独占的「现在绑了谁」，重验要覆盖，并把旧值记进审计。"""
        order = self._row_with_ck()
        collab.verify_ck(
            self.db,
            self.user,
            order,
            client=FakeProbeClient(FakeBindResp(nick="旧账号", role="旧角色")),
        )

        collab.verify_ck(
            self.db,
            self.user,
            order,
            client=FakeProbeClient(FakeBindResp(nick="新账号", role="新角色")),
        )

        assert order.bind_game_account == "新账号"
        assert order.bind_game_role == "新角色"
        snapshots = [
            row
            for row in self._audits(order)
            if row.action == "huya:collab:probe_snapshot"
        ]
        # 两次验证各记一条（都在「快照变了」时才记），最后一条带 before/after
        assert len(snapshots) == 2
        detail = json.loads(snapshots[-1].detail)
        assert detail["before"]["bind_game_role"] == "旧角色"
        assert detail["after"]["bind_game_role"] == "新角色"

    def test_binding_role_from_probe_satisfies_the_recharge_precondition(self):
        """探针读到绑定角色 + 礼包 + 主页截图 → 自动进入「待充值」。"""
        order = self._row_with_ck()
        collab.update_order(
            self.db,
            self.user,
            order,
            {"gift_name": "不朽伊人大礼包", "required_points": 8000},
        )
        collab.set_images(
            self.db,
            self.user,
            order,
            "home_images",
            ["collab/2026/10/16/home01.png"],
        )
        collab.verify_ck(
            self.db,
            self.user,
            order,
            client=FakeProbeClient(FakeBindResp(nick="随遇而安", role="吃鸡199706")),
        )
        assert order.bind_douyu_account == ""
        assert order.bind_images == []

        assert order.status == collab.STATUS_TO_RECHARGE
        assert order.bound_at is not None

    def test_verify_never_touches_manually_entered_columns(self):
        order = self._row_with_ck()
        collab.update_order(
            self.db,
            self.user,
            order,
            {"bind_douyu_account": "人工填的号", "bind_douyu_uid": "999"},
        )

        collab.verify_ck(
            self.db,
            self.user,
            order,
            client=FakeProbeClient(FakeBindResp(nick="随遇而安", role="吃鸡199706")),
        )

        assert order.bind_douyu_account == "人工填的号"
        assert order.bind_douyu_uid == "999"
        assert order.bind_game_account == "随遇而安"
        assert order.bind_game_role == "吃鸡199706"

    def test_verify_without_cookie_raises_instead_of_marking_invalid(self):
        order = self._create(order_no="P-NOCK")

        with pytest.raises(collab.CollabError) as excinfo:
            collab.verify_ck(self.db, self.user, order)

        assert excinfo.value.code == "no_cookie"
        assert order.ck_status == "unverified"

    def test_verify_refuses_void_row(self):
        order = self._row_with_ck()
        collab.set_status(self.db, self.user, order, "void", reason="客户取消")

        with pytest.raises(collab.CollabError) as excinfo:
            collab.verify_ck(self.db, self.user, order, client=FakeProbeClient(None))

        assert excinfo.value.code == "void_order"

    def test_manual_ck_status_only_allows_disable_and_reset(self):
        """人工只能停用/恢复；「有效/失效」必须由探针给（人工标有效=绕过验证）。"""
        order = self._row_with_ck()

        collab.set_manual_ck_status(
            self.db, self.user, order, "disabled", message="这个号不打算再用"
        )
        assert order.ck_status == "disabled"
        # 人工停用不算「刚跑过探针」，不刷 ck_checked_at
        assert order.ck_checked_at is None

        collab.set_manual_ck_status(self.db, self.user, order, "unverified")
        assert order.ck_status == "unverified"

        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_manual_ck_status(self.db, self.user, order, "valid")
        assert excinfo.value.code == "ck_status_manual_only"
        assert order.ck_status == "unverified"

    def test_verify_never_touches_business_status(self):
        """CK 失效不改业务状态：待充值的单子业务上仍然要等人处理。"""
        order = self._row_with_ck()
        self._bind_ready(order)
        assert order.status == collab.STATUS_TO_RECHARGE

        collab.verify_ck(
            self.db, self.user, order, client=FakeProbeClient(FakeBindResp(status=401))
        )

        assert order.ck_status == "invalid"
        assert order.status == collab.STATUS_TO_RECHARGE
        assert order.bound_at is not None

    def test_verify_never_creates_huya_account(self):
        """边界（设计文档 §2.3）：探针只读，不建 HuyaAccount、不写工作台。"""
        order = self._row_with_ck()

        collab.verify_ck(
            self.db,
            self.user,
            order,
            client=FakeProbeClient(FakeBindResp(nick="随遇而安")),
        )

        assert self.db.query(HuyaAccount).count() == 0
        assert self.db.query(HuyaWorkbenchAccount).count() == 0

    def test_verify_audit_never_carries_the_cookie(self):
        order = self._row_with_ck()
        collab.verify_ck(
            self.db,
            self.user,
            order,
            client=FakeProbeClient(FakeBindResp(nick="随遇而安")),
        )

        payload = json.dumps(
            [row.detail for row in self._audits(order)], ensure_ascii=False
        )
        assert "udb_cred" not in payload
        assert "udb_biztoken" not in payload
        assert order.cookie not in payload

    def test_batch_isolates_single_row_failure(self):
        rows = [
            self._row_with_ck(order_no="B1", uid="1001"),
            self._row_with_ck(order_no="B2", uid="1002"),
            self._row_with_ck(order_no="B3", uid="1003"),
        ]
        client = FakeProbeClient(
            FakeBindResp(nick="A"),
            RuntimeError("boom"),
            FakeBindResp(status=500, msg="服务端异常"),
        )

        result = collab.verify_ck_batch(
            self.db, self.user, rows, client=client, interval_seconds=0
        )

        assert result["total"] == 3
        assert result["valid"] == 1
        assert result["failed"] == 2
        assert [row.ck_status for row in rows] == ["valid", "invalid", "invalid"]
        assert "RuntimeError" in result["results"][1]["message"]

    def test_batch_reports_rows_without_ck_without_lying(self):
        with_ck = self._row_with_ck(order_no="B1", uid="1001")
        without_ck = self._create(order_no="B2")
        client = FakeProbeClient(FakeBindResp(nick="A"))

        result = collab.verify_ck_batch(
            self.db, self.user, [with_ck, without_ck], client=client, interval_seconds=0
        )

        assert result["valid"] == 1
        assert result["failed"] == 1
        # 没 CK 的行保持「未验证」，不能谎报成「已失效」
        assert without_ck.ck_status == "unverified"
        assert "CK" in result["results"][1]["message"]
        assert len(client.calls) == 1  # 没 CK 的行不发请求

    def test_batch_keeps_interval_between_rows(self, monkeypatch):
        """人工触发也要串行 + 间隔，别把账号送进风控名单。"""
        rows = [
            self._row_with_ck(order_no="B1", uid="1001"),
            self._row_with_ck(order_no="B2", uid="1002"),
            self._row_with_ck(order_no="B3", uid="1003"),
        ]
        client = FakeProbeClient(*[FakeBindResp(nick="A") for _ in rows])
        slept: list[float] = []
        monkeypatch.setattr(
            collab, "time", SimpleNamespace(sleep=lambda seconds: slept.append(seconds))
        )

        collab.verify_ck_batch(
            self.db, self.user, rows, client=client, interval_seconds=0.5
        )

        assert slept == [0.5, 0.5]  # 3 条之间睡 2 次，第一条前不睡

    def test_batch_rejects_too_many_rows(self):
        with pytest.raises(collab.CollabError) as excinfo:
            collab.verify_ck_batch(
                self.db, self.user, [None] * (collab.PROBE_BATCH_MAX + 1)
            )

        assert excinfo.value.code == "verify_batch_too_large"

    def test_list_orders_by_ids_keeps_request_order_and_ignores_missing(self):
        first = self._row_with_ck(order_no="B1", uid="1001")
        second = self._row_with_ck(order_no="B2", uid="1002")

        rows = collab.list_orders_by_ids(self.db, [second.id, first.id, 99999999])

        assert [row.id for row in rows] == [second.id, first.id]
        with pytest.raises(collab.CollabError) as excinfo:
            collab.list_orders_by_ids(self.db, [])
        assert excinfo.value.code == "verify_empty"


class TestCollabCkPreview(TestCollabBase):
    """粘贴 CK 后的即时预览：探针 + 顺带拉礼包，一行都不写库。"""

    def test_preview_returns_binding_and_gifts(self):
        client = FakeProbeClient(
            FakeBindResp(
                nick="随遇而安", role="吃鸡199706", plat="手Q", area="苹果(iOS)"
            )
        )

        preview = collab.preview_ck(fake_ck("1199647239697"), client=client)

        assert preview["ok"] is True
        assert preview["has_role"] is True
        assert preview["uid"] == "1199647239697"
        assert preview["hy_username"] == "hy_1199647239697"
        assert preview["bind_game_role"] == "吃鸡199706"
        assert preview["gifts"] == [
            {
                "product_id": "hy-1001",
                "name": "不朽伊人大礼包",
                "price": 200,
                "remain_text": "剩39",
                "stock": 39,
                "score": 200,
                "category_name": "礼包",
                "icon": "",
            }
        ]
        assert "已拉取 1 个礼包（积分中心）" in preview["gifts_message"]
        assert client.points_calls[0]["uid"] == 1199647239697
        assert client.points_calls[0]["deep"] is True
        assert client.gift_calls == []

    def test_preview_of_invalid_ck_does_not_fetch_gifts(self):
        client = FakeProbeClient(None)

        preview = collab.preview_ck(fake_ck("1199647239697"), client=client)

        assert preview["ok"] is False
        assert preview["has_role"] is False
        assert preview["gifts"] == []
        assert client.points_calls == []  # CK 都没通，不去拉礼包

    def test_preview_of_probe_without_role_is_not_submittable(self):
        """探针通了但读不到绑定角色 → 不允许提交（has_role=False）。"""
        preview = collab.preview_ck(
            fake_ck("1199647239697"),
            client=FakeProbeClient(FakeBindResp(nick="随遇而安", role="")),
        )
        assert preview["ok"] is True
        assert preview["has_role"] is False

    def test_preview_writes_nothing(self):
        order = self._create(order_no="PREVIEW-1", shop="佳佳")
        before = order.version
        collab.preview_ck(
            fake_ck("1199647239697"),
            client=FakeProbeClient(FakeBindResp(nick="随遇而安", role="吃鸡199706")),
        )
        self.db.refresh(order)
        assert order.version == before
        assert self.db.query(HuyaExternalOrder).count() == 1

    def test_preview_rejects_unparsable_ck(self):
        with pytest.raises(collab.CollabError) as excinfo:
            collab.preview_ck("这不是一份 Cookie", client=FakeProbeClient(None))
        assert excinfo.value.code == "ck_invalid"


class TestCollabGifts(TestCollabBase):
    def test_gift_fetch_skips_dirty_rows(self):
        gifts, message = collab.fetch_activity_gifts(
            1199647239697, fake_ck("1199647239697"), client=FakeProbeClient()
        )
        assert [gift["name"] for gift in gifts] == ["不朽伊人大礼包"]
        assert message == "已拉取 1 个礼包"

    def test_gift_fetch_reports_failure_without_raising(self):
        gifts, message = collab.fetch_activity_gifts(
            1199647239697,
            fake_ck("1199647239697"),
            client=FakeProbeClient(gifts=[FakeGiftResp(status=500, msg="活动已结束")]),
        )
        assert gifts == []
        assert "拉取礼包失败" in message and "500" in message

    def test_gift_fetch_survives_exception(self):
        gifts, message = collab.fetch_activity_gifts(
            1199647239697,
            fake_ck("1199647239697"),
            client=FakeProbeClient(gifts=[RuntimeError("boom")]),
        )
        assert gifts == []
        assert "RuntimeError" in message

    def test_gift_fetch_empty_list_hints_stale_sid(self):
        """200 但 0 个商品 = SID 不是当前活动（真机 2026-10-16：2203 → 0，2277 → 30）。"""
        gifts, message = collab.fetch_activity_gifts(
            1199647239697,
            fake_ck("1199647239697"),
            sid=2203,
            client=FakeProbeClient(gifts=[FakeGiftResp(goods=[])]),
        )
        assert gifts == []
        assert "SID=2203" in message and "工作台" in message

    def test_points_center_gift_fetch_parses_fields(self):
        gifts, message = collab.fetch_points_center_gifts(
            1199647239697,
            fake_ck("1199647239697"),
            client=FakeProbeClient(),
        )

        assert gifts == [
            {
                "product_id": "hy-1001",
                "name": "不朽伊人大礼包",
                "price": 200,
                "remain_text": "剩39",
                "stock": 39,
                "score": 200,
                "category_name": "礼包",
                "icon": "",
            }
        ]
        assert message == "已拉取 1 个礼包（积分中心）"

    def test_points_center_gift_fetch_reports_failure(self):
        gifts, message = collab.fetch_points_center_gifts(
            1199647239697,
            fake_ck("1199647239697"),
            client=FakeProbeClient(
                points_gifts=[FakePointsCenterResp(code=500, message="接口炸了")]
            ),
        )
        assert gifts == []
        assert "拉取礼包失败" in message and "500" in message

    def test_points_center_gift_fetch_survives_exception(self):
        gifts, message = collab.fetch_points_center_gifts(
            1199647239697,
            fake_ck("1199647239697"),
            client=FakeProbeClient(points_gifts=[RuntimeError("boom")]),
        )
        assert gifts == []
        assert "RuntimeError" in message

    def test_points_center_gift_fetch_empty_list(self):
        gifts, message = collab.fetch_points_center_gifts(
            1199647239697,
            fake_ck("1199647239697"),
            client=FakeProbeClient(points_gifts=[FakePointsCenterResp(goods=[])]),
        )
        assert gifts == []
        assert "积分中心" in message and "空的" in message

    def test_points_center_parser_reads_tag75_points(self):
        from core.huya.shop_structs import parse_points_center_gifts

        gifts = parse_points_center_gifts(
            {"tag2": [{"tag5": points_center_blob(points=888, stock=7)}]}
        )
        assert gifts[0]["price"] == 888
        assert gifts[0]["score"] == 888
        assert gifts[0]["stock"] == 7

    def test_resolve_gifts_sid_prefers_workbench_config(self):
        self.db.add(HuyaConfig(sid="2277"))
        self.db.commit()

        assert collab.resolve_gifts_sid(self.db) == 2277

    def test_resolve_gifts_sid_falls_back_when_unset_or_garbage(self):
        assert collab.resolve_gifts_sid(self.db) == collab.GIFTS_SID  # 没有配置行

        row = HuyaConfig(sid="")
        self.db.add(row)
        self.db.commit()
        assert collab.resolve_gifts_sid(self.db) == collab.GIFTS_SID

        row.sid = "不是数字"
        self.db.commit()
        assert collab.resolve_gifts_sid(self.db) == collab.GIFTS_SID

    def test_preview_ignores_legacy_sid_and_uses_points_center(self):
        self.db.add(HuyaConfig(sid="2277"))
        self.db.commit()
        client = FakeProbeClient(FakeBindResp(nick="随遇而安", role="吃鸡199706"))

        preview = collab.preview_ck(fake_ck("1199647239697"), sid=2277, client=client)

        assert client.points_calls
        assert "sid" not in client.points_calls[0]
        assert preview["has_role"] is True

    def test_preview_endpoint_uses_points_center_source(self, monkeypatch):
        """接口层仍可带旧 SID 配置，但预览实际走积分中心 queryAllocation。"""
        from web.backend.routers.huya import collab as collab_router
        from web.backend.schemas import HuyaCollabCkPreviewRequest

        self.db.add(HuyaConfig(sid="2277"))
        self.db.commit()
        client = FakeProbeClient(FakeBindResp(nick="随遇而安", role="吃鸡199706"))
        monkeypatch.setattr(collab, "_build_probe_client", lambda: client)

        out = collab_router.preview_external_order_ck(
            HuyaCollabCkPreviewRequest(ck_text=fake_ck("1199647239697")),
            db=self.db,
            _current=self.user,
        )

        # 直接调路由函数时没走 pydantic 出参，拿到的就是 dict
        assert client.points_calls
        assert out["has_role"] is True


class TestCollabCreateStrict(TestCollabBase):
    """人工新建表单：订单编号 / 店铺 / CK / 礼包名 / 主页截图必填，且 CK 必须读到绑定角色。"""

    HOME_IMAGE = "collab/2026/10/16/home01.png"

    def _row_count(self) -> int:
        return self.db.query(HuyaExternalOrder).count()

    def test_missing_required_fields_blocks_creation(self):
        with pytest.raises(collab.CollabError) as excinfo:
            collab.create_order(
                self.db,
                self.user,
                order_no="STRICT-1",
                require_verified=True,
                client=FakeProbeClient(
                    FakeBindResp(nick="随遇而安", role="吃鸡199706")
                ),
            )
        assert excinfo.value.code == "create_field_required"
        assert "店铺" in excinfo.value.detail and "礼包名" in excinfo.value.detail
        assert "主页截图" in excinfo.value.detail
        assert self._row_count() == 0  # 失败不留半行

    def test_missing_home_screenshot_blocks_creation(self):
        """绑定图不再要求，但主页截图是必填（2026-10-16 用户拍板）。"""
        with pytest.raises(collab.CollabError) as excinfo:
            collab.create_order(
                self.db,
                self.user,
                order_no="STRICT-NOIMG",
                shop="佳佳",
                gift_name="不朽伊人大礼包",
                ck_text=fake_ck("1199647239697"),
                require_verified=True,
                client=FakeProbeClient(
                    FakeBindResp(nick="随遇而安", role="吃鸡199706")
                ),
            )
        assert excinfo.value.code == "create_field_required"
        assert "主页截图" in excinfo.value.detail
        assert self._row_count() == 0

    def test_bad_home_image_key_is_rejected_before_insert(self):
        with pytest.raises(collab.CollabError) as excinfo:
            collab.create_order(
                self.db,
                self.user,
                order_no="STRICT-BADIMG",
                shop="佳佳",
                gift_name="不朽伊人大礼包",
                ck_text=fake_ck("1199647239697"),
                home_images=["../../etc/passwd"],
                require_verified=True,
                client=FakeProbeClient(
                    FakeBindResp(nick="随遇而安", role="吃鸡199706")
                ),
            )
        assert excinfo.value.code == "image_key_invalid"
        assert self._row_count() == 0

    def test_probe_without_role_blocks_creation(self):
        with pytest.raises(collab.CollabError) as excinfo:
            collab.create_order(
                self.db,
                self.user,
                order_no="STRICT-2",
                shop="佳佳",
                gift_name="不朽伊人大礼包",
                ck_text=fake_ck("1199647239697"),
                home_images=[self.HOME_IMAGE],
                require_verified=True,
                client=FakeProbeClient(FakeBindResp(nick="随遇而安", role="")),
            )
        assert excinfo.value.code == "ck_not_verified"
        assert "读不到绑定角色" in excinfo.value.detail
        assert self._row_count() == 0

    def test_invalid_ck_blocks_creation(self):
        with pytest.raises(collab.CollabError) as excinfo:
            collab.create_order(
                self.db,
                self.user,
                order_no="STRICT-3",
                shop="佳佳",
                gift_name="不朽伊人大礼包",
                ck_text=fake_ck("1199647239697"),
                home_images=[self.HOME_IMAGE],
                require_verified=True,
                client=FakeProbeClient(None),
            )
        assert excinfo.value.code == "ck_not_verified"
        assert self._row_count() == 0

    def test_valid_ck_creates_row_with_snapshot_and_audit(self):
        order = collab.create_order(
            self.db,
            self.user,
            order_no="STRICT-4",
            shop="佳佳",
            gift_name="不朽伊人大礼包",
            ck_text=fake_ck("1199647239697"),
            home_images=[self.HOME_IMAGE, "collab/2026/10/16/home02.png"],
            require_verified=True,
            client=FakeProbeClient(
                FakeBindResp(
                    nick="随遇而安", role="吃鸡199706", plat="手Q", area="苹果(iOS)"
                )
            ),
        )

        assert order.ck_status == "valid"
        assert order.ck_checked_at is not None
        assert order.bind_game_account == "随遇而安"
        assert order.bind_game_role == "吃鸡199706"
        assert order.bind_game_channel == "手Q · 苹果(iOS)"
        assert order.bind_snapshot_at is not None
        assert order.gift_name == "不朽伊人大礼包"
        assert order.home_images == [self.HOME_IMAGE, "collab/2026/10/16/home02.png"]
        assert order.bind_images == []  # 绑定图不再人工上传
        assert order.owner_id == self.user.id
        actions = [row.action for row in self._audits(order)]
        assert "huya:collab:create" in actions
        assert "huya:collab:ck_status" in actions

    def test_import_path_is_not_forced_to_verify(self):
        """粘贴导入没有 CK 列，不能被严格模式挡住（走的是默认 require_verified=False）。"""
        result = collab.import_orders(
            self.db, self.user, "佳佳\tIMP-1\t\t用户1\t礼包A\t"
        )
        assert result["created"] == 1
        assert self.db.query(HuyaExternalOrder).one().ck_status == "unverified"

    def test_draft_needs_only_order_no_and_shop(self):
        """先建单后补资料：只填订单号 + 店铺，状态直接进「待补资料」。"""
        order = collab.create_order(
            self.db,
            self.user,
            order_no="DRAFT-1",
            shop="佳佳",
            draft=True,
            require_verified=True,  # 草稿模式必须压过严格模式，不跑探针
        )

        assert order.status == collab.STATUS_INCOMPLETE
        assert order.shop == "佳佳"
        assert order.cookie == ""
        assert order.gift_name == ""
        assert order.home_images == []
        assert order.ck_status == "unverified"
        assert self._row_count() == 1
        detail = self._audits(order)[-1].detail
        assert '"draft"' in detail and "true" in detail
        assert collab.summary(self.db)["by_status"]["incomplete"] == 1

    def test_incomplete_cannot_advance_until_customer_info_complete(self):
        order = collab.create_order(
            self.db, self.user, order_no="DRAFT-2", shop="佳佳", draft=True
        )

        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_status(self.db, self.user, order, collab.STATUS_TO_RECHARGE)

        assert excinfo.value.code == "incomplete_precondition"
        assert "外部 CK" in excinfo.value.detail
        assert "礼包名" in excinfo.value.detail
        assert "主页截图" in excinfo.value.detail
        assert order.status == collab.STATUS_INCOMPLETE

    def test_incomplete_advances_after_info_complete(self):
        order = collab.create_order(
            self.db,
            self.user,
            order_no="DRAFT-3",
            shop="佳佳",
            gift_name="不朽伊人大礼包",
            required_points=8000,
            home_images=[self.HOME_IMAGE],
            draft=True,
        )
        collab.set_ck(self.db, self.user, order, fake_ck("1199647239697"))
        collab.verify_ck(
            self.db,
            self.user,
            order,
            client=FakeProbeClient(FakeBindResp(nick="微信用户", role="吃鸡199706")),
        )

        assert order.status == collab.STATUS_TO_RECHARGE
        assert order.bound_at is not None
        assert order.void_reason == ""


class TestCollabOrderSorting(TestCollabBase):
    """列表默认按创建顺序；表头排序走白名单，升/降序都稳定。"""

    def test_default_order_is_creation_order(self):
        first = self._create(order_no="SORT-1", shop="乙")
        second = self._create(order_no="SORT-2", shop="甲")

        rows, total = collab.list_orders(self.db)

        assert total == 2
        assert [row.id for row in rows] == [first.id, second.id]

    def test_sort_by_order_no_ascending_and_descending(self):
        self._create(order_no="SORT-B")
        self._create(order_no="SORT-A")

        rows, _ = collab.list_orders(self.db, sort_by="order_no", sort_order="asc")
        assert [row.order_no for row in rows] == ["SORT-A", "SORT-B"]

        rows, _ = collab.list_orders(self.db, sort_by="order_no", sort_order="desc")
        assert [row.order_no for row in rows] == ["SORT-B", "SORT-A"]

    def test_sort_by_owner_username(self):
        self._create(order_no="SORT-OWNER-1", owner_id=self.user.id)
        self._create(order_no="SORT-OWNER-2", owner_id=self.other.id)

        rows, _ = collab.list_orders(self.db, sort_by="owner", sort_order="asc")

        assert [row.owner_id for row in rows] == [self.user.id, self.other.id]


class TestCollabKeywordSearch(TestCollabBase):
    """列表搜索：一个输入框打穿订单号 / 店铺 / 绑定账号·角色 / 平台区服 / 虎牙号 / 礼包 / 备注。"""

    def _search(self, text: str) -> list[HuyaExternalOrder]:
        rows, total = collab.list_orders(self.db, keyword=text)
        assert total == len(rows)
        return rows

    def test_keyword_hits_each_searchable_column(self):
        order = self._create(
            order_no="K-1",
            shop="向日葵",
            gift_name="天网终结者大礼包",
            remark="熟人介绍",
            ck_text=fake_ck("1199647123456"),
        )
        collab.verify_ck(
            self.db,
            self.user,
            order,
            client=FakeProbeClient(
                FakeBindResp(
                    nick="微信用户", role="不打PG打老公", plat="苹果(iOS)", area="微信"
                )
            ),
        )

        for text in (
            "K-1",  # 订单编号
            "向日葵",  # 店铺
            "微信用户",  # 绑定账号（探针）
            "不打PG打老公",  # 绑定角色（探针）
            "苹果(iOS)",  # 平台 · 区服
            "1199647123456",  # 虎牙号
            "天网终结者",  # 礼包名
            "熟人",  # 备注
        ):
            assert [row.id for row in self._search(text)] == [order.id], text

    def test_keyword_miss_returns_empty_page(self):
        self._create(order_no="K-2", shop="佳佳")

        assert self._search("查不到的东西") == []


class TestCollabGiftFilter(TestCollabBase):
    """按礼包/皮肤筛选：下拉选项去重，查询按礼包名精确匹配。"""

    def test_list_orders_filters_by_exact_gift_name(self):
        target = self._create(order_no="G-1", gift_name="天网终结者大礼包")
        self._create(order_no="G-2", gift_name="套装-纯白跃动")

        rows, total = collab.list_orders(self.db, gift_name="天网终结者大礼包")
        assert total == 1
        assert [row.id for row in rows] == [target.id]

        # 筛选项是点选值，不做模糊匹配（模糊搜索仍走 keyword）
        rows, total = collab.list_orders(self.db, gift_name="天网")
        assert total == 0
        assert rows == []

    def test_gift_names_are_unique_sorted_and_ignore_empty(self):
        self._create(order_no="G-3", gift_name="套装-纯白跃动")
        self._create(order_no="G-4", gift_name="套装-纯白跃动")
        self._create(order_no="G-5", gift_name="天网终结者大礼包")
        self._create(order_no="G-6", gift_name="")

        assert collab.list_gift_names(self.db) == sorted(
            ["套装-纯白跃动", "天网终结者大礼包"]
        )


class TestCollabRealtimeEvents(TestCollabBase):
    def test_update_order_publishes_invalidate_event(self, monkeypatch):
        order = self._create(order_no="RT-1")
        captured: dict = {}

        def fake_publish(**kwargs):
            captured.update(kwargs)
            return kwargs

        monkeypatch.setattr(collab_events, "publish", fake_publish)
        updated = collab_api.update_external_order(
            order.id,
            HuyaExternalOrderUpdate(remark="实时更新", expected_version=order.version),
            self.db,
            self.user,
        )

        assert updated["remark"] == "实时更新"
        assert captured == {
            "scope": "orders",
            "action": "order.update",
            "actor_id": self.user.id,
        }


class TestCollabBuyerCluster(TestCollabBase):
    """同一买家的多单：按「绑定账号 + 绑定角色」软识别（只读派生，不动 uid 唯一性）。

    真机业务口径：一单一份外部 CK，同一个买家买两件往往是**两份不同的 CK**（uid 不同），
    所以买家指纹只能用探针读到的绑定对象，不能用 uid。
    """

    def _order_with_binding(
        self,
        order_no: str,
        uid: str,
        *,
        account: str,
        role: str,
        gift_name: str = "礼包A",
    ) -> HuyaExternalOrder:
        order = self._create(
            order_no=order_no, shop="佳佳", gift_name=gift_name, ck_text=fake_ck(uid)
        )
        collab.verify_ck(
            self.db,
            self.user,
            order,
            client=FakeProbeClient(
                FakeBindResp(nick=account, role=role, plat="手Q", area="苹果(iOS)")
            ),
        )
        return order

    def _cluster_of(self, order: HuyaExternalOrder) -> dict | None:
        return collab.buyer_clusters(self.db, [order]).get(int(order.id))

    def test_two_orders_of_same_buyer_are_grouped(self):
        first = self._order_with_binding(
            "BUY-1", "1199647000001", account="微信用户", role="不打PG打老公"
        )
        second = self._order_with_binding(
            "BUY-2", "1199647000002", account="微信用户", role="不打PG打老公"
        )

        first_cluster = self._cluster_of(first)
        second_cluster = self._cluster_of(second)

        assert first_cluster is not None
        assert first_cluster["total"] == 2
        assert first_cluster["key"] == "微信用户 · 不打PG打老公"
        assert [item["order_no"] for item in first_cluster["others"]] == ["BUY-2"]
        assert second_cluster is not None
        assert [item["order_no"] for item in second_cluster["others"]] == ["BUY-1"]

    def test_same_account_different_role_is_not_grouped(self):
        first = self._order_with_binding(
            "BUY-3", "1199647000003", account="微信用户", role="角色甲"
        )
        self._order_with_binding(
            "BUY-4", "1199647000004", account="微信用户", role="角色乙"
        )

        assert self._cluster_of(first) is None

    def test_single_order_of_a_buyer_has_no_cluster(self):
        only = self._order_with_binding(
            "BUY-5", "1199647000005", account="微信用户", role="独苗角色"
        )

        assert self._cluster_of(only) is None

    def test_rows_without_probe_snapshot_are_not_grouped(self):
        """没验证过 CK（绑定账号/角色为空）的行不参与归组，也不会被别人的快照误伤。"""
        verified = self._order_with_binding(
            "BUY-6", "1199647000006", account="微信用户", role="角色丙"
        )
        blank = self._create(
            order_no="BUY-7", shop="佳佳", ck_text=fake_ck("1199647000007")
        )

        clusters = collab.buyer_clusters(self.db, [blank, verified])

        assert clusters == {} or int(blank.id) not in clusters
        assert self._cluster_of(verified) is None
        assert collab.buyer_key_of(blank) is None

    def test_serialize_order_carries_cluster_and_defaults_to_null(self):
        first = self._order_with_binding(
            "BUY-8", "1199647000008", account="微信用户", role="角色丁"
        )
        second = self._order_with_binding(
            "BUY-9", "1199647000009", account="微信用户", role="角色丁"
        )
        clusters = collab.buyer_clusters(self.db, [first, second])

        payload = collab.serialize_order(
            first, buyer_cluster=clusters.get(int(first.id))
        )
        assert payload["buyer_cluster"]["total"] == 2
        assert payload["buyer_cluster"]["method"] == collab.BUYER_KEY_METHOD
        assert payload["buyer_cluster"]["others"][0]["gift_name"] == "礼包A"

        alone = collab.serialize_order(first)
        assert alone["buyer_cluster"] is None


class FakeScoreResp:
    """合成 ``getUserScore`` 响应（积分）。"""

    def __init__(
        self,
        *,
        status: int = 200,
        msg: str = "",
        score: int = 1234,
        gained: int = 5000,
        used: int = 3766,
    ):
        self.status = status
        self.msg = msg
        self.score = score
        self.gainScore = gained
        self.usedScore = used
        self.newGainScore = gained
        self.newUsedScore = used

    @property
    def available_score(self) -> int:
        return max(0, int(self.newGainScore or 0) - int(self.newUsedScore or 0))

    def to_dict(self) -> dict:
        return {
            "status": self.status,
            "msg": self.msg,
            "score": self.score,
            "gain_score": self.gainScore,
            "used_score": self.usedScore,
        }


class FakeRecordsResp:
    """合成 ``getUserPrizeRecords`` 响应（兑换记录）。"""

    def __init__(
        self, *, status: int = 200, msg: str = "", records: list | None = None
    ):
        self.status = status
        self.msg = msg
        self._records = (
            records
            if records is not None
            else [
                {
                    "prize_name": "不朽伊人大礼包",
                    "score": 200,
                    "status_label": "已发放",
                    "order_id": "H-ORDER-1",
                    "exchange_time": 1792000000,
                }
            ]
        )

    def to_dict(self) -> dict:
        return {
            "status": self.status,
            "msg": self.msg,
            "record_count": len(self._records),
            "records": self._records,
        }


class FakeActivityClient:
    """离线「积分 + 兑换记录」客户端：两个接口分别可按脚本给出响应 / 异常。"""

    def __init__(self, *, score: object = None, records: object = None):
        self.score = score
        self.records = records
        self.score_calls: list[dict] = []
        self.record_calls: list[dict] = []

    def query_user_score(self, **kwargs):
        self.score_calls.append(kwargs)
        if isinstance(self.score, Exception):
            raise self.score
        return self.score if self.score is not None else FakeScoreResp()

    def get_user_prize_records(self, **kwargs):
        self.record_calls.append(kwargs)
        if isinstance(self.records, Exception):
            raise self.records
        return self.records if self.records is not None else FakeRecordsResp()


class FakeRechargeClient(FakeProbeClient, FakeActivityClient):
    """充值核账用：同一轮里既要跑绑定探针，又要查积分/礼包目录。

    ``scores`` 按 uid 调用顺序出队；不传就所有 uid 用同一个 ``score``。
    """

    def __init__(
        self,
        *outcomes,
        score: object = None,
        scores: list[object] | None = None,
        points_gifts: list[object] | None = None,
    ):
        FakeProbeClient.__init__(self, *outcomes, points_gifts=points_gifts)
        FakeActivityClient.__init__(self, score=score)
        self.scores = list(scores or [])

    def query_user_score(self, **kwargs):
        if not self.scores:
            return super().query_user_score(**kwargs)
        self.score_calls.append(kwargs)
        outcome = self.scores.pop(0)
        if isinstance(outcome, Exception):
            raise outcome
        return outcome


class TestCollabRecharge(TestCollabBase):
    """充值核账：按 bind_game_account 整组校验 CK + 积分，整组落状态。"""

    BIND = "同一个买家"
    ROLE = "角色A"
    HOME_IMAGE = "collab/2026/10/19/home01.png"

    def _ready(
        self,
        order_no: str,
        uid: str,
        *,
        gift_name: str,
        required_points: int | None,
        role: str = ROLE,
        account: str = BIND,
    ) -> HuyaExternalOrder:
        order = self._create(
            order_no=order_no,
            ck_text=fake_ck(uid),
            gift_name=gift_name,
            required_points=required_points,
            home_images=[self.HOME_IMAGE],
        )
        collab.verify_ck(
            self.db,
            self.user,
            order,
            client=FakeProbeClient(
                FakeBindResp(nick=account, role=role, plat="苹果(iOS)", area="微信")
            ),
        )
        assert order.status == collab.STATUS_TO_RECHARGE
        assert order.bind_game_account == account
        return order

    @staticmethod
    def _score(value: int) -> FakeScoreResp:
        return FakeScoreResp(gained=max(0, value), used=0)

    def _check(self, orders, client, *, threshold: int = 100):
        return collab.check_recharge_groups(
            self.db,
            self.user,
            orders,
            threshold=threshold,
            client=client,
        )

    def test_success_recharges_whole_group_and_snapshots_every_order(self):
        first = self._ready("RC-1", "1001", gift_name="礼包A", required_points=200)
        second = self._ready("RC-2", "1001", gift_name="礼包B", required_points=300)
        client = FakeRechargeClient(
            FakeBindResp(nick=self.BIND, role=self.ROLE),
            score=self._score(550),
        )

        result = self._check([first, second], client)

        assert result["updated"] == 2 and result["failed"] == 0
        assert len(result["groups"]) == 1
        group = result["groups"][0]
        assert group["status"] == collab.STATUS_RECHARGED
        assert group["balance"] == 550
        assert group["required_total"] == 500
        assert group["surplus"] == 50
        assert first.status == second.status == collab.STATUS_RECHARGED
        for order in (first, second):
            assert order.points_checked_at is not None
            assert order.points_balance == 550
            assert order.points_required_total == 500
            assert order.points_surplus == 50
            assert order.points_threshold == 100
            assert order.points_check_reason == "ok"
            assert (
                order.points_check_detail["group_key"] == f"{self.BIND} / {self.ROLE}"
            )
        # 同一 uid 只验一次 CK、只查一次积分
        assert len(client.calls) == 1
        assert len(client.score_calls) == 1
        assert "huya:collab:recharge_check" in [
            row.action for row in self._audits(first)
        ]

    def test_threshold_boundary_is_allowed(self):
        order = self._ready("RC-3", "1002", gift_name="礼包A", required_points=500)
        client = FakeRechargeClient(
            FakeBindResp(nick=self.BIND, role=self.ROLE),
            score=self._score(600),
        )

        result = self._check([order], client, threshold=100)

        assert result["groups"][0]["surplus"] == 100
        assert order.status == collab.STATUS_RECHARGED
        assert order.points_check_reason == "ok"

    def test_shortage_marks_whole_group_points_error(self):
        first = self._ready("RC-4", "1003", gift_name="礼包A", required_points=200)
        second = self._ready("RC-5", "1003", gift_name="礼包B", required_points=300)
        client = FakeRechargeClient(
            FakeBindResp(nick=self.BIND, role=self.ROLE),
            score=self._score(400),
        )

        result = self._check([first, second], client)

        assert result["groups"][0]["reason"] == "shortage"
        assert result["groups"][0]["surplus"] == -100
        assert first.status == second.status == collab.STATUS_POINTS_ERROR
        assert "还差 100 分" in first.points_check_message
        assert first.points_check_reason == second.points_check_reason == "shortage"

    def test_excess_marks_whole_group_points_error(self):
        order = self._ready("RC-6", "1004", gift_name="礼包A", required_points=500)
        client = FakeRechargeClient(
            FakeBindResp(nick=self.BIND, role=self.ROLE),
            score=self._score(650),
        )

        result = self._check([order], client, threshold=100)

        assert result["groups"][0]["reason"] == "excess"
        assert result["groups"][0]["surplus"] == 150
        assert order.status == collab.STATUS_POINTS_ERROR
        assert "多出 150 分" in order.points_check_message

    def test_multiple_uids_are_summed_and_each_queried_once(self):
        first = self._ready("RC-7", "1005", gift_name="礼包A", required_points=300)
        second = self._ready("RC-8", "1006", gift_name="礼包B", required_points=300)
        client = FakeRechargeClient(
            FakeBindResp(nick=self.BIND, role="角色A"),
            FakeBindResp(nick=self.BIND, role="角色A"),
            scores=[self._score(300), self._score(400)],
        )

        result = self._check([first, second], client)

        assert result["groups"][0]["balance"] == 700
        assert result["groups"][0]["surplus"] == 100
        assert first.status == second.status == collab.STATUS_RECHARGED
        assert [call["uid"] for call in client.score_calls] == [1005, 1006]

    def test_same_generic_nickname_with_different_roles_are_separate_groups(self):
        """生产回归：bind_game_account 可能都是「微信用户」，必须再按角色分组。

        只按账号分组会把不同买家合成一组，required_total 被放大成 48000。
        """
        first = self._ready(
            "RC-NICK-1",
            "1013",
            gift_name="礼包A",
            required_points=200,
            role="角色A",
            account="微信用户",
        )
        second = self._ready(
            "RC-NICK-2",
            "1014",
            gift_name="礼包B",
            required_points=200,
            role="角色B",
            account="微信用户",
        )
        client = FakeRechargeClient(
            FakeBindResp(nick="微信用户", role="角色A"),
            FakeBindResp(nick="微信用户", role="角色B"),
            scores=[self._score(250), self._score(300)],
        )

        result = self._check([first, second], client)

        assert len(result["groups"]) == 2
        assert {group["balance"] for group in result["groups"]} == {250, 300}
        assert {group["required_total"] for group in result["groups"]} == {200}
        assert first.status == second.status == collab.STATUS_RECHARGED
        assert first.points_required_total == second.points_required_total == 200
        assert first.points_check_detail["group_key"] == "微信用户 / 角色A"
        assert second.points_check_detail["group_key"] == "微信用户 / 角色B"

    def test_ck_invalid_keeps_group_status_unchanged(self):
        order = self._ready("RC-9", "1007", gift_name="礼包A", required_points=200)
        client = FakeRechargeClient(
            FakeBindResp(status=401, msg="登录态过期"),
            score=self._score(200),
        )

        result = self._check([order], client)

        assert result["failed"] == 1
        assert result["groups"][0]["status"] == ""
        assert "CK 验证失败" in result["groups"][0]["message"]
        assert order.status == collab.STATUS_TO_RECHARGE
        assert order.points_checked_at is None

    def test_points_query_failure_keeps_group_status_unchanged(self):
        order = self._ready("RC-10", "1008", gift_name="礼包A", required_points=200)
        client = FakeRechargeClient(
            FakeBindResp(nick=self.BIND, role=self.ROLE),
            score=RuntimeError("boom"),
        )

        result = self._check([order], client)

        assert result["failed"] == 1
        assert "积分查询失败" in result["groups"][0]["message"]
        assert order.status == collab.STATUS_TO_RECHARGE
        assert order.points_checked_at is None

    def test_missing_required_points_can_be_resolved_from_gift_catalog(self):
        order = self._ready(
            "RC-11", "1009", gift_name="不朽伊人大礼包", required_points=None
        )
        client = FakeRechargeClient(
            FakeBindResp(nick=self.BIND, role=self.ROLE),
            score=self._score(200),
            points_gifts=[FakePointsCenterResp()],
        )

        result = self._check([order], client)

        assert order.required_points == 200
        assert result["groups"][0]["required_total"] == 200
        assert result["groups"][0]["surplus"] == 0
        assert order.status == collab.STATUS_RECHARGED
        assert len(client.points_calls) == 1

    def test_unresolved_required_points_keeps_status_unchanged(self):
        order = self._ready(
            "RC-12", "1010", gift_name="目录里没有的礼包", required_points=None
        )
        client = FakeRechargeClient(
            FakeBindResp(nick=self.BIND, role=self.ROLE),
            score=self._score(200),
            points_gifts=[FakePointsCenterResp()],
        )

        result = self._check([order], client)

        assert result["failed"] == 1
        assert "缺少礼包所需积分" in result["groups"][0]["message"]
        assert order.status == collab.STATUS_TO_RECHARGE
        assert order.required_points is None

    def test_group_order_without_cookie_blocks_whole_group(self):
        order = self._ready("RC-14", "1012", gift_name="礼包A", required_points=200)
        order.cookie = ""
        self.db.commit()

        result = self._check([order], FakeRechargeClient())

        assert result["failed"] == 1
        assert "组内订单缺少 CK" in result["groups"][0]["message"]
        assert order.status == collab.STATUS_TO_RECHARGE

    def test_missing_bind_account_is_demoted_to_incomplete(self):
        order = self._ready("RC-13", "1011", gift_name="礼包A", required_points=200)
        order.bind_game_account = ""
        self.db.commit()

        result = self._check([order], FakeRechargeClient())

        assert result["updated"] == 0
        assert result["failed"] == 1
        assert "缺少绑定账号" in result["failures"][0]["message"]
        assert order.status == collab.STATUS_INCOMPLETE


class TestCollabStatusBatch(TestCollabBase):
    """勾选多行一次改状态：部分失败不影响其它行（2026-10-16 用户要的批量切换）。"""

    def test_batch_updates_selected_rows(self):
        first = self._bind_ready(self._create(order_no="BATCH-1"))
        second = self._bind_ready(self._create(order_no="BATCH-2"))

        result = collab.set_status_batch(
            self.db, self.user, [first, second], "scripted"
        )

        assert result["updated"] == 2 and result["failed"] == 0
        assert first.status == second.status == "scripted"

    def test_batch_reports_per_row_failure_without_touching_others(self):
        ready = self._bind_ready(self._create(order_no="BATCH-3"))
        not_ready = self._create(order_no="BATCH-4")  # 没 CK / 没角色 / 没主页截图

        result = collab.set_status_batch(
            self.db, self.user, [ready, not_ready], "scripted"
        )

        assert result["updated"] == 1 and result["failed"] == 1
        by_no = {item["order_no"]: item for item in result["results"]}
        assert by_no["BATCH-3"]["ok"] is True
        assert by_no["BATCH-4"]["ok"] is False
        assert "外部 CK" in by_no["BATCH-4"]["message"]
        assert ready.status == "scripted"
        assert not_ready.status == collab.STATUS_INCOMPLETE

    def test_batch_rejects_reconcile_statuses(self):
        order = self._bind_ready(self._create(order_no="BATCH-5"))
        for target in (collab.STATUS_RECHARGED, collab.STATUS_POINTS_ERROR):
            with pytest.raises(collab.CollabError) as excinfo:
                collab.set_status_batch(self.db, self.user, [order], target)
            assert excinfo.value.code == "recharge_check_required"
        assert order.status == collab.STATUS_TO_RECHARGE

    def test_batch_backward_needs_rollback_flag(self):
        order = self._bind_ready(self._create(order_no="BATCH-6"))
        collab.set_status(self.db, self.user, order, "scripted")

        rejected = collab.set_status_batch(
            self.db, self.user, [order], collab.STATUS_TO_RECHARGE
        )
        assert rejected["failed"] == 1
        assert order.status == "scripted"

        accepted = collab.set_status_batch(
            self.db,
            self.user,
            [order],
            collab.STATUS_TO_RECHARGE,
            rollback=True,
        )
        assert accepted["updated"] == 1
        assert order.status == collab.STATUS_TO_RECHARGE

    def test_batch_rejects_empty_and_too_many(self):
        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_status_batch(self.db, self.user, [], "scripted")
        assert excinfo.value.code == "status_batch_empty"

        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_status_batch(
                self.db,
                self.user,
                [None] * (collab.PROBE_BATCH_MAX + 1),
                "scripted",
            )
        assert excinfo.value.code == "status_batch_too_large"


class TestCollabArchive(TestCollabBase):
    """归档是可见性状态：不影响业务状态，默认隐藏，归档视图可搜索。"""

    def _done(self, order_no: str, **kwargs) -> HuyaExternalOrder:
        order = self._create(order_no=order_no, **kwargs)
        order.status = collab.STATUS_DONE
        self.db.commit()
        self.db.refresh(order)
        return order

    def test_archive_hides_by_default_but_is_searchable_in_archive_scope(self):
        archived = self._done("ARCH-1", remark="客户已验收")
        active = self._create(order_no="ARCH-2")

        result = collab.archive_orders(
            self.db, self.user, [archived], reason="活动结束，完成对账"
        )

        assert result["archived"] == 1
        assert archived.archived_at is not None
        assert archived.archive_reason == "活动结束，完成对账"
        assert archived.status == collab.STATUS_DONE

        rows, total = collab.list_orders(self.db)
        assert total == 1
        assert [row.id for row in rows] == [active.id]

        rows, total = collab.list_orders(
            self.db, archive_scope="archived", keyword="客户已验收"
        )
        assert total == 1
        assert [row.id for row in rows] == [archived.id]

        default_summary = collab.summary(self.db)
        archived_summary = collab.summary(self.db, archive_scope="archived")
        assert default_summary["by_status"]["done"] == 0
        assert default_summary["archived_total"] == 1
        assert archived_summary["total"] == 1

        audit = self._audits(archived)[-1]
        assert audit.action == "huya:collab:archive"
        assert "完成对账" in audit.detail

    def test_only_done_orders_can_be_archived(self):
        pending = self._create(order_no="ARCH-PENDING")

        result = collab.archive_orders(self.db, self.user, [pending])

        assert result["archived"] == 0
        assert result["skipped"] == 1
        assert "抢购成功" in result["results"][0]["message"]
        assert pending.archived_at is None

    def test_archived_order_is_read_only_until_unarchived(self):
        order = self._done("ARCH-READONLY")
        collab.archive_orders(self.db, self.user, [order])

        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_status(
                self.db,
                self.user,
                order,
                collab.STATUS_INCOMPLETE,
                rollback=True,
            )
        assert excinfo.value.code == "order_archived"

        with pytest.raises(collab.CollabError) as excinfo:
            collab.update_order(
                self.db,
                self.user,
                order,
                {"remark": "归档后不应该能改"},
                expected_version=order.version,
            )
        assert excinfo.value.code == "order_archived"

    def test_unarchive_restores_row_and_clears_archive_note(self):
        order = self._done("ARCH-UNARCHIVE")
        collab.archive_orders(self.db, self.user, [order], reason="临时隐藏")

        result = collab.unarchive_orders(self.db, self.user, [order])

        assert result["unarchived"] == 1
        assert order.archived_at is None
        assert order.archived_by is None
        assert order.archive_reason == ""
        rows, total = collab.list_orders(self.db)
        assert total == 1
        assert [row.id for row in rows] == [order.id]
        assert self._audits(order)[-1].action == "huya:collab:unarchive"

    def test_archive_by_filter_checks_expected_count_and_only_archives_done(self):
        first = self._done("ARCH-FILTER-1", shop="佳佳")
        second = self._done("ARCH-FILTER-2", shop="佳佳")
        self._create(order_no="ARCH-FILTER-PENDING", shop="佳佳")

        with pytest.raises(collab.CollabError) as excinfo:
            collab.archive_orders_by_filter(
                self.db,
                self.user,
                expected_count=1,
                shop="佳佳",
                status=collab.STATUS_DONE,
            )
        assert excinfo.value.code == "archive_count_changed"

        result = collab.archive_orders_by_filter(
            self.db,
            self.user,
            expected_count=2,
            shop="佳佳",
            status=collab.STATUS_DONE,
            reason="批量清理",
        )
        assert result["archived"] == 2
        assert first.archived_at is not None
        assert second.archived_at is not None
        _, total = collab.list_orders(self.db, shop="佳佳", archive_scope="active")
        assert total == 1


class TestCollabActivityLifecycle(TestCollabBase):
    """抢购之后：查积分 + 兑换记录 → 落生命周期快照 → 按结果自动改状态。"""

    def _scripted_order(
        self,
        order_no: str = "ACT-1",
        *,
        gift_name: str = "不朽伊人大礼包",
    ) -> HuyaExternalOrder:
        # 离开「待补资料」要求礼包名非空；这里先用占位礼包推进到已上脚本，
        # 再按需清空礼包名，模拟「上脚本后人工把礼包删了」的脏数据。
        order = self._bind_ready(
            self._create(order_no=order_no, gift_name=gift_name or "占位礼包")
        )
        collab.set_status(self.db, self.user, order, "scripted")
        if not gift_name:
            collab.update_order(self.db, self.user, order, {"gift_name": ""})
        return order

    def _metrics(self, order: HuyaExternalOrder) -> list:
        return collab.list_metrics(self.db, order)

    def test_success_verdict_promotes_scripted_order_to_done(self):
        order = self._scripted_order()
        client = FakeActivityClient()

        result = collab.query_order_activity(self.db, self.user, order, client=client)

        assert result["verdict"] == collab.VERDICT_SUCCESS
        assert result["records_matched"] == 1
        assert result["auto_status"] == "done"
        assert order.status == "done"
        assert order.done_at is not None
        assert client.score_calls[0]["uid"] == 1199647239697
        assert client.record_calls[0]["sid"] == collab.GIFTS_SID

        metrics = self._metrics(order)
        assert len(metrics) == 1
        assert metrics[0].score == 1234
        assert metrics[0].verdict == "success"
        assert metrics[0].records_total == 1
        assert "huya:collab:activity_query" in [
            row.action for row in self._audits(order)
        ]

    def test_missing_gift_marks_failed_without_clearing_scripted_time(self):
        order = self._scripted_order("ACT-2")
        client = FakeActivityClient(
            records=FakeRecordsResp(
                records=[{"prize_name": "别的礼包", "status_label": "已发放"}]
            )
        )

        result = collab.query_order_activity(self.db, self.user, order, client=client)

        assert result["verdict"] == collab.VERDICT_FAILED
        assert result["auto_status"] == "failed"
        assert order.status == "failed"
        # failed 是旁路终态：上脚本的时间戳必须留着（不然不知道什么时候上的）
        assert order.scripted_at is not None
        assert order.done_at is None

        # 从「抢购失败」回流程内要显式回退
        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_status(self.db, self.user, order, "scripted")
        assert excinfo.value.code == "status_not_forward"
        collab.set_status(self.db, self.user, order, "scripted", rollback=True)
        assert order.status == "scripted"

    def test_status_untouched_when_order_is_not_scripted(self):
        order = self._bind_ready(
            self._create(order_no="ACT-3", gift_name="不朽伊人大礼包")
        )
        assert order.status == collab.STATUS_TO_RECHARGE

        result = collab.query_order_activity(
            self.db, self.user, order, client=FakeActivityClient()
        )

        assert result["verdict"] == collab.VERDICT_SUCCESS
        assert result["auto_status"] == ""
        assert order.status == collab.STATUS_TO_RECHARGE
        assert "只有「已上脚本」" in result["message"]

    def test_unknown_verdict_when_interface_fails(self):
        order = self._scripted_order("ACT-4")
        client = FakeActivityClient(
            score=RuntimeError("boom"),
            records=FakeRecordsResp(status=500, msg="活动已结束"),
        )

        result = collab.query_order_activity(self.db, self.user, order, client=client)

        assert result["verdict"] == collab.VERDICT_UNKNOWN
        assert result["auto_status"] == ""
        assert order.status == "scripted"  # 数据不够，不许瞎判
        metrics = self._metrics(order)
        assert len(metrics) == 1 and metrics[0].verdict == "unknown"

    def test_unknown_when_gift_name_missing(self):
        order = self._scripted_order("ACT-5", gift_name="")

        result = collab.query_order_activity(
            self.db, self.user, order, client=FakeActivityClient()
        )

        assert result["verdict"] == collab.VERDICT_UNKNOWN
        assert "没填礼包名" in result["message"]
        assert order.status == "scripted"

    def test_query_requires_cookie_and_rejects_void(self):
        order = self._create(order_no="ACT-6")
        with pytest.raises(collab.CollabError) as excinfo:
            collab.query_order_activity(
                self.db, self.user, order, client=FakeActivityClient()
            )
        assert excinfo.value.code == "no_cookie"

        void = self._scripted_order("ACT-7")
        collab.set_status(self.db, self.user, void, "void", reason="客户退款")
        with pytest.raises(collab.CollabError) as excinfo:
            collab.query_order_activity(
                self.db, self.user, void, client=FakeActivityClient()
            )
        assert excinfo.value.code == "void_order"

    def test_batch_query_isolates_failures(self):
        first = self._scripted_order("ACT-8")
        broken = self._create(order_no="ACT-9")  # 没 CK

        result = collab.query_activity_batch(
            self.db,
            self.user,
            [first, broken],
            client=FakeActivityClient(),
            interval_seconds=0,
        )

        assert result["total"] == 2
        assert result["success"] == 1 and result["unknown"] == 1
        assert result["auto_changed"] == 1
        assert first.status == "done"
        assert broken.status == collab.STATUS_INCOMPLETE
        assert any("还没有 CK" in item["message"] for item in result["results"])

    def test_metrics_timeline_is_newest_first(self):
        order = self._scripted_order("ACT-10")

        collab.query_order_activity(
            self.db,
            self.user,
            order,
            client=FakeActivityClient(
                score=FakeScoreResp(score=1500, gained=5000, used=3500)
            ),
        )
        collab.query_order_activity(
            self.db,
            self.user,
            order,
            client=FakeActivityClient(
                score=FakeScoreResp(score=800, gained=5000, used=4200)
            ),
        )

        metrics = self._metrics(order)
        assert [row.score for row in metrics] == [800, 1500]

        latest = collab.latest_metrics(self.db, [order])[int(order.id)]
        assert latest.score == 800
        payload = collab.serialize_order(order, last_metric=latest)
        assert payload["last_score"] == 800
        assert payload["last_verdict"] == "success"


class TestCollabActivityScopeSplit(TestCollabBase):
    """批量「查积分 / 查兑换记录」拆成两个按钮（2026-10-17 用户要求）。

    重点全在**互相不冲掉**上：

    - 只查积分时：兑换记录一个请求都不发，判定留空（≠「无法判定」），**绝不自动改状态**；
    - 只查兑换记录时：不许把列表上的积分清成空；
    - 列表上的「判定徽标」取**最近一条真的查过兑换记录**的快照 —— 否则点一次
      「查积分」就会把整页「兑换成功 / 兑换失败」冲成「未判定」；
    - 但**新的 unknown（查了判不了）必须照样盖掉旧的 success**，
      不能"为了保住徽标"就把抖动的结果藏起来。
    """

    def _scripted_order(
        self, order_no: str, *, gift_name: str = "不朽伊人大礼包"
    ) -> HuyaExternalOrder:
        order = self._bind_ready(self._create(order_no=order_no, gift_name=gift_name))
        collab.set_status(self.db, self.user, order, "scripted")
        return order

    def _row_payload(self, order: HuyaExternalOrder) -> dict:
        """列表接口给前端的那份输出（走 order_metric_views，与真实路由同一条路径）。"""
        view = collab.order_metric_views(self.db, [order]).get(int(order.id))
        return collab.serialize_order(order, metric_view=view)

    def test_score_scope_only_calls_the_score_api(self):
        order = self._scripted_order("SCOPE-1")
        client = FakeActivityClient()

        result = collab.query_order_activity(
            self.db, self.user, order, scope="score", client=client
        )

        assert len(client.score_calls) == 1
        assert client.record_calls == []  # 兑换记录接口一次都没打
        assert result["scope"] == "score"
        assert result["verdict"] == collab.VERDICT_NOT_QUERIED == ""
        assert result["score"] == 1234
        assert result["ok"] is True
        assert result["message"].startswith("积分 1234")

        metric = collab.list_metrics(self.db, order)[0]
        assert metric.score == 1234
        assert metric.verdict == ""
        assert metric.raw["scope"] == "score"

    def test_score_scope_never_auto_changes_status(self):
        """反向验证：这单的兑换记录里**确实有**它的礼包（旧口径会判「抢购成功」），
        但只查积分时连记录都没查 —— 状态必须一动不动，也不许给出任何判定。"""
        order = self._scripted_order("SCOPE-2")
        client = FakeActivityClient()  # 默认记录里就有「不朽伊人大礼包」

        result = collab.query_order_activity(
            self.db, self.user, order, scope="score", client=client
        )

        assert result["auto_status"] == ""
        assert order.status == "scripted"
        assert order.done_at is None
        assert result["verdict"] == ""

    def test_records_scope_only_calls_the_records_api(self):
        order = self._scripted_order("SCOPE-3")
        client = FakeActivityClient()

        result = collab.query_order_activity(
            self.db, self.user, order, scope="records", client=client
        )

        assert client.score_calls == []  # 积分接口一次都没打
        assert len(client.record_calls) == 1
        assert result["verdict"] == collab.VERDICT_SUCCESS
        assert result["auto_status"] == "done"
        assert order.status == "done"

        metric = collab.list_metrics(self.db, order)[0]
        assert metric.score is None  # 本次没查积分：不许把旧积分抄一份进来冒充
        assert metric.verdict == "success"

    def test_score_only_query_keeps_the_previous_verdict_on_the_list(self):
        order = self._scripted_order("SCOPE-4")
        collab.query_order_activity(
            self.db, self.user, order, scope="records", client=FakeActivityClient()
        )
        assert order.status == "done"

        collab.query_order_activity(
            self.db,
            self.user,
            order,
            scope="score",
            client=FakeActivityClient(
                score=FakeScoreResp(score=200, gained=5000, used=4800)
            ),
        )

        payload = self._row_payload(order)
        assert payload["last_score"] == 200  # 积分是新的
        assert payload["last_verdict"] == "success"  # 判定还是上一次那条
        assert payload["last_verdict_label"] == "兑换成功"
        assert payload["last_verdict_at"] is not None
        assert payload["last_verdict_message"]
        assert "积分 200" in payload["last_metric_message"]

    def test_records_only_query_keeps_the_previous_score_on_the_list(self):
        order = self._scripted_order("SCOPE-5")
        # 注意 FakeScoreResp 的可用积分是按 gained - used 算的，score 参数不生效
        collab.query_order_activity(
            self.db,
            self.user,
            order,
            scope="score",
            client=FakeActivityClient(score=FakeScoreResp(gained=5000, used=4223)),
        )
        assert self._row_payload(order)["last_score"] == 777

        collab.query_order_activity(
            self.db, self.user, order, scope="records", client=FakeActivityClient()
        )

        payload = self._row_payload(order)
        assert payload["last_score"] == 777  # 积分没被这次查询清空
        assert payload["last_verdict"] == "success"  # 判定是这次的

    def test_newer_unknown_verdict_still_overrides_older_success(self):
        """反向验证的另一半：这次接口抖了、**判不了**（unknown），
        列表绝不能继续挂着上一次的「兑换成功」—— 那等于把失败藏起来。"""
        order = self._scripted_order("SCOPE-6")
        collab.query_order_activity(
            self.db, self.user, order, scope="records", client=FakeActivityClient()
        )
        collab.query_order_activity(
            self.db,
            self.user,
            order,
            scope="records",
            client=FakeActivityClient(
                records=FakeRecordsResp(status=500, msg="活动已结束")
            ),
        )

        payload = self._row_payload(order)
        assert payload["last_verdict"] == "unknown"
        assert payload["last_verdict_label"] == "无法判定"

    def test_metric_timeline_reports_which_half_was_queried(self):
        order = self._scripted_order("SCOPE-7")
        collab.query_order_activity(
            self.db, self.user, order, scope="score", client=FakeActivityClient()
        )
        collab.query_order_activity(
            self.db, self.user, order, scope="records", client=FakeActivityClient()
        )

        rows = [
            collab.serialize_metric(row) for row in collab.list_metrics(self.db, order)
        ]
        # 时间线倒序：最近一次在前
        assert [row["scope"] for row in rows] == ["records", "score"]
        assert (
            rows[0]["verdict"] == "success" and rows[0]["verdict_label"] == "兑换成功"
        )
        # 只查积分那条：判定留空 + 明确叫「未判定」，不是「无法判定」
        assert rows[1]["verdict"] == ""
        assert rows[1]["verdict_label"] == "未判定"
        assert rows[1]["records_total"] == 0
        assert rows[1]["verdict_label"] != collab.VERDICT_LABELS[collab.VERDICT_UNKNOWN]

    def test_batch_score_scope_counts_rows_as_not_judged(self):
        first = self._scripted_order("SCOPE-8")
        second = self._scripted_order("SCOPE-9")

        result = collab.query_activity_batch(
            self.db,
            self.user,
            [first, second],
            scope="score",
            client=FakeActivityClient(),
            interval_seconds=0,
        )

        assert result["total"] == 2
        assert result["not_judged"] == 2
        assert result["unknown"] == 0  # 「没查」不算「查了判不了」
        assert result["success"] == 0 and result["failed"] == 0
        assert result["auto_changed"] == 0
        assert first.status == second.status == "scripted"

    def test_batch_rejects_unknown_scope(self):
        order = self._scripted_order("SCOPE-10")

        with pytest.raises(collab.CollabError) as excinfo:
            collab.query_activity_batch(
                self.db,
                self.user,
                [order],
                scope="everything",
                client=FakeActivityClient(),
                interval_seconds=0,
            )

        assert excinfo.value.code == "activity_scope_unknown"

    def test_router_passes_scope_through_and_rejects_unknown(self, monkeypatch):
        order = self._scripted_order("SCOPE-11")
        client = FakeActivityClient()
        monkeypatch.setattr(collab, "_build_probe_client", lambda: client)

        result = collab_api.batch_query_external_order_activity(
            collab_api.HuyaExternalOrderActivityQueryRequest(
                ids=[int(order.id)], scope="score"
            ),
            self.db,
            self.user,
        )

        assert client.record_calls == []
        assert result["scope"] == "score" and result["scope_label"] == "积分"
        assert result["not_judged"] == 1
        assert result["results"][0]["verdict_label"] == "未判定"

        with pytest.raises(collab_api.HTTPException) as excinfo:
            collab_api.batch_query_external_order_activity(
                collab_api.HuyaExternalOrderActivityQueryRequest(
                    ids=[int(order.id)], scope="nope"
                ),
                self.db,
                self.user,
            )
        assert excinfo.value.status_code == 400

    def test_default_scope_still_queries_both(self):
        """默认（不传 scope）保持原行为：老的调用方一个字段不加也照旧。"""
        order = self._scripted_order("SCOPE-12")
        client = FakeActivityClient()

        result = collab.query_order_activity(self.db, self.user, order, client=client)

        assert len(client.score_calls) == 1 and len(client.record_calls) == 1
        assert result["scope"] == "both"
        assert result["verdict"] == collab.VERDICT_SUCCESS
        assert result["auto_status"] == "done"

    def test_new_list_fields_survive_pydantic_serialization(self):
        """过一遍真实的响应模型（坑 #2：字段声明了、`model_dump()` 里没有 = 假绿）。"""
        order = self._scripted_order("SCOPE-13")
        collab.query_order_activity(
            self.db, self.user, order, scope="records", client=FakeActivityClient()
        )
        collab.query_order_activity(
            self.db,
            self.user,
            order,
            scope="score",
            client=FakeActivityClient(score=FakeScoreResp(gained=5000, used=4800)),
        )

        dumped = HuyaExternalOrderListOut(
            items=[collab_api._serialize_one(self.db, order)],
            total=1,
            page=1,
            page_size=50,
        ).model_dump()
        row = dumped["items"][0]

        assert row["last_score"] == 200  # 查积分那次写的
        assert row["last_verdict"] == "success"  # 判定停在查兑换记录那次
        assert row["last_verdict_label"] == "兑换成功"
        assert row["last_verdict_at"] is not None
        assert row["last_verdict_message"]


class TestCollabFrequentGifts(TestCollabBase):
    """「常用礼包」：礼包名默认空白，只在上面留几个用得最多的备选（2026-10-18 用户要求）。

    以前新建会自动带上「上次选的那个」礼包，每开一单都顶着上一次的选择、
    没注意就建错单。现在改成：**字段默认空白**，想要少点两下就点「常用礼包」按钮。
    """

    def test_orders_by_usage_count_then_recency(self):
        for index in range(3):
            self._create(order_no=f"FREQ-A{index}", gift_name="甲礼包")
        self._create(order_no="FREQ-B1", gift_name="乙礼包")
        self._create(order_no="FREQ-B2", gift_name="乙礼包")
        self._create(order_no="FREQ-C1", gift_name="丙礼包")

        rows = collab.frequent_gift_names(self.db)

        assert [row["name"] for row in rows] == ["甲礼包", "乙礼包", "丙礼包"]
        assert [row["used"] for row in rows] == [3, 2, 1]

    def test_blank_gift_name_is_not_a_gift(self):
        self._create(order_no="FREQ-EMPTY1")
        self._create(order_no="FREQ-EMPTY2")
        self._create(order_no="FREQ-ONE", gift_name="真礼包")

        rows = collab.frequent_gift_names(self.db)

        assert [row["name"] for row in rows] == ["真礼包"]

    def test_same_count_falls_back_to_most_recently_used(self):
        self._create(order_no="FREQ-R1", gift_name="老礼包")
        self._create(order_no="FREQ-R2", gift_name="新礼包")

        rows = collab.frequent_gift_names(self.db)

        assert [row["name"] for row in rows] == ["新礼包", "老礼包"]

    def test_limit_caps_the_list(self):
        for index in range(4):
            self._create(order_no=f"FREQ-L{index}", gift_name="同次数礼包")

        rows = collab.frequent_gift_names(self.db, limit=1)

        assert [row["name"] for row in rows] == ["同次数礼包"]

    def test_api_returns_name_and_used(self):
        self._create(order_no="FREQ-API1", gift_name="接口礼包")
        self._create(order_no="FREQ-API2", gift_name="接口礼包")

        rows = collab_api.external_orders_frequent_gifts(
            limit=8, db=self.db, _current=self.user
        )

        assert rows[0]["name"] == "接口礼包"
        assert rows[0]["used"] == 2


class TestCollabCkExport(TestCollabBase):
    """勾选导出 CK（写剪贴板）：凭据外带要留痕，但审计里不许出现 CK 本体。"""

    def test_export_returns_plaintext_and_audits_each_exported_row(self):
        first = self._create(order_no="EXP-1", ck_text=fake_ck("1199647000001"))
        second = self._create(order_no="EXP-2", ck_text=fake_ck("1199647000002"))
        blank = self._create(order_no="EXP-3")  # 没贴 CK

        result = collab.export_cks(self.db, self.user, [first, second, blank])

        assert result["total"] == 3
        assert result["exported"] == 2
        assert [item["order_no"] for item in result["items"]] == ["EXP-1", "EXP-2"]
        # 明文就是库里的那一份（不是脱敏/预览；库里存的是**归一化后**的 CK，
        # 所以跟 `fake_ck()` 的原始串不完全一样：重复的 guid 已被去掉）
        assert result["items"][0]["ck_text"] == first.cookie
        assert "udb_uid=1199647000001" in result["items"][0]["ck_text"]
        assert result["items"][0]["ck_text"].count("guid=") == 1
        assert result["items"][0]["uid"] == "1199647000001"

        # 没 CK 的行显式跳过并回原因，不静默丢
        assert [row["order_no"] for row in result["skipped"]] == ["EXP-3"]
        assert "CK" in result["skipped"][0]["reason"]

    def test_export_audit_never_contains_the_credential_body(self):
        order = self._create(order_no="EXP-4", ck_text=fake_ck("1199647000004"))
        collab.export_cks(self.db, self.user, [order])

        audits = self._audits(order)
        assert audits[-1].action == "huya:collab:ck_export"
        assert audits[-1].username == self.user.username
        assert order.order_no in audits[-1].detail  # 只记「导出了哪一单」
        # 凭据本体（udb_cred / udb_biztoken 那两段超长值）绝不能进审计
        assert "udb_cred" not in audits[-1].detail
        assert "C" * 152 not in audits[-1].detail
        assert "T" * 200 not in audits[-1].detail

    def test_export_skips_blank_rows_without_audit(self):
        blank = self._create(order_no="EXP-5")
        collab.export_cks(self.db, self.user, [blank])
        assert [row.action for row in self._audits(blank)] == ["huya:collab:create"]

    def test_export_keeps_void_and_disabled_rows_with_labels(self):
        """作废 / 停用的单只要还有 CK 就照导 —— 但状态标签要一起给，由人自己判断。"""
        disabled = self._create(order_no="EXP-6", ck_text=fake_ck("1199647000006"))
        collab.set_ck_status(
            self.db, self.user, disabled, "disabled", message="不再使用"
        )
        voided = self._create(order_no="EXP-7", ck_text=fake_ck("1199647000007"))
        collab.set_status(self.db, self.user, voided, "void", reason="客户退单")

        result = collab.export_cks(self.db, self.user, [disabled, voided])

        assert result["exported"] == 2
        by_no = {item["order_no"]: item for item in result["items"]}
        assert by_no["EXP-6"]["ck_status_label"] == "已停用"
        assert by_no["EXP-7"]["status_label"] == "已作废"
        assert by_no["EXP-7"]["status"] == "void"

    def test_export_rejects_empty_and_too_many(self):
        with pytest.raises(collab.CollabError) as excinfo:
            collab.export_cks(self.db, self.user, [])
        assert excinfo.value.code == "ck_export_empty"

        with pytest.raises(collab.CollabError) as excinfo:
            collab.export_cks(self.db, self.user, [None] * (collab.PROBE_BATCH_MAX + 1))
        assert excinfo.value.code == "ck_export_too_large"

    def test_single_row_copy_is_super_admin_only(self):
        """操作列「复制 CK」只给管理员；运营/客服即使有导出权限也不能走这个入口。"""
        order = self._create(order_no="EXP-ADMIN-1", ck_text=fake_ck("1199647000099"))

        with pytest.raises(collab_api.HTTPException) as excinfo:
            collab_api.copy_external_order_ck(order.id, self.db, self.user)
        assert excinfo.value.status_code == 403

        admin = User(username="collab_admin", password_hash="h", role="super_admin")
        self.db.add(admin)
        self.db.commit()
        result = collab_api.copy_external_order_ck(order.id, self.db, admin)
        assert result["exported"] == 1
        assert result["items"][0]["order_no"] == "EXP-ADMIN-1"
        assert result["items"][0]["ck_text"] == order.cookie
        assert self._audits(order)[-1].username == "collab_admin"


class TestProbeCache(TestCollabBase):
    def test_same_ck_probed_twice_only_hits_the_api_once(self, monkeypatch):
        calls: list[dict] = []

        class CountingClient:
            def check_user_bind_game_account(self, **kwargs):
                calls.append(kwargs)
                return FakeBindResp(nick="随遇而安", role="吃鸡199706")

        monkeypatch.setattr(collab, "_build_probe_client", lambda: CountingClient())
        collab._probe_cache.clear()

        order = self._create(order_no="CACHE-1", ck_text=fake_ck("1199647239697"))
        collab.probe_ck(order)
        collab.probe_ck(order)

        assert len(calls) == 1
        collab._probe_cache.clear()

    def test_different_ck_is_probed_separately(self, monkeypatch):
        calls: list[dict] = []

        class CountingClient:
            def check_user_bind_game_account(self, **kwargs):
                calls.append(kwargs)
                return FakeBindResp(nick="随遇而安", role="吃鸡199706")

        monkeypatch.setattr(collab, "_build_probe_client", lambda: CountingClient())
        collab._probe_cache.clear()

        first = self._create(order_no="CACHE-2", ck_text=fake_ck("1001"))
        second = self._create(order_no="CACHE-3", ck_text=fake_ck("1002"))
        collab.probe_ck(first)
        collab.probe_ck(second)

        assert len(calls) == 2
        collab._probe_cache.clear()


class TestCollabPermissions:
    def test_role_defaults_match_the_agreed_matrix(self):
        """运营能看能改能删；客服能看能改但不能删。"""
        for role in ("operation", "support"):
            assert "huya:collab:view" in ROLE_PERMISSIONS[role]
            assert "huya:collab:edit" in ROLE_PERMISSIONS[role]
        assert "huya:collab:manage" in ROLE_PERMISSIONS["operation"]
        assert "huya:collab:manage" not in ROLE_PERMISSIONS["support"]

    def test_upload_permissions_needed_by_the_page(self):
        """页面要上传图 + 预览图，这两个权限必须给到协作的角色。"""
        for role in ("operation", "support"):
            assert "file:view" in ROLE_PERMISSIONS[role]
            assert "file:upload" in ROLE_PERMISSIONS[role]

    def test_ck_export_is_operation_only_by_default(self):
        """导出 CK = 把凭据带出系统：默认只给运营（要发给客服就在权限页单独勾）。"""
        assert "huya:collab:ck_export" in ROLE_PERMISSIONS["operation"]
        assert "huya:collab:ck_export" not in ROLE_PERMISSIONS["support"]
        assert "huya:collab:ck_export" in ROLE_PERMISSIONS["super_admin"]

    def test_collab_module_does_not_touch_huya_accounts(self):
        """边界（设计文档 §2.3）：本模块不建 HuyaAccount、不写工作台。"""
        assert collab.HuyaExternalOrder.__tablename__ == "huya_external_orders"
        # 协作台账模型里没有任何指向 huya_accounts 的外键
        targets = {
            fk.target_fullname for fk in collab.HuyaExternalOrder.__table__.foreign_keys
        }
        assert targets == {"users.id"}
        assert HuyaWorkbenchAccount.__tablename__ == "huya_workbench_accounts"