"""S3 回归：账号生命周期状态。 本文件守住四件事： 1. **推导正确**：`derive_status` 是真值表 —— 底层判据的每种组合对应哪个状态； 2. **人工态优先**：`disabled` / `archived` 不被自动推导冲掉； 3. **收口幂等**：状态没变时 `sync_status` 不写库、不发事件； 4. **全生命周期闭环**：导入 → 登录 → 判失效 → 重登 → 转人工 → 人工救回 → 清 CK。 外加「停用/归档的账号不参与任何自动化」这条硬约束（S3 之前是个真缺口： 作业只按底层判据选行，停用的账号照样会被自动重登）。 """ from datetime import UTC, datetime, timedelta from typing import ClassVar import pytest from tests.support.db import get_test_engine, make_test_session from web.backend.models import Account, AccountEvent, AutomationJob, User from web.backend.services import account_lifecycle as life from web.backend.services.automation.jobs import ck_health, ck_relogin NOW = datetime(2026, 10, 4, 12, 0, 0, tzinfo=UTC) class _Base: def setup_method(self): self.engine = get_test_engine() self.session = make_test_session(self.engine) self.user = User(username="life-admin", password_hash="h", role="super_admin") self.session.add(self.user) self.session.commit() def teardown_method(self): self.session.close() def make_account(self, name="acc", **kwargs) -> Account: """建账号：只给**底层判据**，状态一律由生产的推导函数给出。""" account = Account( username=name, password="p", email=f"{name}@e.com", email_password="ep", **kwargs, ) account.status = life.derive_status(account) self.session.add(account) self.session.commit() return account def events(self, account_id: int, event_type: str) -> list[AccountEvent]: return ( self.session.query(AccountEvent) .filter( AccountEvent.account_id == account_id, AccountEvent.event_type == event_type, ) .order_by(AccountEvent.id.asc()) .all() ) class TestDeriveStatus: """真值表：底层判据 → 状态。""" def _derive(self, **kwargs) -> str: # 不落库，纯函数直接算 return life.derive_status(Account(**kwargs)) def test_no_cookie_is_imported(self): assert self._derive(cookie="", ck_check_status="valid") == life.STATUS_IMPORTED def test_no_cookie_wins_over_invalid_ck(self): """CK 被清空后，即使 ck_check_status 还是 invalid，也回到「未登录」。""" assert ( self._derive(cookie="", ck_check_status="invalid", relogin_attempts=3) == life.STATUS_IMPORTED ) @pytest.mark.parametrize("ck", ["", "valid", "error"]) def test_cookie_without_invalid_verdict_is_active(self, ck): assert self._derive(cookie="ck", ck_check_status=ck) == life.STATUS_ACTIVE, ( f"ck_check_status={ck!r} 不该被判成失效" ) @pytest.mark.parametrize("kind", ["captcha", "credential"]) def test_manual_kinds_go_to_manual(self, kind): assert ( self._derive( cookie="ck", ck_check_status="invalid", relogin_failure_kind=kind ) == life.STATUS_MANUAL ) def test_exhausted_retries_go_to_manual(self): """试过、但没有下次排期 = 已放弃自动重登。""" assert ( self._derive( cookie="ck", ck_check_status="invalid", relogin_failure_kind="email_code", relogin_attempts=2, relogin_next_at=None, ) == life.STATUS_MANUAL ) def test_retryable_with_schedule_stays_invalid(self): assert ( self._derive( cookie="ck", ck_check_status="invalid", relogin_failure_kind="email_code", relogin_attempts=1, relogin_next_at=NOW + timedelta(hours=1), ) == life.STATUS_INVALID ) def test_never_attempted_stays_invalid(self): """刚判失效、还没试过 —— 仍是 invalid（会进重登队列）。""" assert ( self._derive(cookie="ck", ck_check_status="invalid") == life.STATUS_INVALID ) @pytest.mark.parametrize("manual_state", list(life.MANUAL_STATES)) def test_manual_states_override_everything(self, manual_state): """人工态优先：账号 CK 好好的，也不该被推导回 active。""" assert ( self._derive( cookie="ck", ck_check_status="valid", status=manual_state, ) == manual_state ) class TestSyncStatus(_Base): def test_writes_and_emits_when_changed(self): account = self.make_account("a", cookie="ck", ck_check_status="valid") assert account.status == life.STATUS_ACTIVE account.ck_check_status = "invalid" changed = life.sync_status(self.session, account, occurred_at=NOW) self.session.commit() assert changed is True assert account.status == life.STATUS_INVALID assert account.status_changed_at == NOW events = self.events(account.id, "status_changed") assert len(events) == 1 assert events[0].category == life.CATEGORY_LIFECYCLE assert events[0].detail["from"] == life.STATUS_ACTIVE assert events[0].detail["to"] == life.STATUS_INVALID def test_is_noop_when_unchanged(self): """幂等：状态没变时不写库、不发事件（否则巡检每轮都刷一条时间线）。""" account = self.make_account("a", cookie="ck", ck_check_status="valid") before_changed_at = account.status_changed_at changed = life.sync_status(self.session, account, occurred_at=NOW) self.session.commit() assert changed is False assert account.status == life.STATUS_ACTIVE assert account.status_changed_at == before_changed_at assert self.events(account.id, "status_changed") == [] def test_never_overrides_manual_state(self): account = self.make_account("a", cookie="ck", ck_check_status="valid") life.set_manual_state(self.session, account, life.STATUS_DISABLED) self.session.commit() assert account.status == life.STATUS_DISABLED # 底层判据怎么变都不该把人工态冲掉 account.ck_check_status = "invalid" changed = life.sync_status(self.session, account) self.session.commit() assert changed is False assert account.status == life.STATUS_DISABLED class TestManualState(_Base): def test_disable_sets_flag_and_event(self): account = self.make_account("a", cookie="ck", ck_check_status="valid") changed = life.set_manual_state( self.session, account, life.STATUS_DISABLED, reason="客户退单", actor_id=self.user.id, occurred_at=NOW, ) self.session.commit() assert changed is True assert account.status == life.STATUS_DISABLED assert account.disabled_at == NOW assert account.status_reason == "客户退单" event = self.events(account.id, "status_changed")[0] assert event.actor_type == "user" assert event.actor_id == self.user.id assert event.detail["reason"] == "客户退单" def test_archive_sets_archived_at(self): account = self.make_account("a", cookie="ck", ck_check_status="valid") life.set_manual_state( self.session, account, life.STATUS_ARCHIVED, occurred_at=NOW ) self.session.commit() assert account.status == life.STATUS_ARCHIVED assert account.archived_at == NOW def test_auto_restores_derived_state(self): """恢复 = 摘掉人工态后立刻按判据重算（账号还有效 → active）。""" account = self.make_account("a", cookie="ck", ck_check_status="valid") life.set_manual_state(self.session, account, life.STATUS_DISABLED) self.session.commit() changed = life.set_manual_state(self.session, account, "auto", occurred_at=NOW) self.session.commit() assert changed is True assert account.status == life.STATUS_ACTIVE assert account.disabled_at is None def test_rejects_unknown_state(self): account = self.make_account("a") with pytest.raises(ValueError): life.set_manual_state(self.session, account, "bogus") class TestFullLifecycle(_Base): """一条走查串起所有阶段 —— 这是 S3 的验收核心。""" def test_walkthrough(self): # ① 导入：没有 CK account = self.make_account("walk") assert account.status == life.STATUS_IMPORTED # ② 登录成功 → 正常 account.cookie = "acf_uid=1" account.ck_check_status = "" life.on_login_success(self.session, account, occurred_at=NOW) self.session.commit() assert account.status == life.STATUS_ACTIVE # ③ 巡检判失效 → CK 失效（会进自动重登） account.ck_check_status = "invalid" life.sync_status(self.session, account, occurred_at=NOW) self.session.commit() assert account.status == life.STATUS_INVALID # ④ 重登成功 → 回到正常 account.cookie = "acf_uid=2" account.ck_check_status = "valid" life.on_login_success(self.session, account, occurred_at=NOW) self.session.commit() assert account.status == life.STATUS_ACTIVE # ⑤ 又失效，且是极验 → 待人工 account.ck_check_status = "invalid" account.relogin_failure_kind = "captcha" account.relogin_attempts = 1 life.sync_status(self.session, account, occurred_at=NOW) self.session.commit() assert account.status == life.STATUS_MANUAL # ⑥ 人工手动登录救回来 → 状态自动恢复（关键：manual 不该卡死） account.cookie = "acf_uid=3" account.ck_check_status = "valid" life.on_login_success(self.session, account, occurred_at=NOW) self.session.commit() assert account.status == life.STATUS_ACTIVE assert account.relogin_failure_kind == "" # ⑦ 清掉 CK → 回到「未登录」 account.cookie = "" life.sync_status(self.session, account, occurred_at=NOW) self.session.commit() assert account.status == life.STATUS_IMPORTED # ⑧ 人工停用 → 之后无论判据怎么变都保持停用 life.set_manual_state(self.session, account, life.STATUS_DISABLED) self.session.commit() account.cookie = "acf_uid=4" account.ck_check_status = "valid" life.sync_status(self.session, account) self.session.commit() assert account.status == life.STATUS_DISABLED # 状态变迁都进了时间线：imported→active→invalid→active→manual→active→imported→disabled events = self.events(account.id, "status_changed") assert [e.detail["to"] for e in events] == [ life.STATUS_ACTIVE, life.STATUS_INVALID, life.STATUS_ACTIVE, life.STATUS_MANUAL, life.STATUS_ACTIVE, life.STATUS_IMPORTED, life.STATUS_DISABLED, ] class TestAutomationSkipsManualStates(_Base): """停用/归档的账号**不参与任何自动化**（S3 之前的真缺口）。""" def _job(self) -> AutomationJob: return AutomationJob( key=ck_health.JOB_KEY, interval_seconds=6 * 3600, batch_size=200 ) def test_ck_health_skips_disabled_and_archived(self): normal = self.make_account("ok", cookie="ck", ck_check_status="valid") disabled = self.make_account("off", cookie="ck", ck_check_status="valid") archived = self.make_account("arch", cookie="ck", ck_check_status="valid") for acc, state in ( (disabled, life.STATUS_DISABLED), (archived, life.STATUS_ARCHIVED), ): life.set_manual_state(self.session, acc, state) self.session.commit() selected = { acc.id for acc in ck_health.build_select_query( self.session, self._job(), NOW ).all() } assert selected == {normal.id} def test_ck_relogin_skips_disabled_and_archived(self): def broken(name): return self.make_account( name, cookie="ck", ck_check_status="invalid", relogin_failure_kind="email_code", relogin_attempts=0, ) eligible = broken("eligible") disabled = broken("disabled") archived = broken("archived") life.set_manual_state(self.session, disabled, life.STATUS_DISABLED) life.set_manual_state(self.session, archived, life.STATUS_ARCHIVED) self.session.commit() selected = { acc.id for acc in ck_relogin.build_select_query( self.session, self._job(), NOW, max_attempts=2 ).all() } assert selected == {eligible.id} def test_pending_backlog_excludes_manual_states(self): def broken(name): return self.make_account( name, cookie="ck", ck_check_status="invalid", relogin_failure_kind="email_code", relogin_attempts=0, ) broken("eligible") disabled = broken("disabled") life.set_manual_state(self.session, disabled, life.STATUS_DISABLED) self.session.commit() assert ck_relogin.count_pending_relogin(self.session, max_attempts=2) == 1 def test_ck_health_run_advances_lifecycle_status(self): """接线点：巡检写 ck_check_status 之后，状态必须跟着变。""" account = self.make_account("a", cookie="ck", ck_check_status="valid") assert account.status == life.STATUS_ACTIVE original = ck_health.check_douyu_cookie ck_health.check_douyu_cookie = lambda _cookie: { "valid": False, "message": "鱼丸接口: 账户登陆异常，请重新登陆", "fish_ball": None, "nickname": None, "level": None, "checked_at": NOW, } try: ck_health.run_ck_health(self.session, self._job()) finally: ck_health.check_douyu_cookie = original self.session.expire_all() fresh = self.session.query(Account).filter(Account.id == account.id).one() assert fresh.ck_check_status == "invalid" assert fresh.status == life.STATUS_INVALID class TestMigrationBackfillMatchesDerive(_Base): """迁移 0042 的状态回填 SQL 必须与 `derive_status()` **同规则**。 这是最容易悄悄分叉的地方：一边是 Python 纯函数，一边是迁移里的 SQL CASE。 分叉了不会报错，只会表现为「迁移刚跑完是 A，下一轮巡检又改成 B」。 这里把迁移里那段 UPDATE 真跑一遍，与纯函数逐条比对。 """ def _load_migration(self): import importlib.util from pathlib import Path path = ( Path(__file__).resolve().parents[1] / "web/backend/migrations/versions/20261003_0042_account_lifecycle.py" ) spec = importlib.util.spec_from_file_location( "mig_0042_account_lifecycle", path ) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def _reset_status_to_imported(self, names: list[str]) -> None: """把落库状态重置成「迁移回填前」的样子，好让回填的 WHERE 命中。""" self.session.query(Account).filter(Account.username.in_(names)).update( {"status": life.STATUS_IMPORTED}, synchronize_session=False ) self.session.commit() def test_sql_backfill_agrees_with_derive_status(self): cases = [ # 覆盖每个状态的入口 ("no_cookie", {"cookie": ""}), ("no_cookie_invalid", {"cookie": "", "ck_check_status": "invalid"}), ("fresh_cookie", {"cookie": "ck"}), ("valid", {"cookie": "ck", "ck_check_status": "valid"}), ("error_verdict", {"cookie": "ck", "ck_check_status": "error"}), ("invalid_never_tried", {"cookie": "ck", "ck_check_status": "invalid"}), ( "invalid_backoff", { "cookie": "ck", "ck_check_status": "invalid", "relogin_failure_kind": "email_code", "relogin_attempts": 1, "relogin_next_at": NOW + timedelta(hours=1), }, ), ( "captcha", { "cookie": "ck", "ck_check_status": "invalid", "relogin_failure_kind": "captcha", "relogin_attempts": 1, }, ), ( "credential", { "cookie": "ck", "ck_check_status": "invalid", "relogin_failure_kind": "credential", "relogin_attempts": 1, }, ), ( "exhausted", { "cookie": "ck", "ck_check_status": "invalid", "relogin_failure_kind": "email_code", "relogin_attempts": 2, "relogin_next_at": None, }, ), ] expected: dict[str, str] = {} for name, kwargs in cases: expected[name] = life.derive_status(self.make_account(name, **kwargs)) names = [name for name, _ in cases] self._reset_status_to_imported(names) migration = self._load_migration() with self.engine.begin() as conn: migration._backfill_lifecycle_status(conn) self.session.expire_all() for name in names: actual = ( self.session.query(Account) .filter(Account.username == name) .one() .status ) assert actual == expected[name], ( f"{name}: 迁移回填 {actual} 与 derive_status {expected[name]} 不一致" ) def test_backfill_never_writes_removed_relogin_state(self): """`relogin` 状态已在 S3 砍掉，迁移不该再写出它。""" self.make_account( "backoff", cookie="ck", ck_check_status="invalid", relogin_failure_kind="email_code", relogin_attempts=1, relogin_next_at=NOW + timedelta(hours=1), ) self._reset_status_to_imported(["backoff"]) migration = self._load_migration() with self.engine.begin() as conn: migration._backfill_lifecycle_status(conn) self.session.expire_all() status = ( self.session.query(Account) .filter(Account.username == "backoff") .one() .status ) assert status in life.ALL_STATES assert status == life.STATUS_INVALID class TestListStatusFilter(_Base): """列表按状态筛选；归档默认不出现。""" def _names(self, **kwargs) -> list[str]: from web.backend.routers.accounts import list_accounts result = list_accounts( assigned_only=False, tag="", has_cookie=False, search="", page=1, page_size=50, include_sensitive=False, status=kwargs.get("status", ""), db=self.session, current=self.user, ) return [item.username for item in result["items"]] def _seed(self): self.make_account("act", cookie="ck", ck_check_status="valid") self.make_account("imp") off = self.make_account("off", cookie="ck", ck_check_status="valid") arch = self.make_account("arch", cookie="ck", ck_check_status="valid") life.set_manual_state(self.session, off, life.STATUS_DISABLED) life.set_manual_state(self.session, arch, life.STATUS_ARCHIVED) self.session.commit() def test_default_hides_archived(self): self._seed() names = self._names() assert "arch" not in names assert {"act", "imp", "off"} <= set(names) def test_filter_by_status(self): self._seed() assert self._names(status=life.STATUS_MANUAL) == [] assert self._names(status=life.STATUS_ARCHIVED) == ["arch"] # disabled 是独立状态，不会出现在 active 里 assert set(self._names(status=life.STATUS_ACTIVE)) == {"act"} assert set(self._names(status=life.STATUS_DISABLED)) == {"off"} def test_repeated_same_target_transitions_all_land_in_timeline(self): """回归：同一秒内两次切到同一状态，两条事件都要落库。 曾经的幂等键 `status:{id}:{target}:{秒级时间戳}` 会在这时撞唯一键， `emit` 静默吞掉 → **状态改了、时间线少一条**。 """ account = self.make_account("a", cookie="ck", ck_check_status="valid") account.ck_check_status = "invalid" life.sync_status(self.session, account, occurred_at=NOW) self.session.commit() account.ck_check_status = "valid" life.sync_status(self.session, account, occurred_at=NOW) self.session.commit() account.ck_check_status = "invalid" life.sync_status(self.session, account, occurred_at=NOW) self.session.commit() events = self.events(account.id, "status_changed") assert [e.detail["to"] for e in events] == [ life.STATUS_INVALID, life.STATUS_ACTIVE, life.STATUS_INVALID, ] def test_list_item_carries_status_fields(self): self.make_account("act", cookie="ck", ck_check_status="valid") from web.backend.routers.accounts import list_accounts result = list_accounts( assigned_only=False, tag="", has_cookie=False, search="", page=1, page_size=50, include_sensitive=False, status="", db=self.session, current=self.user, ) item = result["items"][0] assert item.status == life.STATUS_ACTIVE assert item.status_label == "正常" class TestLifecycleApi(_Base): def test_disable_then_restore(self): from web.backend.routers.accounts import ( set_account_automation, set_account_lifecycle, ) from web.backend.schemas import ( AccountAutomationUpdate, AccountLifecycleUpdate, ) account = self.make_account("api", cookie="ck", ck_check_status="valid") res = set_account_lifecycle( account.id, AccountLifecycleUpdate(status=life.STATUS_DISABLED, reason="退单"), db=self.session, current=self.user, ) self.session.refresh(account) assert res["success"] and account.status == life.STATUS_DISABLED res = set_account_lifecycle( account.id, AccountLifecycleUpdate(status="auto"), db=self.session, current=self.user, ) self.session.refresh(account) assert res["success"] and account.status == life.STATUS_ACTIVE res = set_account_automation( account.id, AccountAutomationUpdate(auto_managed=False), db=self.session, current=self.user, ) self.session.refresh(account) assert res["success"] and account.auto_managed is False # 关掉自动化不等于停用：账号状态仍可用 assert account.status == life.STATUS_ACTIVE class TestStatusWriteFunnel: """架构不变量：**`account.status` 只能由 `account_lifecycle` 写**。 `accounts.bind_status` 就是这么烂掉的 —— 谁都能写、语义没人管，最后 36 种取值 互相覆盖，连「这账号到底绑没绑」都查不出来。 这条用源码扫描把它钉死：任何地方直接给 `status` 赋值都会让用例变红，逼你去用 `sync_status()`（按判据推导）或 `set_manual_state()`（人工态）。 比写在文档里的约定可靠 —— 约定会被忘记，红了的用例不会。 """ #: 允许直接写 status 的地方（模型定义本体、收口模块、以及迁移脚本的历史回填） ALLOWED: ClassVar[frozenset[str]] = frozenset( { "web/backend/models.py", "web/backend/services/account_lifecycle.py", } ) def test_only_lifecycle_module_writes_account_status(self): import re from pathlib import Path root = Path(__file__).resolve().parents[1] backend = root / "web" / "backend" patterns = ( re.compile(r"\b(?:account|acc)\.status\s*=(?!=)"), # 批量更新：update({Account.status: ...}) re.compile(r"\{\s*Account\.status\s*:"), # bulk update 的多行写法 re.compile(r"^\s*\{\s*Account\.status\s*,"), ) offenders: list[str] = [] for path in backend.rglob("*.py"): rel = path.relative_to(root).as_posix() # 虎牙是**另一套账号模型**（`huya_accounts.status` 自由文本， # 20+ 种取值当状态机用，与这里的 bind_status 是同一个病）。 # 那一摊单独治理，不在本不变量范围内。 if rel in self.ALLOWED or "/migrations/" in rel or "huya" in rel: continue for lineno, line in enumerate( path.read_text(encoding="utf-8").splitlines(), 1 ): if line.lstrip().startswith("#"): continue if any(p.search(line) for p in patterns): offenders.append(f"{rel}:{lineno}: {line.strip()}") assert not offenders, ( "account.status 只能由 account_lifecycle 写（bind_status 的教训）：\n" "请改用 sync_status() / set_manual_state()。违规点：\n" + "\n".join(offenders) ) class TestListViewKeyColumns(_Base): """账号列表的「关键列」：客服要一眼看出「能不能用、要不要我处理」。 这些字段全部直接来自 `accounts` 行 —— 不能为了列表去 JOIN 或聚合 （列表是全站最热的接口之一）。 """ def _items(self, **kwargs): from web.backend.routers.accounts import list_accounts return list_accounts( assigned_only=False, tag="", has_cookie=False, search="", page=1, page_size=50, include_sensitive=False, status=kwargs.get("status", ""), ck_status=kwargs.get("ck_status", ""), bind_status=kwargs.get("bind_status", ""), db=self.session, current=self.user, )["items"] def _item(self, username: str, **kwargs): """按用户名取自己那条 —— 库里会累积前面用例建的账号，不能假设顺序。""" for item in self._items(**kwargs): if item.username == username: return item raise AssertionError(f"列表里没有 {username}") def _item_by_id(self, account, **kwargs): """按 **id** 取刚建的那条。 ⚠️ 按用户名取是不可靠的：`make_account("key")` 每次都建**新**账号， 测试库却一直在累积，于是"用户名 == key"可能命中**很早以前**建的那个 （它的绑定状态当然是 unknown）。这个用例就因此假红过一次。 """ for item in self._items(**kwargs): if item.id == account.id: return item raise AssertionError(f"列表里没有 id={account.id}") def test_item_carries_key_columns(self): from web.backend.services.account_binding import ( BIND_BOUND, SCOPE_ELITE, set_bind_status, ) account = self.make_account("key", cookie="ck", ck_check_status="valid") account.points = 1234 account.gold_balance = 56 set_bind_status(account, SCOPE_ELITE, BIND_BOUND) self.session.commit() item = self._item_by_id(account) assert item.has_cookie is True assert item.ck_check_status == "valid" assert item.points == 1234 assert item.gold_balance == 56 assert item.role_bind_status == BIND_BOUND assert item.role_bind_label == "已绑定" # 另外两个作用域没绑，但字段必须在（前端要显示「?」而不是崩） assert item.esports_bind_status == "unknown" assert item.xpd_bind_label == "未知" def test_no_cookie_is_reported(self): self.make_account("nocookie", cookie="") assert self._item("nocookie").has_cookie is False def test_filter_by_ck_status(self): self.make_account("bad", cookie="ck", ck_check_status="invalid") self.make_account("good", cookie="ck", ck_check_status="valid") names = [i.username for i in self._items(ck_status="invalid")] assert "bad" in names assert "good" not in names def test_filter_by_bind_status_matches_any_scope(self): from web.backend.services.account_binding import ( BIND_BOUND, BIND_UNBOUND, SCOPE_ELITE, SCOPE_PEACE, set_bind_status, ) elite_bound = self.make_account("elite-bound") set_bind_status(elite_bound, SCOPE_ELITE, BIND_BOUND) peace_unbound = self.make_account("peace-unbound") set_bind_status(peace_unbound, SCOPE_PEACE, BIND_UNBOUND) self.make_account("all-unknown") self.session.commit() bound = {i.username for i in self._items(bind_status=BIND_BOUND)} assert "elite-bound" in bound assert "peace-unbound" not in bound unbound = {i.username for i in self._items(bind_status=BIND_UNBOUND)} assert "peace-unbound" in unbound assert "elite-bound" not in unbound # unknown 会把没绑过的也算进来（默认值就是 unknown） unknown = {i.username for i in self._items(bind_status="unknown")} assert "all-unknown" in unknown class TestSummaryByStatus(_Base): """账号列表顶部的卡片是**快捷筛选入口**，不是统计装饰。 所以它必须按**生命周期状态**分布 —— 「有几个要处理」才可行动， 原来的「标签数 / 已分配 / 未分配」看了也不知道该干什么。 """ def _summary(self): from web.backend.routers.accounts import accounts_summary return accounts_summary(db=self.session, current=self.user) def test_distribution_counts_each_status(self): self.make_account("act", cookie="ck", ck_check_status="valid") self.make_account("imp") invalid = self.make_account("bad", cookie="ck", ck_check_status="invalid") off = self.make_account("off", cookie="ck", ck_check_status="valid") life.set_manual_state(self.session, off, life.STATUS_DISABLED) self.session.commit() summary = self._summary() by = summary["by_status"] # 库里可能有其它用例留下的账号，所以按「至少」断言 assert by.get("active", 0) >= 1 assert by.get("imported", 0) >= 1 assert by.get("invalid", 0) >= 1 assert by.get("disabled", 0) >= 1 assert invalid.id # 用掉变量 def test_archived_excluded_from_distribution(self): """归档默认不出现在列表里，卡片口径必须与列表一致。""" arch = self.make_account("arch", cookie="ck", ck_check_status="valid") life.set_manual_state(self.session, arch, life.STATUS_ARCHIVED) self.session.commit() summary = self._summary() assert "archived" not in summary["by_status"] assert summary["visible_total"] == sum(summary["by_status"].values()) def test_ck_invalid_and_no_cookie_counts(self): self.make_account("bad", cookie="ck", ck_check_status="invalid") self.make_account("nocookie") summary = self._summary() assert summary["ck_invalid_count"] >= 1 assert summary["no_cookie_count"] >= 1 class TestExchangeRecordColumn(_Base): """「兑换记录」列：次数 + 最近一次（商品名 + 时间）。 数据源是**账号自己的事件流**（`account_events.exchange`），不是 `douyu_tasks` 的 JSON —— 后者是执行日志，S5 起业务事实已经在事件里了。 """ def _items(self, **kwargs): from web.backend.routers.accounts import list_accounts return list_accounts( assigned_only=False, tag="", has_cookie=False, search="", page=1, page_size=50, include_sensitive=False, status=kwargs.get("status", ""), ck_status="", bind_status="", db=self.session, current=self.user, )["items"] def _item(self, username: str): for item in self._items(): if item.username == username: return item raise AssertionError(f"列表里没有 {username}") def _exchange(self, account_id: int, name: str, *, days_ago: int = 0): from datetime import timedelta from web.backend.services.account_events_service import ( OFFICIAL_SYNC_SOURCE, emit, ) emit( self.session, account_id=account_id, event_type="exchange", title=f"兑换 {name}", category="business", subject_key=name, occurred_at=NOW - timedelta(days=days_ago), # 列表的「兑换记录」列**只聚合斗鱼官方记录**（source=official_sync）， # 我们当场记的那些留给详情页时间线 —— 这是刻意的口径划分。 source=OFFICIAL_SYNC_SOURCE, ) self.session.commit() def test_no_exchange_yields_zero(self): self.make_account("noex") item = self._item("noex") assert item.exchange_count == 0 assert item.last_exchange_at is None assert item.last_exchange_name == "" def test_counts_and_keeps_latest_name(self): account = self.make_account("hasex") self._exchange(account.id, "兔兔白日梦", days_ago=3) self._exchange(account.id, "冠军鸡腿", days_ago=1) self._exchange(account.id, "烟花", days_ago=2) item = self._item("hasex") assert item.exchange_count == 3 # 「最近一次」按**事件时间**取：三条分别是 3 天前 / 1 天前 / 2 天前， # 所以最近的是「冠军鸡腿」（1 天前）。 # # ⚠️ 这里原来断言的是「烟花」（**最后插入**的那条）—— 那是错的， # 因为当时后端按事件 **id** 排序，于是「最近一次」实际是"最后插入的一次"。 # 一键查询把斗鱼官方的**历史**兑换补成事件后这个 bug 才暴露： # 补录的事件 id 最大、时间却最老，会把几个月前的老兑换显示成"最近一次"。 # 现在排序是 `occurred_at DESC, id DESC`（时间相同才用 id 兜底）。 # 后端返回事件原始 title；剥掉「兑换 」前缀是前端展示层的事。 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): """别的类型（充值 / 登录 / 状态）不能混进兑换次数。""" from web.backend.services.account_events_service import emit account = self.make_account("mixed") emit( self.session, account_id=account.id, event_type="recharge", title="充值 100", category="business", amount=100, ) emit( self.session, account_id=account.id, event_type="status_changed", title="状态：未登录 → 正常", category="lifecycle", ) self.session.commit() assert self._item("mixed").exchange_count == 0