From b89d7f2d27c06155ecb6805042b4c97f5a5da5b2 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sat, 3 Oct 2026 10:52:02 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E7=A7=BB=E9=99=A4=E8=AF=AF=E6=8F=90?= =?UTF-8?q?=E4=BA=A4=E7=9A=84=20.bak=20=E6=96=87=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_login_tasks_pagination.py.bak | 378 ----------------------- 1 file changed, 378 deletions(-) delete mode 100644 tests/test_login_tasks_pagination.py.bak diff --git a/tests/test_login_tasks_pagination.py.bak b/tests/test_login_tasks_pagination.py.bak deleted file mode 100644 index 23be7fc..0000000 --- a/tests/test_login_tasks_pagination.py.bak +++ /dev/null @@ -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)