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 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,
    HuyaExternalOrder,
    HuyaWorkbenchAccount,
    User,
)
from web.backend.permissions import ROLE_PERMISSIONS
from web.backend.services import huya_collab_service as collab


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:
        """把一行推到「可以标记已绑定」的状态。"""
        collab.set_ck(self.db, self.user, order, fake_ck("1199647239697"))
        collab.update_order(
            self.db, self.user, order, {"bind_douyu_account": "用户4241534208"}
        )
        collab.set_images(
            self.db,
            self.user,
            order,
            "bind_images",
            ["collab/2026/10/16/bind01.png"],
        )
        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 == "pending"
        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_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_create_rejects_uid_already_used_by_another_order(self):
        self._create(order_no="A1", ck_text=fake_ck("1001"))
        with pytest.raises(collab.CollabError) as excinfo:
            self._create(order_no="A2", ck_text=fake_ck("1001"))
        assert excinfo.value.status_code == 409
        assert excinfo.value.code == "uid_conflict"
        # 消息里要说清被谁占了，不然客服不知道该找哪一行
        assert "A1" in excinfo.value.detail

    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_cannot_skip_forward_without_explicit_rollback(self):
        """pending 只能走到 bound，直接跳到 recharged 要被挡下。"""
        order = self._create()
        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_status(self.db, self.user, order, "recharged")
        assert excinfo.value.code == "status_not_forward"
        assert order.status == "pending"

    def test_bound_requires_ck_bind_account_and_image(self):
        order = self._create()
        with pytest.raises(collab.CollabError) as excinfo:
            collab.set_status(self.db, self.user, order, "bound")
        message = excinfo.value.detail
        # 错误信息必须说清缺什么，否则客服只能猜
        assert "CK" in message
        assert "绑定斗鱼号" in message
        assert "绑定图" in message

    def test_forward_flow_sets_timestamps(self):
        order = self._bind_ready(self._create())

        collab.set_status(self.db, self.user, order, "bound")
        assert order.status == "bound"
        assert order.bound_at is not None

        collab.set_status(self.db, self.user, order, "recharged")
        collab.set_status(self.db, self.user, order, "scripted")
        collab.set_status(self.db, self.user, order, "done")

        assert order.status == "done"
        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))
        for target in ("bound", "recharged", "scripted", "done"):
            collab.set_status(self.db, self.user, order, target)
        after = self._audits(order)
        assert len(after) - before == 4
        status_audits = [row for row in after if row.action == "huya:collab:status"]
        assert len(status_audits) == 4

    def test_rollback_requires_flag_and_clears_later_timestamps(self):
        order = self._bind_ready(self._create())
        for target in ("bound", "recharged", "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, "recharged")

        collab.set_status(self.db, self.user, order, "recharged", rollback=True)
        assert order.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, "pending", 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, "pending") == {}
        # 幂等点击不该刷出审计噪声
        assert len(self._audits(order)) == before

    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


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 FakeProbeClient:
    """离线探针客户端：按脚本依次给出 ``None`` / 异常 / 响应，并记录调用参数。"""

    def __init__(self, *outcomes):
        self.outcomes = list(outcomes)
        self.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


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_bound_precondition(self):
        """只验证过 CK、读到绑定角色的行，不该再被「绑定斗鱼号」挡在 bound 外面。"""
        order = self._row_with_ck()
        collab.set_images(
            self.db,
            self.user,
            order,
            "bind_images",
            ["collab/2026/10/16/bind01.png"],
        )
        collab.verify_ck(
            self.db,
            self.user,
            order,
            client=FakeProbeClient(FakeBindResp(nick="随遇而安", role="吃鸡199706")),
        )
        assert order.bind_douyu_account == ""

        collab.set_status(self.db, self.user, order, "bound")

        assert order.status == "bound"
        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 失效不改业务状态：已绑定的单子业务上可能仍然有效（设计文档 §4.2）。"""
        order = self._row_with_ck()
        self._bind_ready(order)
        collab.set_status(self.db, self.user, order, "bound")

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

        assert order.ck_status == "invalid"
        assert order.status == "bound"
        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 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_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"