yml2213
8310a9115e
事故：斗鱼账号的 CK **只存在 `login_tasks.cookie`**（`accounts` 表没有 cookie 列），删掉记录即永久丢失登录态。而列表显示的「状态」是最近一次尝试 （`attempt_status`），于是一条**显示「失败」但实际仍持有有效 CK** 的记录 （status=success + cookie + 上次尝试失败）会被「清空失败」当成垃圾删掉。 修复： - 批量删除只删空 cookie 的记录，持有 CK 的一律跳过，并在响应里报告跳过数； - 单条删除遇到持有 CK 的记录直接 409，提示先处理该账号的 CK； - 前端确认弹窗写明「持有 CK 的记录会跳过」。 顺带修统计口径：分页后 success/failed 只统计当前页，却和「共 N 个任务」 并排显示（曾出现「共 8508，成功 48」）。改走 /tasks/summary 的全局统计。 回归测试补 5 条，专门盯「显示失败但持有 CK」这个事故场景。
383 lines
13 KiB
Python
383 lines
13 KiB
Python
"""回归：登录任务列表分页、跨页全选删除、以及「不删在途任务」的护栏。
|
||
|
||
任务量早就过了单页能装下的规模（库里 1.4 万条、单批就 177 条），前端只能看
|
||
200 条于是「全选」其实选不到全部。分页 + ``all_matching`` 是为了让「清空失败
|
||
任务」这类操作真能一键做完。
|
||
|
||
三条必须守住：
|
||
|
||
1. **不传 page 时返回裸数组** —— Cookie 页等老调用方依赖这个契约，改了就崩；
|
||
2. **传 page 才返回 {items,total}**，total 用于跨页全选；
|
||
3. **pending/running 一律不删** —— 删掉会让批次线程找不到记录。要停批次
|
||
走 ``/stop/{batch_id}``，不是删除。
|
||
"""
|
||
|
||
import pytest
|
||
from sqlalchemy.orm import sessionmaker
|
||
|
||
import web.backend.database as database_module
|
||
from tests.support.db import get_test_engine, make_test_session
|
||
from web.backend.models import Account, LoginTask, User
|
||
from web.backend.routers.login import delete_task, delete_tasks, list_tasks
|
||
|
||
|
||
class TestTaskPagination:
|
||
def setup_method(self):
|
||
self.engine = get_test_engine()
|
||
database_module.SessionLocal.configure(bind=self.engine)
|
||
self.session = make_test_session(self.engine)
|
||
|
||
self.user = User(username="admin", password_hash="h", role="super_admin")
|
||
self.session.add(self.user)
|
||
self.session.flush()
|
||
|
||
self.account = Account(
|
||
username="分页测试账号",
|
||
password="p",
|
||
email="paged@example.com",
|
||
email_password="ep",
|
||
)
|
||
self.session.add(self.account)
|
||
self.session.flush()
|
||
|
||
# 建 12 条：7 条 failed、5 条 success，用于验证筛选与分页
|
||
self.task_ids = []
|
||
for i in range(12):
|
||
task = LoginTask(
|
||
batch_id="paged-batch",
|
||
account_id=self.account.id,
|
||
created_by=self.user.id,
|
||
status="failed" if i < 7 else "success",
|
||
message=f"任务{i}",
|
||
)
|
||
self.session.add(task)
|
||
self.session.flush()
|
||
self.task_ids.append(task.id)
|
||
self.session.commit()
|
||
|
||
def teardown_method(self):
|
||
self.session.close()
|
||
|
||
def _fresh(self):
|
||
return sessionmaker(bind=self.engine)()
|
||
|
||
def test_without_page_returns_plain_list_for_backward_compat(self):
|
||
"""老调用方（Cookie 页）拿到的必须是数组，不能是 {items,...}。"""
|
||
db = self._fresh()
|
||
try:
|
||
rows = list_tasks(db=db, current=self.user)
|
||
finally:
|
||
db.close()
|
||
assert isinstance(rows, list)
|
||
assert len(rows) == 12
|
||
|
||
def test_with_page_returns_items_and_total(self):
|
||
db = self._fresh()
|
||
try:
|
||
data = list_tasks(page=1, page_size=5, db=db, current=self.user)
|
||
finally:
|
||
db.close()
|
||
assert isinstance(data, dict)
|
||
assert data["total"] == 12
|
||
assert data["page"] == 1
|
||
assert len(data["items"]) == 5
|
||
|
||
def test_pages_do_not_overlap_and_cover_all(self):
|
||
db = self._fresh()
|
||
try:
|
||
seen = []
|
||
for page in (1, 2, 3):
|
||
data = list_tasks(page=page, page_size=5, db=db, current=self.user)
|
||
seen.extend(item.id for item in data["items"])
|
||
finally:
|
||
db.close()
|
||
assert len(seen) == 12
|
||
assert len(set(seen)) == 12, "分页出现重复"
|
||
|
||
def test_status_filter_narrows_total(self):
|
||
"""total 必须是**筛选后**的总数，否则跨页全选会多删。"""
|
||
db = self._fresh()
|
||
try:
|
||
data = list_tasks(
|
||
page=1, page_size=5, status="failed", db=db, current=self.user
|
||
)
|
||
finally:
|
||
db.close()
|
||
assert data["total"] == 7
|
||
assert all(item.status == "failed" for item in data["items"])
|
||
|
||
def test_page_beyond_range_is_empty(self):
|
||
db = self._fresh()
|
||
try:
|
||
data = list_tasks(page=99, page_size=5, db=db, current=self.user)
|
||
finally:
|
||
db.close()
|
||
assert data["items"] == []
|
||
assert data["total"] == 12
|
||
|
||
|
||
class TestBulkDelete:
|
||
def setup_method(self):
|
||
self.engine = get_test_engine()
|
||
database_module.SessionLocal.configure(bind=self.engine)
|
||
self.session = make_test_session(self.engine)
|
||
|
||
self.user = User(username="admin2", password_hash="h", role="super_admin")
|
||
self.session.add(self.user)
|
||
self.session.flush()
|
||
self.account = Account(
|
||
username="删除测试账号",
|
||
password="p",
|
||
email="del@example.com",
|
||
email_password="ep",
|
||
)
|
||
self.session.add(self.account)
|
||
self.session.flush()
|
||
|
||
self.failed_ids = []
|
||
self.pending_ids = []
|
||
for i in range(5):
|
||
for status, bucket in (
|
||
("failed", self.failed_ids),
|
||
("pending", self.pending_ids),
|
||
):
|
||
task = LoginTask(
|
||
batch_id="del-batch",
|
||
account_id=self.account.id,
|
||
created_by=self.user.id,
|
||
status=status,
|
||
message=f"{status}{i}",
|
||
)
|
||
self.session.add(task)
|
||
self.session.flush()
|
||
bucket.append(task.id)
|
||
self.session.commit()
|
||
|
||
def teardown_method(self):
|
||
self.session.close()
|
||
|
||
def _fresh(self):
|
||
return sessionmaker(bind=self.engine)()
|
||
|
||
def _remaining(self):
|
||
db = self._fresh()
|
||
try:
|
||
return {
|
||
row[0]
|
||
for row in db.query(LoginTask.id).filter(
|
||
LoginTask.batch_id == "del-batch"
|
||
)
|
||
}
|
||
finally:
|
||
db.close()
|
||
|
||
def test_delete_by_ids(self):
|
||
db = self._fresh()
|
||
try:
|
||
res = delete_tasks(
|
||
task_ids=",".join(str(i) for i in self.failed_ids[:3]),
|
||
db=db,
|
||
current=self.user,
|
||
)
|
||
finally:
|
||
db.close()
|
||
assert res["deleted"] == 3
|
||
remaining = self._remaining()
|
||
assert all(i not in remaining for i in self.failed_ids[:3])
|
||
|
||
def test_delete_all_matching_status(self):
|
||
"""跨页全选：删掉筛选条件下的全部 failed。"""
|
||
db = self._fresh()
|
||
try:
|
||
res = delete_tasks(
|
||
status="failed", all_matching=True, db=db, current=self.user
|
||
)
|
||
finally:
|
||
db.close()
|
||
assert res["deleted"] == 5
|
||
remaining = self._remaining()
|
||
assert all(i not in remaining for i in self.failed_ids)
|
||
assert all(i in remaining for i in self.pending_ids)
|
||
|
||
def test_delete_all_matching_never_touches_in_flight(self):
|
||
"""核心护栏：pending/running 必须留下，否则批次线程会找不到记录。"""
|
||
db = self._fresh()
|
||
try:
|
||
res = delete_tasks(all_matching=True, db=db, current=self.user)
|
||
finally:
|
||
db.close()
|
||
# 只有 5 条 failed 可删，5 条 pending 被护栏挡住
|
||
assert res["deleted"] == 5
|
||
remaining = self._remaining()
|
||
assert all(i in remaining for i in self.pending_ids)
|
||
|
||
def test_delete_without_criteria_rejected(self):
|
||
"""不给条件就拒绝 —— 绝不默认全删。"""
|
||
db = self._fresh()
|
||
try:
|
||
with pytest.raises(Exception) as exc:
|
||
delete_tasks(db=db, current=self.user)
|
||
finally:
|
||
db.close()
|
||
assert "请指定任务ID" in str(exc.value)
|
||
|
||
def test_delete_invalid_ids_rejected(self):
|
||
db = self._fresh()
|
||
try:
|
||
with pytest.raises(Exception) as exc:
|
||
delete_tasks(task_ids="abc,xyz", db=db, current=self.user)
|
||
finally:
|
||
db.close()
|
||
assert "无效的任务ID" in str(exc.value)
|
||
|
||
|
||
class TestCkProtection:
|
||
"""⚠️ 最重要的护栏：删除不能丢 CK。
|
||
|
||
事故背景（2026-10-03）：斗鱼账号的 CK **只存在 `login_tasks.cookie`**，
|
||
`accounts` 表没有 cookie 列。列表显示的「状态」是最近一次尝试
|
||
（``attempt_status``），于是一条**显示「失败」但实际持有有效 CK** 的记录
|
||
（``status='success'`` + cookie + 上次尝试失败）会被「清空失败」当成垃圾删掉，
|
||
账号的登录态随之永久丢失。
|
||
|
||
所以判定必须看 ``cookie``，不能看显示状态。
|
||
"""
|
||
|
||
def setup_method(self):
|
||
self.engine = get_test_engine()
|
||
database_module.SessionLocal.configure(bind=self.engine)
|
||
self.session = make_test_session(self.engine)
|
||
|
||
self.user = User(username="ck-admin", password_hash="h", role="super_admin")
|
||
self.session.add(self.user)
|
||
self.session.flush()
|
||
self.account = Account(
|
||
username="CK保护账号",
|
||
password="p",
|
||
email="ck@example.com",
|
||
email_password="ep",
|
||
)
|
||
self.session.add(self.account)
|
||
self.session.flush()
|
||
|
||
# ① 显示失败、但持有有效 CK —— 生产事故就是删到了这种
|
||
self.holds_ck_shows_failed = self._add(
|
||
status="success", attempt_status="failed", cookie="valid-ck-1"
|
||
)
|
||
# ② 成功且持有 CK
|
||
self.holds_ck_shows_success = self._add(
|
||
status="success", attempt_status="success", cookie="valid-ck-2"
|
||
)
|
||
# ③ 真垃圾：没有任何 CK
|
||
self.no_ck = self._add(status="failed", attempt_status="failed", cookie="")
|
||
|
||
# 用两个账号，避免同一账号多条记录干扰「账号是否还有 CK」的判断
|
||
self.other = Account(
|
||
username="CK保护账号2",
|
||
password="p",
|
||
email="ck2@example.com",
|
||
email_password="ep",
|
||
)
|
||
self.session.add(self.other)
|
||
self.session.flush()
|
||
self.holds_ck_other = self._add(
|
||
status="success",
|
||
attempt_status="failed",
|
||
cookie="valid-ck-3",
|
||
account_id=self.other.id,
|
||
)
|
||
self.session.commit()
|
||
|
||
def _add(self, *, status, attempt_status, cookie, account_id=None):
|
||
task = LoginTask(
|
||
batch_id="ck-batch",
|
||
account_id=account_id or self.account.id,
|
||
created_by=self.user.id,
|
||
status=status,
|
||
cookie=cookie,
|
||
attempt_status=attempt_status,
|
||
attempt_batch_id="ck-batch",
|
||
message="x",
|
||
)
|
||
self.session.add(task)
|
||
self.session.flush()
|
||
return task.id
|
||
|
||
def teardown_method(self):
|
||
self.session.close()
|
||
|
||
def _fresh(self):
|
||
return sessionmaker(bind=self.engine)()
|
||
|
||
def _exists(self, task_id):
|
||
db = self._fresh()
|
||
try:
|
||
return (
|
||
db.query(LoginTask).filter(LoginTask.id == task_id).first() is not None
|
||
)
|
||
finally:
|
||
db.close()
|
||
|
||
def test_bulk_delete_skips_ck_holding_records(self):
|
||
"""「清空失败」绝不能删掉持有 CK 的记录，哪怕它显示为失败。"""
|
||
db = self._fresh()
|
||
try:
|
||
res = delete_tasks(
|
||
status="failed", all_matching=True, db=db, current=self.user
|
||
)
|
||
finally:
|
||
db.close()
|
||
|
||
assert self._exists(self.holds_ck_shows_failed), (
|
||
"显示失败但持有 CK 的记录被删了！"
|
||
)
|
||
assert self._exists(self.holds_ck_other), "另一个账号的 CK 记录被删了！"
|
||
assert not self._exists(self.no_ck), "无 CK 的历史记录应被删除"
|
||
assert res["deleted"] == 1
|
||
assert res["skipped"] >= 2
|
||
|
||
def test_bulk_delete_reports_skipped_count(self):
|
||
db = self._fresh()
|
||
try:
|
||
res = delete_tasks(all_matching=True, db=db, current=self.user)
|
||
finally:
|
||
db.close()
|
||
# 3 条持有 CK（含另一个账号）、1 条无 CK
|
||
assert res["deleted"] == 1
|
||
assert res["skipped"] == 3
|
||
assert "持有 CK" in res["message"]
|
||
|
||
def test_explicit_ids_cannot_delete_ck_holder(self):
|
||
"""即使用户手动勾选，持有 CK 的也不能删。"""
|
||
db = self._fresh()
|
||
try:
|
||
res = delete_tasks(
|
||
task_ids=f"{self.holds_ck_shows_failed},{self.no_ck}",
|
||
db=db,
|
||
current=self.user,
|
||
)
|
||
finally:
|
||
db.close()
|
||
assert self._exists(self.holds_ck_shows_failed)
|
||
assert not self._exists(self.no_ck)
|
||
assert res["deleted"] == 1
|
||
|
||
def test_single_delete_refuses_ck_holder(self):
|
||
db = self._fresh()
|
||
try:
|
||
with pytest.raises(Exception) as exc:
|
||
delete_task(self.holds_ck_shows_success, db=db, current=self.user)
|
||
finally:
|
||
db.close()
|
||
assert "持有 CK" in str(exc.value)
|
||
assert self._exists(self.holds_ck_shows_success)
|
||
|
||
def test_single_delete_allows_empty_cookie(self):
|
||
db = self._fresh()
|
||
try:
|
||
res = delete_task(self.no_ck, db=db, current=self.user)
|
||
finally:
|
||
db.close()
|
||
assert res["success"] is True
|
||
assert not self._exists(self.no_ck)
|