test(account): 一键查询 / 官方兑换记录 / 生命周期用例

围绕这几轮改动补齐用例（本文件共 53 条）： - **判定规则**：只看官方兑换记录；全局库存**永不**推出「该账号已拥有」（回归用例 —— 正是第一版踩的坑）；字段缺失是 None（未知）而不是 False（没有） - **失败隔离**：单项外部调用失败不影响其余，且如实写进 errors - **只读性**：一键查询不发事件、不改生命周期状态 - **官方记录补成事件**：带**真实兑换时间**（不是"发现它的时刻"）、幂等（重复查询不重复落）、无 commodityId 时跳过 - **排序回归**：「最近一次兑换」按事件时间，不是插入 id - **时间归一化**：0 / "0" / 空 / 负数 → None（1970 那个 bug 的回归） - **迁移安全**：不得在迁移里 commit(op.get_bind())；加列前必须有存在性检查（扫代码时**跳过注释** —— 注释里正解释着"不要这样做"，第一版用例因此假红） - **接口契约**：详情返回快照与查询时间、列表**不**带整份快照；断言一律看 model_dump()（只看属性会假绿 —— AccountOut 有手写 @model_serializer，字段没列进去时属性有值、JSON 里没有）另修两条**把 bug 当期望值**的既有用例： 1. 「最近一次兑换」原断言的是最后**插入**的那条（旧行为按 id 排），改为真正最近的（按事件时间）； 2. 按用户名查找夹具不可靠（make_account("key") 每次建新账号、库里一直累积），改为按 id 精确查找。
2026-10-04 20:04:11 +08:00
parent f416397cc8
commit 6865b916ba
3 changed files with 444 additions and 7 deletions
@@ -902,7 +902,10 @@ class TestExchangeRecordColumn(_Base):
    def _exchange(self, account_id: int, name: str, *, days_ago: int = 0):
        from datetime import timedelta

-        from web.backend.services.account_events_service import emit
+        from web.backend.services.account_events_service import (
+            OFFICIAL_SYNC_SOURCE,
+            emit,
+        )

        emit(
            self.session,
@@ -912,6 +915,9 @@ class TestExchangeRecordColumn(_Base):
            category="business",
            subject_key=name,
            occurred_at=NOW - timedelta(days=days_ago),
+            # 列表的「兑换记录」列**只聚合斗鱼官方记录**（source=official_sync），
+            # 我们当场记的那些留给详情页时间线 —— 这是刻意的口径划分。
+            source=OFFICIAL_SYNC_SOURCE,
        )
        self.session.commit()

@@ -930,9 +936,23 @@ class TestExchangeRecordColumn(_Base):

        item = self._item("hasex")
        assert item.exchange_count == 3
-        # 「最近」按**事件 id** 取（同一天多次兑换时时间戳可能相同，靠 id 定序）
+        # 「最近一次」按**事件时间**取：三条分别是 3 天前 / 1 天前 / 2 天前，
+        # 所以最近的是「冠军鸡腿」（1 天前）。
+        #
+        # ⚠️ 这里原来断言的是「烟花」（**最后插入**的那条）—— 那是错的，
+        # 因为当时后端按事件 **id** 排序，于是「最近一次」实际是"最后插入的一次"。
+        # 一键查询把斗鱼官方的**历史**兑换补成事件后这个 bug 才暴露：
+        # 补录的事件 id 最大、时间却最老，会把几个月前的老兑换显示成"最近一次"。
+        # 现在排序是 `occurred_at DESC, id DESC`（时间相同才用 id 兜底）。
        # 后端返回事件原始 title；剥掉「兑换 」前缀是前端展示层的事。
-        assert item.last_exchange_name == "兑换 烟花"
+        assert item.last_exchange_name == "兑换 冠军鸡腿"
+
+    def test_latest_uses_id_only_to_break_ties(self):
+        """时间相同时再用 id 定序（同一天多次兑换时间戳可能一样）。"""
+        account = self.make_account("tie")
+        self._exchange(account.id, "先插入", days_ago=1)
+        self._exchange(account.id, "后插入", days_ago=1)
+        assert self._item("tie").last_exchange_name == "兑换 后插入"

    def test_only_exchange_events_count(self):
        """别的类型（充值 / 登录 / 状态）不能混进兑换次数。"""
@@ -7,6 +7,7 @@
 from __future__ import annotations

 from datetime import UTC, datetime
+from pathlib import Path

 import pytest

