live-hub-py/tests/test_account_overview.py

"""账号「一键查询」：判定规则、失败隔离、只读性。

判定规则用的是**生产镜像库的真实返回样本**（见 `SAMPLES`），
不是我自己编的 JSON —— 这样官方字段改名/结构变动时用例才会红。
"""

from __future__ import annotations

from datetime import UTC, datetime

from tests.support.db import get_test_engine, make_test_session
from web.backend.models import Account, AccountEvent, DouyuTask, User
from web.backend.services.douyu_runner_overview import (
    DEFAULT_HIGH_VALUE_GOODS,
    OverviewMixin,
    _goods_config,
    extract_high_value_goods,
)

RABBIT = "兔兔白日梦"
RABBIT_ID = "20260120QYOOB_bag10"
#: 账号 15407（未兑换）的真实商品项，取自生产镜像库任务 253579
GOODS_AVAILABLE = {
    "commodityId": RABBIT_ID,
    "commodityName": "套装-兔兔白日梦",
    "count": 96982,
    "countLimited": 1,
    "foreverCount": 96982,
    "openStatus": 1,
    "score": 1280,
    "startTime": "1769153181",
    "endTime": "1795967999",
    "storeStatus": 1,
    "pic": "https://sta-op.douyucdn.cn/x.png",
}
#: 账号 558（已兑换）的同一商品项，只有这三项不同
GOODS_OWNED = {**GOODS_AVAILABLE, "count": 0, "foreverCount": 0, "storeStatus": 0}


class TestOwnedJudgement:
    """「是否已获得」的判定 —— 本功能的核心，也是最容易判错的地方。

    ⚠️ **判定只看官方兑换记录**。第一版拿限兑商品列表的 `count`/`storeStatus` 判，
    依据是「558 兑换过 count=0 / 15407 没兑换 count=96982」——
    那两条记录相隔两个月，我把**时间差异**当成了**账号差异**。

    同一时刻对比才发现：三个账号（两个兑换过、一个没有）的商品项**字段完全一致**，
    `count`/`foreverCount`/`storeStatus` 是**全局库存**，不暴露账号维度信息。
    改用官方兑换记录后完美区分（558/607 有「破片手榴弹-兔兔白日梦」，15507 是 0 条）。
    """

    def test_owned_comes_from_official_records(self):
        got = extract_high_value_goods(
            [GOODS_AVAILABLE], [RABBIT], [{"commodityName": "破片手榴弹-兔兔白日梦"}]
        )
        assert got[0]["owned"] is True
        assert got[0]["obtained_names"] == ["破片手榴弹-兔兔白日梦"]

    def test_not_owned_when_records_empty(self):
        got = extract_high_value_goods([GOODS_AVAILABLE], [RABBIT], [])
        assert got[0]["owned"] is False

    def test_global_stock_never_implies_owned(self):
        """**回归用例**：全局库存为 0 或售罄，都不能推出「这个账号已拥有」。

        这正是第一版踩的坑 —— 用全局字段判账号维度的事。
        """
        sold_out = {**GOODS_AVAILABLE, "count": 0, "foreverCount": 0, "storeStatus": 0}
        got = extract_high_value_goods([sold_out], [RABBIT], [])
        assert got[0]["owned"] is False, "全局售罄不等于该账号拥有"

    def test_stock_and_owned_are_independent(self):
        """展示用的全局信息与账号判定互不影响。"""
        sold_out = {**GOODS_AVAILABLE, "count": 0, "storeStatus": 0}
        got = extract_high_value_goods(
            [sold_out], [RABBIT], [{"commodityName": "套装-兔兔白日梦"}]
        )
        assert got[0]["owned"] is True
        assert got[0]["count"] == 0  # 全局库存如实展示

    def test_records_match_by_substring_family(self):
        """同一件东西有多个商品名（套装- / 破片手榴弹-），都要算数。"""
        got = extract_high_value_goods(
            [GOODS_AVAILABLE],
            [RABBIT],
            [{"commodityName": "破片手榴弹-兔兔白日梦"}, {"commodityName": "别的"}],
        )
        assert got[0]["owned"] is True
        assert len(got[0]["obtained_names"]) == 1

    def test_missing_goods_is_not_silently_dropped(self):
        """既不在商品列表、也不在兑换记录里的，必须留 missing。"""
        got = extract_high_value_goods([GOODS_AVAILABLE], [RABBIT, "不存在的套装"], [])
        assert len(got) == 2
        assert got[1]["missing"] is True
        assert "都没有这个商品" in got[1]["reason"]

    def test_goods_in_records_but_not_list_is_not_missing(self):
        """兑换记录里有、但当前限兑列表里没有 —— 不是 missing，是已获得。

        实测就有这种情况（活动窗口结束但记录仍在），不能误报成「没这个商品」。
        """
        got = extract_high_value_goods(
            [], [RABBIT], [{"commodityName": "套装-兔兔白日梦"}]
        )
        assert got[0]["owned"] is True
        assert got[0]["missing"] is False
        assert got[0]["in_goods_list"] is False

    def test_empty_list_and_no_records_keeps_all_names(self):
        got = extract_high_value_goods([], DEFAULT_HIGH_VALUE_GOODS, [])
        assert [g["name"] for g in got] == list(DEFAULT_HIGH_VALUE_GOODS)
        assert all(g["missing"] for g in got)

    def test_both_suits(self):
        items = [
            {**GOODS_AVAILABLE, "commodityName": "套装-兔兔白日梦"},
            {**GOODS_AVAILABLE, "commodityName": "套装-狂战萌兔", "commodityId": "X"},
        ]
        got = extract_high_value_goods(
            items, list(DEFAULT_HIGH_VALUE_GOODS), [{"commodityName": "套装-狂战萌兔"}]
        )
        assert [g["owned"] for g in got] == [False, True]

    def test_tolerates_garbage(self):
        got = extract_high_value_goods(
            [None, "x", 1, GOODS_AVAILABLE],
            [RABBIT],
            [None, "y", {"commodityName": ""}],
        )
        assert got[0]["owned"] is False

    def test_custom_goods_config_overrides_default(self):
        assert _goods_config({"high_value_goods": ["A", "B"]}) == ["A", "B"]
        assert _goods_config({"high_value_goods": "A,B"}) == ["A", "B"]
        assert _goods_config({"high_value_goods": '["A", "B"]'}) == ["A", "B"]
        assert _goods_config({"high_value_goods": "A，B"}) == ["A", "B"]
        assert _goods_config({}) == list(DEFAULT_HIGH_VALUE_GOODS)
        assert _goods_config({"high_value_goods": []}) == list(DEFAULT_HIGH_VALUE_GOODS)


