yml2213
885a780bd3
S6 的展示层部分（数据模型部分见上一个提交）。目标：客服打开账号页要能一眼回答
「这账号能不能用、要不要我处理、账上还有多少」。
## 账号列表：加关键列 + 三个筛选
新增四列，**全部直接来自 `accounts` 行**（不 JOIN、不聚合 —— 列表是全站最热接口）：
| 列 | 内容 |
| --- | --- |
| **CK** | 有无 CK / 正常 / 已失效 / 检测异常 + 最近检测时间 |
| **绑定** | 精/电/店 三个作用域各一个小标签（✓ / … / ✗ / ?） |
| **积分 / 鱼翅** | 两行小字 |
新增筛选：**CK 健康**、**角色绑定**（三个作用域任一命中即算）。
`AccountOut` 相应补齐 `has_cookie` / `ck_check_status` / `ck_checked_at` /
`points` / `gold_balance` / 三组绑定状态与标签。
## 账号详情页：生命周期主页
顶部新增**生命周期状态卡**，把之前散落的信息聚成一眼可读的几格：
- **生命周期状态**：状态标签 + 不参与自动化标记 + 变更时间与原因
- **CK**：有无 / 健康 / 最近检测时间
- **角色绑定**：精/电/店三格
- **账上还有多少**：积分 / 鱼翅 / 兑换余额
- **归属与标注**：分配客服（下拉）+ 标签（就地编辑）—— 之前详情页只能看不能改
业务台账补上**登录成功次数**与**最近登录**（`stats` 里本来就有，只是没展示）。
时间线已有分类筛选，生命周期操作入口（停用/归档/恢复/自动化）S3 已加。
## ⚠️ 顺带修掉一个静默 bug
`bind_payload()` 的字段名按**作用域名**拼成了 `elite_bind_status` / `peace_bind_status`，
而 schema 声明的是 `role_bind_status` / `xpd_bind_status`。Pydantic 默认
`extra='ignore'`，于是多出来的键**被静默丢弃**、声明过的字段保持默认值 ——
**接口永远返回 unknown 而不报任何错**（这正是它躲过前面几轮测试的原因）。
修法：前缀改为由**列名**推导（`SCOPE_PAYLOAD_PREFIX`），并在模块导入时自检
「载荷前缀 == 列名前缀」。另加两条用例：`bind_payload` 的键必须都被
`DouyuTaskAccountOut` / `AccountOut` 声明（这是能抓到这类错的那种用例）。
顺带发现我自己之前那条 payload 用例**把错误的前缀当成了预期**，一并修正。
## 自己踩的坑
详情页加客服列表时又把 `useEffect` 写在了早期 `return` 之后 ——
和本次会话前面修 `LoginTasksPage` 的是同一个错。eslint（`react-hooks/rules-of-hooks`）
当场抓住，已上移。**`tsc` 与 `vite build` 都不会报这个**。
另外：列表关键列那处编辑我曾以为已生效（脚本打印了成功），实际没落到文件里 ——
后面是靠「读回文件确认」和用例失败才发现。已改为每次都读回核对。
## 验证
- `TZ=Asia/Shanghai`：**497 passed / 6 skipped**（S6 前半 490 → +7）
- `TZ=UTC`（相关文件）通过；ruff 全绿；前端 build 通过、lint **0 errors**
- **真实开发库端到端**：列表与工作台接口都返回了真实值
（`has_cookie=True` / `ck_check_status=valid` / `role_bind_label=未知`），
修 bug 前这两处返回的是被丢弃后的默认值
未做：`bind_status` 旧列待 S8 drop；保留策略里「状态派生类查询」已可下探到 7 天档（待确认）。
810 lines
30 KiB
Python
810 lines
30 KiB
Python
"""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 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("key")
|
||
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
|