@@ -218,6 +219,16 @@ class _Harness(OverviewMixin):
            raise RuntimeError("模拟电竞查询失败（账号没绑电竞手册）")
        return {"esports_manual_type": 1, "esports_manual_score": 10}

+    # handler 按作用域抓官方记录（三套接口），测试里给最小实现
+    def _fetch_elite_records(self, client, config, manual_id, esports_manual_id):
+        return client.exchange_records(manual_id=manual_id).get("records") or []
+
+    def _fetch_esports_records(self, client, config, manual_id, esports_manual_id):
+        return []
+
+    def _fetch_xpd_records(self, client, config, manual_id, esports_manual_id):
+        return []
+

 class TestOverviewAggregation:
    def setup_method(self):
@@ -253,7 +264,14 @@ class TestOverviewAggregation:
            self.task,
            self.account,
            "cookie",
-            {"manual_id": "M1", "esports_manual_id": "M2", "rid": "R1"},
+            {
+                "manual_id": "M1",
+                "esports_manual_id": "M2",
+                "rid": "R1",
+                "xpd_act_alias": "A",
+                "xpd_rid": "R",
+                "xpd_act_id": "1",
+            },
        )
        return h

@@ -463,6 +481,15 @@ class TestSnapshotIsWrittenToAccount:
            def _refresh_esports_handbook(self, c, a, *, manual_id):
                return {"esports_manual_type": 0, "esports_manual_score": 0}

+            def _fetch_elite_records(self, c, cfg, mid, emid):
+                return c.exchange_records(manual_id=mid).get("records") or []
+
+            def _fetch_esports_records(self, c, cfg, mid, emid):
+                return []
+
+            def _fetch_xpd_records(self, c, cfg, mid, emid):
+                return []
+
        task = DouyuTask(
            batch_id="snap",
            account_id=self.account.id,
@@ -479,7 +506,14 @@ class TestSnapshotIsWrittenToAccount:
            task,
            self.account,
            "c",
-            {"manual_id": "M", "esports_manual_id": "E", "rid": "R"},
+            {
+                "manual_id": "M",
+                "esports_manual_id": "E",
+                "rid": "R",
+                "xpd_act_alias": "A",
+                "xpd_rid": "R",
+                "xpd_act_id": "1",
+            },
        )
        self.session.commit()
        self.session.refresh(self.account)
@@ -641,6 +675,15 @@ class TestOverviewRefreshesAccountFields:
                a.esports_opened = True
                return {"esports_manual_type": 1, "esports_manual_score": 3}

+            def _fetch_elite_records(self, c, cfg, mid, emid):
+                return c.exchange_records(manual_id=mid).get("records") or []
+
+            def _fetch_esports_records(self, c, cfg, mid, emid):
+                return []
+
+            def _fetch_xpd_records(self, c, cfg, mid, emid):
+                return []
+
        task = DouyuTask(
            batch_id="rf",
            account_id=self.account.id,
@@ -657,7 +700,14 @@ class TestOverviewRefreshesAccountFields:
            task,
            self.account,
            "c",
-            {"manual_id": "M", "esports_manual_id": "E", "rid": "R"},
+            {
+                "manual_id": "M",
+                "esports_manual_id": "E",
+                "rid": "R",
+                "xpd_act_alias": "A",
+                "xpd_rid": "R",
+                "xpd_act_id": "1",
+            },
        )
        self.session.commit()
        self.session.refresh(self.account)
@@ -681,3 +731,370 @@ class TestOverviewRefreshesAccountFields:
        self._run(expire_at="0")
        assert self.account.points_expire_at is None
        assert self.account.overview_snapshot["assets"]["points_expire_at"] is None