class _FakeClient:
    """最小假客户端：只实现聚合用到的几个方法。"""

    def __init__(self, *, fail: set[str] | None = None):
        self.fail = fail or set()

    def _maybe(self, key: str):
        if key in self.fail:
            raise RuntimeError(f"模拟 {key} 失败")

    def acf_ccn(self, **_):
        return "ctn"

    def query_limited_goods(self, **_):
        self._maybe("goods")
        return {
            "limited_goods": [
                GOODS_OWNED,
                {**GOODS_AVAILABLE, "commodityName": "套装-狂战萌兔"},
            ]
        }

    def exchange_records(self, **_):
        self._maybe("records")
        return {
            "records": [{"commodityName": "破片手榴弹-兔兔白日梦", "createTime": "1"}]
        }


class _Harness(OverviewMixin):
    """把聚合 handler 单独跑起来：假客户端 + 假的 refresh helper。"""

    def __init__(self, client: _FakeClient, *, fail_helpers: set[str] | None = None):
        self.client = client
        self.fail_helpers = fail_helpers or set()
        self.marked: dict = {}

    def _client(self, cookie: str):
        return self.client

    def _mark_task(self, db, task, status, message, result=None):
        self.marked = {
            "status": status,
            "message": message,
            "result": result,
        }

    def _push_log(self, *a, **k):
        pass

    def _refresh_account_gold_balance(self, client, account):
        if "gold" in self.fail_helpers:
            raise RuntimeError("模拟鱼翅查询失败")
        return {"gold_balance": 38, "exchange_balance": 120}

    def _refresh_account_points(self, client, account, cookie, *, ctn=None):
        if "points" in self.fail_helpers:
            raise RuntimeError("模拟积分查询失败")
        return {
            "points": 990,
            "points_query": {"raw": {"data": {"willExpireTime": "1795967999"}}},
        }

    def _refresh_elite_handbook(self, client, account, *, manual_id, rid=""):
        if "elite" in self.fail_helpers:
            raise RuntimeError("模拟精英查询失败")
        return {
            "elite_opened": True,
            "elite_expire_time": 1795967999,
            "elite_manual_score": 990,
        }

    def _refresh_esports_handbook(self, client, account, *, manual_id):
        if "esports" in self.fail_helpers:
            raise RuntimeError("模拟电竞查询失败（账号没绑电竞手册）")
        return {"esports_manual_type": 1, "esports_manual_score": 10}


