chore: 移除误提交的 .bak 文件
This commit is contained in:
yml2213
@@ -1,378 +0,0 @@
|
||||
"""回归：登录任务列表分页、跨页全选删除、以及「不删在途任务」的护栏。
|
||||
|
||||
任务量早就过了单页能装下的规模（库里 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_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)
|
||||
Reference in New Issue
Block a user