yml2213
eec42aa8a7
按反馈是**增加**而不是**替换**：旧路径一行未动，新路径另起一条。 ## 两套来源并存 | | 手动输入（旧，默认） | 选择账号（新） | | --- | --- | --- | | 来源 | 粘贴文本 | 库里账号的凭据（不用人工粘贴） | | 条目 `account_id` | **无** | 有 | | 结果落账号时间线 | **否**（保持旧行为） | 是，新事件类型 `account_check` | | 批次 / 并发 / 重试 / 代理 / 导出 zip | 完全不变 | 完全不变 | **旧路径的「不发事件」是刻意的**：粘贴进来的账号没有归属，凭空造一条没有账号的 事件只是污染时间线。`parse_account_check_lines`、`export_line`、zip 分组全部未动， 有用例专门钉住这一点。 ## 「选择账号」支持筛选（按反馈补充） 不是给一个扁平下拉，而是先按条件缩小范围再决定选法： - 筛选：**标签 / 生命周期状态 / CK 健康**（复用账号列表同一套 `_filter_accounts_query`， 两处口径不会不一致） - 选法：手动多选（选项随筛选刷新）**或**「按筛选全选」（含列表里没显示出来的） 服务端支持 `select_all_matching + 各筛选条件`，与账号列表批量操作的字段形状一致。 可见性沿用账号域口径：客服只能检测自己名下的账号（越权 id 直接 403）。 ## 结果表加账号回链 「选择账号」模式下条目带 `account_id`，账号列变成可点链接 → 直接跳到账号详情， 结果也在那里的时间线上。 ## 排查中修掉两个 bug（都不是测试发现的） **① `registry.create` 漏传 `account_id`** 它构造 `AccountCheckItemState` 时没把 `account_id` 从输入复制过去。 后果：批次照跑、结果照显示、zip 照导出，**事件永远不发且没有任何报错**。 是「跑一次真实检测再查事件表」才发现的（查出来 0 条）。已补，并加了用例。 **② 终态判断写成「键存在」而不是「非空」** `_run_one` 置 `running` 时会显式传 `finished_at=None`，而判断写的是 `"finished_at" in updates` → **中间态也发事件**，于是时间线上记的是 `账号检查：running`；真正的终态再发时撞同一个幂等键被**静默丢弃** —— 结果是「只有一条，但记错了」，比没有更糟。 改成判非空。此处暴露了一个**假绿的用例**：我原来的「中间态不发事件」用例 没传 `finished_at=None`（与真实调用不一致），所以放过了这个 bug； 而且它还缺 `session.commit()`，在 MySQL REPEATABLE READ 下根本看不到 另一个会话刚提交的行 —— 双重假绿。两处都修了，并做了反向验证 （改回旧判断 → 用例立刻红）。 ## 验证 - `TZ=Asia/Shanghai`：**520 passed / 6 skipped**（新增 17 条检测模式用例） - `TZ=UTC`（检测 + 迁移）20 passed；ruff 全绿；前端 build 通过、lint 0 errors - **真实端到端**：用真实库账号起批次 → 检测跑完（斗鱼返回「账号已认证」）→ 事件确实落库；清掉脏数据后用**与 `_run_one` 完全相同的调用序列**复验： 两次 `_set_item` 只产生一条事件，且 title 是 `账号检测：账号已认证`、`status=account_verified`（不再记成 running） 注：验证时真实触发了几次斗鱼登录检测（开发库那两个账号），结果都是「已认证」。
333 lines
12 KiB
Python
333 lines
12 KiB
Python
"""S7：账号检测保留旧的手动输入，新增「按账号选择」。
|
||
|
||
本文件守住的核心是**两套来源并存**，而不是替换：
|
||
|
||
| | 手动输入（旧） | 选择账号（新） |
|
||
| --- | --- | --- |
|
||
| 来源 | 粘贴文本 | 库里账号的凭据 |
|
||
| `account_id` | **无** | 有 |
|
||
| 结果落账号时间线 | **否**（保持旧行为） | 是 |
|
||
|
||
最容易做错的两件事：① 把旧路径一起改成发事件（那会凭空造出没有归属的事件）；
|
||
② 新路径忘了按可见性过滤（客服就能检测别人名下的账号）。
|
||
"""
|
||
|
||
from datetime import UTC, datetime
|
||
|
||
import pytest
|
||
|
||
from tests.support.db import get_test_engine, make_test_session
|
||
from web.backend.models import Account, AccountEvent, User
|
||
from web.backend.services import account_check_service as svc
|
||
|
||
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.session.query(AccountEvent).delete()
|
||
self.session.commit()
|
||
self.user = User(username="chk-admin", password_hash="h", role="super_admin")
|
||
self.session.add(self.user)
|
||
self.session.commit()
|
||
|
||
def teardown_method(self):
|
||
self.session.close()
|
||
|
||
def account(self, name: str, **kwargs) -> Account:
|
||
acc = Account(
|
||
username=name,
|
||
password=f"pw-{name}",
|
||
email=f"{name}@e.com",
|
||
email_password=f"ep-{name}",
|
||
**kwargs,
|
||
)
|
||
self.session.add(acc)
|
||
self.session.commit()
|
||
return acc
|
||
|
||
|
||
class TestLegacyTextPathUnchanged(_Base):
|
||
"""**旧逻辑一字未动** —— 这是本次明确要求保留的部分。"""
|
||
|
||
def test_parses_without_account_id(self):
|
||
inputs = svc.parse_account_check_lines("u1----p1----e1----ep1")
|
||
assert len(inputs) == 1
|
||
assert inputs[0].username == "u1"
|
||
assert inputs[0].password == "p1"
|
||
# 关键：没有归属 → 后续不会往任何账号写事件
|
||
assert inputs[0].account_id is None
|
||
|
||
def test_parse_supports_both_separators(self):
|
||
inputs = svc.parse_account_check_lines("u1----p1----e1----ep1\nu2|p2|e2|ep2")
|
||
assert [i.username for i in inputs] == ["u1", "u2"]
|
||
|
||
def test_export_line_unchanged(self):
|
||
inputs = svc.parse_account_check_lines("u1----p1----e1----ep1")
|
||
assert inputs[0].export_line() == "u1----p1----e1----ep1"
|
||
|
||
|
||
class TestSelectAccountsPath(_Base):
|
||
def test_builds_inputs_with_account_id(self):
|
||
a = self.account("sa-1")
|
||
b = self.account("sa-2")
|
||
rows = self.session.query(Account).filter(Account.id.in_([a.id, b.id])).all()
|
||
|
||
inputs = svc.build_inputs_from_accounts(rows, [a.id, b.id])
|
||
assert [i.account_id for i in inputs] == [a.id, b.id]
|
||
# 凭据从库里取，不用人工粘贴
|
||
assert inputs[0].username == "sa-1"
|
||
assert inputs[0].password == "pw-sa-1"
|
||
assert inputs[0].email == "sa-1@e.com"
|
||
|
||
def test_order_follows_requested_ids(self):
|
||
a = self.account("ord-1")
|
||
b = self.account("ord-2")
|
||
rows = self.session.query(Account).filter(Account.id.in_([a.id, b.id])).all()
|
||
inputs = svc.build_inputs_from_accounts(rows, [b.id, a.id])
|
||
assert [i.account_id for i in inputs] == [b.id, a.id]
|
||
|
||
def test_unknown_id_is_skipped_not_crashing(self):
|
||
a = self.account("sk-1")
|
||
rows = self.session.query(Account).filter(Account.id == a.id).all()
|
||
inputs = svc.build_inputs_from_accounts(rows, [a.id, 10_000_000])
|
||
assert [i.account_id for i in inputs] == [a.id]
|
||
|
||
|
||
class TestResultGoesToAccountTimeline(_Base):
|
||
"""只有「按账号选择」的条目才写事件。"""
|
||
|
||
def _runner(self):
|
||
runner = svc.AccountCheckRunner.__new__(svc.AccountCheckRunner)
|
||
runner.batch = svc.AccountCheckBatch(
|
||
batch_id="batch-x",
|
||
created_by=self.user.id,
|
||
concurrency=1,
|
||
max_login_retries=0,
|
||
max_total_time=60,
|
||
use_proxy=False,
|
||
items=[],
|
||
)
|
||
import threading
|
||
|
||
runner._lock = threading.Lock()
|
||
runner._stop = threading.Event()
|
||
return runner
|
||
|
||
def test_account_linked_item_emits_event(self):
|
||
a = self.account("evt-1")
|
||
runner = self._runner()
|
||
runner.batch.items = [
|
||
svc.AccountCheckItemState(
|
||
line=1,
|
||
username="evt-1",
|
||
email="evt-1@e.com",
|
||
export_text="evt-1----pw----e----ep",
|
||
account_id=a.id,
|
||
)
|
||
]
|
||
# 终态（带 finished_at）→ 写事件
|
||
runner._set_item(
|
||
0,
|
||
status="account_verified",
|
||
message="账号已认证",
|
||
finished_at=NOW,
|
||
)
|
||
# 事件由 runner 自己开的会话提交；本测试会话在 REPEATABLE READ 下
|
||
# 沿用旧快照看不到它 —— 先结束自己的事务再查。
|
||
self.session.commit()
|
||
|
||
events = (
|
||
self.session.query(AccountEvent)
|
||
.filter(
|
||
AccountEvent.account_id == a.id,
|
||
AccountEvent.event_type == svc.EVENT_ACCOUNT_CHECK,
|
||
)
|
||
.all()
|
||
)
|
||
assert len(events) == 1
|
||
assert events[0].detail["status"] == "account_verified"
|
||
assert events[0].detail["batch_id"] == "batch-x"
|
||
|
||
def test_text_item_emits_nothing(self):
|
||
"""手动粘贴的条目没有归属 —— 绝不能凭空造出没有账号的事件。"""
|
||
self.account("evt-2")
|
||
runner = self._runner()
|
||
runner.batch.items = [
|
||
svc.AccountCheckItemState(
|
||
line=1,
|
||
username="pasted",
|
||
email="pasted@e.com",
|
||
export_text="pasted----pw----e----ep",
|
||
account_id=None,
|
||
)
|
||
]
|
||
runner._set_item(0, status="account_verified", message="ok", finished_at=NOW)
|
||
|
||
assert self.session.query(AccountEvent).count() == 0
|
||
|
||
def test_non_terminal_update_emits_nothing(self):
|
||
"""中间态（running 等，不带 finished_at）不写事件，否则时间线会被刷屏。"""
|
||
a = self.account("evt-3")
|
||
runner = self._runner()
|
||
runner.batch.items = [
|
||
svc.AccountCheckItemState(
|
||
line=1,
|
||
username="evt-3",
|
||
email="e@e.com",
|
||
export_text="x",
|
||
account_id=a.id,
|
||
)
|
||
]
|
||
# ⚠️ 必须**照真实调用**传 `finished_at=None`：`_run_one` 就是这么传的。
|
||
# 第一版用例漏了它，于是「中间态也发事件」这个 bug 逃过了用例，
|
||
# 最后是靠跑一次真实检测、看到事件 title 是「running」才发现的。
|
||
runner._set_item(
|
||
0, status="running", message="检测中", started_at=NOW, finished_at=None
|
||
)
|
||
# 事件由 runner 自己的会话提交；本会话要结束事务才能看到新快照，
|
||
# 否则「没查到」可能只是快照旧了 —— 那会让这条断言假绿。
|
||
self.session.commit()
|
||
assert self.session.query(AccountEvent).count() == 0
|
||
|
||
def test_same_batch_does_not_duplicate(self):
|
||
"""同一次检测重试/重放不会写两条（幂等键带 account_id 与 batch_id）。"""
|
||
a = self.account("evt-4")
|
||
runner = self._runner()
|
||
runner.batch.items = [
|
||
svc.AccountCheckItemState(
|
||
line=1,
|
||
username="evt-4",
|
||
email="e@e.com",
|
||
export_text="x",
|
||
account_id=a.id,
|
||
)
|
||
]
|
||
runner._set_item(0, status="account_verified", message="ok", finished_at=NOW)
|
||
runner._set_item(0, status="account_verified", message="ok", finished_at=NOW)
|
||
self.session.commit()
|
||
|
||
assert (
|
||
self.session.query(AccountEvent)
|
||
.filter(AccountEvent.event_type == svc.EVENT_ACCOUNT_CHECK)
|
||
.count()
|
||
== 1
|
||
)
|
||
|
||
|
||
class TestRequestValidation:
|
||
"""两种来源二选一：都给或都不给都该被拒。"""
|
||
|
||
@staticmethod
|
||
def _call(**kwargs):
|
||
from fastapi import HTTPException
|
||
|
||
from web.backend.routers.account_check import create_account_check_batch
|
||
from web.backend.schemas import AccountCheckBatchRequest
|
||
|
||
return (
|
||
create_account_check_batch,
|
||
AccountCheckBatchRequest,
|
||
HTTPException,
|
||
kwargs,
|
||
)
|
||
|
||
def test_both_sources_rejected(self):
|
||
create, Request, HTTPException, _ = self._call()
|
||
with pytest.raises(HTTPException) as exc:
|
||
create(
|
||
Request(text="u----p----e----ep", account_ids=[1]),
|
||
db=None,
|
||
current=User(username="x", password_hash="h", role="super_admin"),
|
||
)
|
||
assert exc.value.status_code == 400
|
||
assert "二选一" in str(exc.value.detail)
|
||
|
||
def test_neither_source_rejected(self):
|
||
create, Request, HTTPException, _ = self._call()
|
||
with pytest.raises(HTTPException) as exc:
|
||
create(
|
||
Request(text=" ", account_ids=[]),
|
||
db=None,
|
||
current=User(username="x", password_hash="h", role="super_admin"),
|
||
)
|
||
assert exc.value.status_code == 400
|
||
|
||
|
||
class TestFilterBasedSelection(_Base):
|
||
"""「按筛选全选」：与账号列表同一套条件，列表筛出什么就检测什么。"""
|
||
|
||
def _resolve(self, **kwargs):
|
||
"""复现路由里的解析逻辑（用同一套 _filter_accounts_query）。"""
|
||
from web.backend.routers.accounts import (
|
||
_filter_accounts_query,
|
||
_visible_accounts_query,
|
||
)
|
||
|
||
query = _filter_accounts_query(
|
||
self.session,
|
||
_visible_accounts_query(self.session, self.user),
|
||
self.user,
|
||
search=kwargs.get("search", ""),
|
||
tag=kwargs.get("tag", ""),
|
||
status=kwargs.get("status", ""),
|
||
ck_status=kwargs.get("ck_status", ""),
|
||
bind_status="",
|
||
)
|
||
return [a.username for a in query.order_by(Account.id.asc()).all()]
|
||
|
||
def test_filters_by_tag(self):
|
||
self.account("tag-a", tag="青龙")
|
||
self.account("tag-b", tag="白虎")
|
||
assert self._resolve(tag="青龙") == ["tag-a"]
|
||
|
||
def test_filters_by_status_and_ck(self):
|
||
self.account("st-imp")
|
||
self.account("st-bad", cookie="ck", ck_check_status="invalid")
|
||
assert self._resolve(ck_status="invalid") == ["st-bad"]
|
||
assert "st-imp" in self._resolve(status="imported")
|
||
|
||
def test_no_filter_returns_all_visible(self):
|
||
self.account("all-1")
|
||
self.account("all-2")
|
||
assert {"all-1", "all-2"} <= set(self._resolve())
|
||
|
||
|
||
class TestInputToItemCarriesAccountId(_Base):
|
||
"""`registry.create` 必须把 `account_id` 从输入复制到条目。
|
||
|
||
漏了这一行时**没有任何报错** —— 批次照跑、结果照显示、zip 照导出，
|
||
只是事件永远不发。第一版就漏了，靠「跑一次真实检测再查事件」才发现。
|
||
"""
|
||
|
||
def test_item_keeps_account_id(self):
|
||
from web.backend.services.account_check_service import AccountCheckRegistry
|
||
|
||
a = self.account("reg-1")
|
||
rows = self.session.query(Account).filter(Account.id == a.id).all()
|
||
inputs = svc.build_inputs_from_accounts(rows, [a.id])
|
||
|
||
runner = AccountCheckRegistry().create(
|
||
accounts=inputs,
|
||
created_by=self.user.id,
|
||
concurrency=1,
|
||
max_login_retries=0,
|
||
max_total_time=60,
|
||
)
|
||
assert [i.account_id for i in runner.batch.items] == [a.id]
|
||
|
||
def test_text_mode_items_have_no_account_id(self):
|
||
from web.backend.services.account_check_service import AccountCheckRegistry
|
||
|
||
inputs = svc.parse_account_check_lines("u----p----e----ep")
|
||
runner = AccountCheckRegistry().create(
|
||
accounts=inputs,
|
||
created_by=self.user.id,
|
||
concurrency=1,
|
||
max_login_retries=0,
|
||
max_total_time=60,
|
||
)
|
||
assert [i.account_id for i in runner.batch.items] == [None]
|