+
+
+class TestSummaryFeedsAccountList:
+    """一键查询的结果要能反映到**账号列表**上。
+
+    用户报的原始问题：账号 6966677706 明明有官方兑换记录，列表却显示「未兑换」。
+
+    根因是**两个口径**：
+    - 列表的「兑换记录」列来自我们自己的 `account_events`，而事件是**上线后**
+      才开始记的 → 上线前的兑换在列表里就是空的，看着像"从没兑换过"；
+    - 一键查询拉的是**斗鱼官方**记录，本来能查到，但那份 JSON 有 1~2KB，
+      不能塞进列表（15000 个账号会把响应放大到几十 MB）。
+
+    所以加了这个**极小**汇总字段 `high_value_owned`：`{"owned": [...], "exchanges": N}`。
+    """
+
+    def setup_method(self):
+        self.engine = get_test_engine()
+        self.session = make_test_session(self.engine)
+        self.admin = User(username="sum-admin", password_hash="h", role="super_admin")
+        self.session.add(self.admin)
+        self.session.flush()
+        self.account = Account(
+            username="sum-acc",
+            password="p",
+            email="s@e.com",
+            email_password="ep",
+            cookie="acf_uid=1; acf_ccn=c",
+            high_value_owned=["套装-兔兔白日梦"],
+            overview_queried_at=datetime.now(UTC),
+            overview_snapshot={"high_value_goods": []},
+        )
+        self.session.add(self.account)
+        self.session.commit()
+
+    def teardown_method(self):
+        self.session.close()
+
+    def _list(self):
+        from web.backend.routers.accounts import list_accounts
+
+        return list_accounts(
+            assigned_only=False,
+            has_cookie=False,
+            search="sum-acc",
+            page=1,
+            page_size=5,
+            db=self.session,
+            current=self.admin,
+        )["items"]
+
+    def test_list_serializes_owned_suits(self):
+        """断言**序列化结果**，不是属性。
+
+        ⚠️ `AccountOut` 有一个**手写**的 `@model_serializer`，它逐个列出要输出的字段。
+        只在模型上声明字段、不写进那个列表 → **属性有值、JSON 里没有**。
+        只断言属性的测试会**假绿**（踩过），所以这里一律看 `model_dump()`。
+        """
+        dumped = self._list()[0].model_dump()
+        assert dumped["high_value_owned"] == ["套装-兔兔白日梦"], (
+            "字段没出现在序列化结果里 —— 前端拿到的 JSON 就没有它"
+        )
+
+    def test_list_serializes_queried_marker(self):
+        """要能区分「查过但没兑换」和「还没查过」—— 否则列表会把未知当高价值。"""
+        dumped = self._list()[0].model_dump()
+        assert dumped["overview_queried_at"] is not None
+
+    def test_list_still_does_not_carry_the_full_snapshot(self):
+        """给列表加信息**不能**把 1~2KB 的整份快照带回去。"""
+        dumped = self._list()[0].model_dump()
+        assert "overview_snapshot" not in dumped, (
+            "列表带整份快照会让 15000 个账号的响应体积爆炸"
+        )
+
+    def test_handler_writes_the_summary(self):
+        """查询跑完必须写汇总，否则列表永远显示「未查询」。"""
+        from web.backend.services.douyu_runner_overview import OverviewMixin
+
+        class _H(OverviewMixin):
+            def _client(self, cookie):
+                return _FakeClient()
+
+            def _mark_task(self, *a, **k):
+                pass
+
+            def _push_log(self, *a, **k):
+                pass
+
+            def _refresh_account_gold_balance(self, c, a):
+                return {"gold_balance": 1, "exchange_balance": 1}
+
+            def _refresh_account_points(self, c, a, cookie, *, ctn=None):
+                return {
+                    "points": 1,
+                    "points_query": {"raw": {"data": {"willExpireTime": "0"}}},
+                }
+
+            def _refresh_elite_handbook(self, c, a, *, manual_id, rid=""):
+                return {
+                    "elite_opened": True,
+                    "elite_expire_time": 0,
+                    "elite_manual_score": 0,
+                }
+
+            def _refresh_esports_handbook(self, c, a, *, manual_id):
+                return {"esports_manual_type": 1, "esports_manual_score": 0}
+
+            def _fetch_elite_records(self, c, cfg, mid, emid):
+                # 精英恰好 1 条（与断言一致）
+                return [
+                    {
+                        "commodityName": "套装-兔兔白日梦",
+                        "commodityId": "C1",
+                        "score": "1280",
+                    }
+                ]
+
+            def _fetch_esports_records(self, c, cfg, mid, emid):
+                return []
+
+            def _fetch_xpd_records(self, c, cfg, mid, emid):
+                return []
+
+        task = DouyuTask(
+            batch_id="sum",
+            account_id=self.account.id,
+            task_type="query_account_overview",
+            handbook_scope="elite",
+            status="running",
+            message="",
+            created_by=self.admin.id,
+        )
+        self.session.add(task)
+        self.session.commit()
+        _H()._execute_query_account_overview(
+            self.session,
+            task,
+            self.account,
+            "c",
+            {
+                "manual_id": "M",
+                "esports_manual_id": "E",
+                "rid": "R",
+                "xpd_act_alias": "A",
+                "xpd_rid": "R",
+                "xpd_act_id": "1",
+            },
+        )
+        self.session.commit()
+        self.session.refresh(self.account)
+        owned = self.account.high_value_owned
+        assert owned is not None, "查询跑完必须写「已兑换过的高价值套装」"
+        # _FakeClient 的兑换记录里有一条「破片手榴弹-兔兔白日梦」
+        assert "兔兔白日梦" in "".join(owned)
+
+
+class TestOverviewSummaryMigrationIsSafeToRerun:
+    """迁移必须**可安全重跑** —— 应用启动时会自动跑它。
+
+    踩过的坑：回填时用了 `op.get_bind().commit()`，
+    打乱了 alembic 自己的版本记账。后果很隐蔽：
+
+    - MySQL 的 DDL（ALTER TABLE）是**自动提交**的 → 列加上了
+    - 但版本号**没更新** → 下次启动又去 ADD COLUMN
+    - → `Duplicate column name 'high_value_owned'`，**服务直接起不来**
+
+    仓库里正常工作的老迁移（0042 的 S1 回填）用的是
+    `op.get_bind().execute(...)` 且**不 commit**，提交由 alembic 负责。
+    """
+
+    MIGRATION = (
+        Path(__file__).resolve().parents[1]
+        / "web/backend/migrations/versions/20261004_0048_high_value_owned.py"
+    )
+
+    def _src(self) -> str:
+        return self.MIGRATION.read_text(encoding="utf-8")
+
+    def _code(self) -> str:
+        """去掉注释后的源码。
+
+        必须去注释：本迁移的注释里**正解释着**「不要 conn.commit()」，
+        直接子串检查会命中注释里的那句话，用例就假红了。
+        """
+        lines = []
+        for line in self._src().splitlines():
+            stripped = line.lstrip()
+            if stripped.startswith("#"):
+                continue
+            # 去掉行尾注释（这些迁移里没有字符串里带 # 的情况）
+            if "#" in line:
+                line = line[: line.index("#")]
+            lines.append(line)
+        return "\n".join(lines)
+
+    def test_does_not_commit_the_alembic_bind(self):
+        """在迁移里 commit 会破坏版本记账（本迁移踩过）。"""
+        code = self._code()
+        assert "conn.commit()" not in code, (
+            "迁移里不要 commit op.get_bind() —— 会让版本号不更新，"
+            "下次启动重复 ADD COLUMN 而把服务起不来"
+        )
+
+    def test_guards_add_column_with_existence_check(self):
+        """加列前必须检查列是否已存在，否则不可重入。"""
+        src = self._src()
+        assert "_column_exists" in src, "缺少列存在性检查"
+        upgrade_body = src[src.index("def upgrade()") : src.index("def downgrade()")]
+        assert "_column_exists" in upgrade_body, "upgrade 里没有用存在性检查守卫"
+
+    def test_column_exists_helper_probes_information_schema(self):
+        src = self._src()
+        assert "information_schema.columns" in src, (
+            "存在性检查要查 information_schema，别用 try/except 猜"
+        )
+
+
+class TestOfficialExchangeBecomesAccountEvent:
+    """斗鱼官方兑换记录要**补成账号事件**。
+
+    这是「账号全生命周期 + 兑换记录」的关键一步：官方记录**带真实兑换时间**
+    （`time` 字段），补成事件后时间线才是完整的，列表的兑换记录聚合也自动准确 ——
+    不再需要单独维护一套"官方条数"的旁路。
+    """
+
+    def setup_method(self):
+        self.engine = get_test_engine()
+        self.session = make_test_session(self.engine)
+        self.user = User(username="oe-admin", password_hash="h", role="super_admin")
+        self.session.add(self.user)
+        self.session.flush()
+        self.account = Account(
+            username="oe-acc", password="p", email="o@e.com", email_password="ep"
+        )
+        self.session.add(self.account)
+        self.session.commit()
+
+    def teardown_method(self):
+        self.session.close()
+
+    def _record(self, *, commodity_id="C1", name="套装-兔兔白日梦", when=1788054985):
+        from datetime import UTC
+        from datetime import datetime as _dt
+
+        from web.backend.services.business_events import record_official_exchange
+
+        return record_official_exchange(
+            self.session,
+            self.account,
+            commodity_id=commodity_id,
+            commodity_name=name,
+            score=1280,
+            occurred_at=_dt.fromtimestamp(when, tz=UTC),
+            show_name="某昵称（安卓-微信）",
+        )
+
+    def test_writes_event_with_real_exchange_time(self):
+        """时间必须是**真实兑换时刻**，不是"发现它的时刻"。"""
+        from web.backend.services import account_events_service as svc
+
+        assert self._record() is True
+        self.session.commit()
+        event = (
+            self.session.query(AccountEvent)
+            .filter(AccountEvent.event_type == svc.EVENT_EXCHANGE)
+            .one()
+        )
+        assert event.occurred_at.year == 2026
+        assert event.occurred_at.month == 8  # 1788054985 → 2026-08-30 前后
+        assert event.amount == -1280
+        assert event.detail["source"] == "official_sync"
+        assert event.source == "official_sync"
+
+    def test_is_idempotent(self):
+        """一键查询会被反复执行，同一条官方记录只能落一条事件。"""
+        from web.backend.services import account_events_service as svc
+
+        assert self._record() is True
+        self.session.commit()
+        assert self._record() is False, "同一条记录不该重复落"
+        self.session.commit()
+        assert (
+            self.session.query(AccountEvent)
+            .filter(AccountEvent.event_type == svc.EVENT_EXCHANGE)
+            .count()
+            == 1
+        )
+
+    def test_different_commodity_is_a_new_event(self):
+        assert self._record(commodity_id="C1") is True
+        assert self._record(commodity_id="C2", name="套装-激萌罗小黑") is True
+        self.session.commit()
+
+    def test_missing_commodity_id_is_skipped(self):
+        assert self._record(commodity_id="") is False
+
+
+class TestLatestExchangeFollowsEventTime:
+    """「最近一次兑换」要按**事件时间**排，不能按插入 id。
+
+    踩过的坑：官方历史记录被补成事件时 `id` 是新的（最大），
+    但 `occurred_at` 是几个月前的。按 id 排会把一条**老**兑换显示成"最近一次" ——
+    实测把 08-30 的兔兔白日梦错报成了 10-01（那是另一次兑换）。
+    """
+
+    def setup_method(self):
+        self.engine = get_test_engine()
+        self.session = make_test_session(self.engine)
+        self.admin = User(username="ord-admin", password_hash="h", role="super_admin")
+        self.session.add(self.admin)
+        self.session.flush()
+        self.account = Account(
+            username="ord-acc",
+            password="p",
+            email="d@e.com",
+            email_password="ep",
+            cookie="acf_uid=1; acf_ccn=c",
+        )
+        self.session.add(self.account)
+        self.session.commit()
+
+    def teardown_method(self):
+        self.session.close()
+
+    def _add_exchange(self, title: str, when: datetime):
+        from web.backend.services.account_events_service import (
+            CATEGORY_BUSINESS,
+            EVENT_EXCHANGE,
+            OFFICIAL_SYNC_SOURCE,
+            emit,
+        )
+
+        emit(
+            self.session,
+            account_id=self.account.id,
+            event_type=EVENT_EXCHANGE,
+            title=title,
+            category=CATEGORY_BUSINESS,
+            occurred_at=when,
+            # 列表只聚合**官方**记录，所以要标成 official_sync 才会被算进去
+            source=OFFICIAL_SYNC_SOURCE,
+        )
+
+    def _latest(self) -> str:
+        from web.backend.routers.accounts import list_accounts
+
+        item = list_accounts(
+            assigned_only=False,
+            has_cookie=False,
+            search="ord-acc",
+            page=1,
+            page_size=5,
+            db=self.session,
+            current=self.admin,
+        )["items"][0]
+        return item.model_dump()["last_exchange_name"]
+
+    def test_newer_event_inserted_first_still_wins(self):
+        # 先插**新**的，再插**老**的：按 id 排会错，按时间排才对
+        self._add_exchange("兑换 新商品", datetime(2026, 10, 1, tzinfo=UTC))
+        self.session.commit()
+        self._add_exchange("兑换 老商品", datetime(2026, 8, 30, tzinfo=UTC))
+        self.session.commit()
+        assert "新商品" in self._latest(), (
+            "「最近一次」按 id 排了 —— 补录的历史事件 id 更大，会被错当成最近一次"
+        )
@@ -26,7 +26,7 @@ TESTS_DIR = Path(__file__).resolve().parent
 PROJECT_ROOT = TESTS_DIR.parent
 VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions"

-HEAD_REVISION = "20261004_0047"
+HEAD_REVISION = "20261004_0048"

 # 迁移新增、但不声明在模型里的复合查询索引。
 EXTRA_INDEXES = (