"""回归：登录任务列表分页、跨页全选删除、以及「不删在途任务」的护栏。 任务量早就过了单页能装下的规模（库里 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)