class TestOverviewAggregation:
    def setup_method(self):
        self.engine = get_test_engine()
        self.session = make_test_session(self.engine)
        self.user = User(username="ov-admin", password_hash="h", role="super_admin")
        self.session.add(self.user)
        self.session.flush()
        self.account = Account(
            username="ov-acc", password="p", email="o@e.com", email_password="ep"
        )
        self.session.add(self.account)
        self.session.commit()
        self.task = DouyuTask(
            batch_id="ov-batch",
            account_id=self.account.id,
            task_type="query_account_overview",
            handbook_scope="elite",
            status="running",
            message="",
            created_by=self.user.id,
        )
        self.session.add(self.task)
        self.session.commit()

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

    def _run(self, client=None, **kw):
        h = _Harness(client or _FakeClient(), **kw)
        h._execute_query_account_overview(
            self.session,
            self.task,
            self.account,
            "cookie",
            {"manual_id": "M1", "esports_manual_id": "M2", "rid": "R1"},
        )
        return h

    def test_aggregates_assets_and_open_states(self):
        h = self._run()
        r = h.marked["result"]
        assert h.marked["status"] == "success"
        assert r["assets"]["gold_balance"] == 38
        assert r["assets"]["exchange_balance"] == 120
        assert r["assets"]["points"] == 990
        assert r["assets"]["points_expire_at"] == "1795967999"
        assert r["elite"]["opened"] is True
        assert r["esports"]["opened"] is True
        assert r["errors"] == {}

    def test_high_value_goods_are_flagged(self):
        r = self._run().marked["result"]
        goods = {g["name"]: g for g in r["high_value_goods"]}
        assert goods["套装-兔兔白日梦"]["owned"] is True  # 兑换记录里有
        assert goods["套装-狂战萌兔"]["owned"] is False  # 记录里没有
        assert "兔兔白日梦" in r["high_value_goods"][0]["name"]

    def test_single_item_failure_does_not_fail_the_batch(self):
        """账号没绑电竞手册是常态 —— 不该让整次查询失败。"""
        r = self._run(fail_helpers={"esports"}).marked["result"]
        assert r["assets"]["points"] == 990  # 其余照常
        assert "esports" in r["errors"]
        assert r["high_value_goods"]  # 关键信息仍在

    def test_all_failures_still_return_success_with_errors(self):
        r = self._run(fail_helpers={"gold", "points", "elite", "esports"}).marked[
            "result"
        ]
        assert set(r["errors"]) >= {"gold", "points", "elite", "esports"}
        assert self._run(fail_helpers={"gold"}).marked["status"] == "success"

    def test_missing_manual_id_is_reported_not_crashed(self):
        h = _Harness(_FakeClient())
        h._execute_query_account_overview(
            self.session, self.task, self.account, "cookie", {}
        )
        r = h.marked["result"]
        assert "未配置精英宝典 manualID" in r["errors"]["elite"]
        assert "未配置精英宝典 manualID" in r["errors"]["high_value_goods"]
        assert h.marked["status"] == "success"
        assert r["high_value_goods"] == []

    def test_message_counts_owned_suits(self):
        assert "兔兔白日梦" in self._run().marked["message"]
        assert "已拥有 1 个" in self._run().marked["message"]

    def test_is_read_only_no_events_and_no_status_change(self):
        """纯查询不是「事实」—— 不能进时间线，也不能改生命周期状态。"""
        before = self.session.query(AccountEvent).count()
        status_before = self.account.status
        self._run()
        self.session.commit()
        assert self.session.query(AccountEvent).count() == before, "一键查询不该发事件"
        self.session.refresh(self.account)
        assert self.account.status == status_before


class TestOverviewTaskTypeWiring:
    def test_declared_for_all_workbenches(self):
        from web.backend.services.douyu_service import (
            DOUYU_HANDBOOK_TASK_TYPES,
            SUPPORTED_DOUYU_TASK_TYPES,
        )

        assert "query_account_overview" in SUPPORTED_DOUYU_TASK_TYPES
        for scope in ("elite", "esports", "peace"):
            assert "query_account_overview" in DOUYU_HANDBOOK_TASK_TYPES[scope]

    def test_handler_is_dispatched(self):
        from web.backend.services.douyu_runner import DouyuBatchRunner

        handler = getattr(DouyuBatchRunner, "_execute_query_account_overview", None)
        assert callable(handler)

    def test_batch_creates_one_task_per_account(self):
        """批量是天然支持的：`create_douyu_planned_tasks` 就是每账号一条任务。"""
        from tests.support.db import get_test_engine as eng
        from web.backend.services.douyu_service import create_douyu_planned_tasks

        session = make_test_session(eng())
        try:
            # 自己造夹具，别依赖别的用例留下的数据（依赖就会 skip，skip 等于没测）
            user = User(
                username="ov-batch-admin", password_hash="h", role="super_admin"
            )
            session.add(user)
            session.flush()
            accs = [
                Account(
                    username=f"ov-batch-{i}",
                    password="p",
                    email=f"b{i}@e.com",
                    email_password="ep",
                    # 必须带 CK：一键查询要用该账号的 CK 去查斗鱼，
                    # `visible_douyu_task_accounts` 会把没 CK 的账号过滤掉。
                    cookie=f"acf_uid=1{i}; acf_username=u{i}; acf_ccn=c{i}",
                )
                for i in range(3)
            ]
            session.add_all(accs)
            session.commit()
            batch_id, count = create_douyu_planned_tasks(
                session,
                [a.id for a in accs],
                "query_account_overview",
                "elite",
                user.id,
            )
            assert count == len(accs)
            created = (
                session.query(DouyuTask)
                .filter(
                    DouyuTask.batch_id == batch_id,
                    DouyuTask.task_type == "query_account_overview",
                )
                .all()
            )
            assert len(created) == len(accs)
        finally:
            session.close()


def test_queried_at_is_utc_isoformat():
    got = extract_high_value_goods([GOODS_OWNED], [RABBIT])
    assert got and isinstance(got, list)
    # 顺手确认 UTC 约定没被破坏
    assert datetime.now(UTC).tzinfo is not None