chengma/cmautobuy
2
0
Fork 0
Code Issues 89 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 安全移除本地任务记录 (#108)

Browse Source
This commit is contained in:
chengma
2026-08-10 17:23:24 +08:00
parent 9d227b431a
commit b2721071c9
10 changed files with 619 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/db_schema.py
+13 -1
View File
@@ -4,7 +4,7 @@
``MIGRATIONS`` 末尾增加版本，不能修改已经发布的迁移。
"""
SCHEMA_VERSION = 3
SCHEMA_VERSION = 4
MIGRATION_1 = (
@@ -142,8 +142,20 @@ MIGRATION_3 = (
)
MIGRATION_4 = (
"""
ALTER TABLE pdd_tasks ADD COLUMN removed_at TEXT
""",
"""
CREATE INDEX idx_pdd_tasks_visible_list
ON pdd_tasks(removed_at, updated_at DESC, id DESC)
""",
)
MIGRATIONS = {
1: MIGRATION_1,
2: MIGRATION_2,
3: MIGRATION_3,
4: MIGRATION_4,
}
client/src/pdd_ui.py
+39 -9
View File
@@ -368,6 +368,7 @@ class PDDTaskPage(QWidget):
rerunRequested = pyqtSignal(str)
rerunCancelRequested = pyqtSignal()
resubmitRequested = pyqtSignal(object)
removeRequested = pyqtSignal(object)
openSettingsRequested = pyqtSignal()
detailRequested = pyqtSignal(str)
@@ -377,6 +378,7 @@ class PDDTaskPage(QWidget):
self._rerun_state = "idle"
self._resubmit_running = False
self._remove_running = False
self.taskModel = TaskTableModel(self)
self._build_command_area()
self._build_content_area()
@@ -429,6 +431,10 @@ class PDDTaskPage(QWidget):
self.resubmitButton.setAccessibleName("重新上报勾选任务的已有结果")
self.resubmitButton.setEnabled(False)
self.removeButton = PushButton(FIF.DELETE, "删除", self)
self.removeButton.setAccessibleName("从任务列表删除勾选的本地任务")
self.removeButton.setEnabled(False)
self.commandCard = CardWidget(self)
self.checkedCountLabel = CaptionLabel("已选 0 条", self.commandCard)
self.checkedCountLabel.setAccessibleName("已勾选任务数量")
@@ -446,7 +452,7 @@ class PDDTaskPage(QWidget):
if compact == self._compact_commands and self.commandLayout.count():
return
self._compact_commands = compact
for column in range(13):
for column in range(14):
self.commandLayout.setColumnStretch(column, 0)
if compact:
@@ -459,6 +465,7 @@ class PDDTaskPage(QWidget):
(self.checkedCountLabel, 0, 6, 1, 1),
(self.rerunButton, 0, 7, 1, 1),
(self.resubmitButton, 0, 8, 1, 1),
(self.removeButton, 0, 9, 1, 1),
(self.keywordLabel, 1, 0, 1, 1),
(self.keywordInput, 1, 1, 1, 5),
(self.searchButton, 1, 6, 1, 1),
@@ -479,6 +486,7 @@ class PDDTaskPage(QWidget):
(self.checkedCountLabel, 0, 10, 1, 1),
(self.rerunButton, 0, 11, 1, 1),
(self.resubmitButton, 0, 12, 1, 1),
(self.removeButton, 0, 13, 1, 1),
)
self.commandLayout.setColumnStretch(6, 4)
self.commandLayout.setColumnStretch(9, 1)
@@ -562,6 +570,7 @@ class PDDTaskPage(QWidget):
self.refreshButton.clicked.connect(self.refreshRequested.emit)
self.rerunButton.clicked.connect(self._request_rerun)
self.resubmitButton.clicked.connect(self._request_resubmit)
self.removeButton.clicked.connect(self._request_remove)
self.clearFiltersButton.clicked.connect(self.clear_filters)
self.taskTable.clicked.connect(self._on_table_clicked)
self.taskTable.activated.connect(self._on_table_activated)
@@ -634,7 +643,9 @@ class PDDTaskPage(QWidget):
text, icon, enabled = states.get(state, states["stopped"])
self.autoFetchButton.setText(text)
self.autoFetchButton.setIcon(icon)
self.autoFetchButton.setEnabled(enabled)
self.autoFetchButton.setEnabled(
enabled and not self._resubmit_running and not self._remove_running
)
self.autoFetchButton.setAccessibleName(f"{text} PDD 任务")
def set_engine_status(self, text: str) -> None:
@@ -669,6 +680,12 @@ class PDDTaskPage(QWidget):
self._resubmit_running = running
self._update_checked_actions(self.taskModel.checked_task_ids())
def set_remove_running(self, running: bool) -> None:
"""设置本地任务删除忙碌状态并防止重复操作。"""
self._remove_running = running
self._update_checked_actions(self.taskModel.checked_task_ids())
def set_rerun_cancelling(self) -> None:
"""显示取消请求已发出，防止用户重复点击。"""
@@ -686,18 +703,26 @@ class PDDTaskPage(QWidget):
self.rerunRequested.emit(task_ids[0])
def _request_resubmit(self) -> None:
if self._resubmit_running:
if self._resubmit_running or self._remove_running:
return
task_ids = self.taskModel.checked_task_ids()
if task_ids:
self.resubmitRequested.emit(task_ids)
def _request_remove(self) -> None:
if self._remove_running or self._resubmit_running:
return
task_ids = self.taskModel.checked_task_ids()
if task_ids:
self.removeRequested.emit(task_ids)
def _update_rerun_button(self) -> None:
if self._rerun_state == "running":
self.rerunButton.setText("停止重新采集")
self.rerunButton.setAccessibleName("停止当前重新采集")
self.rerunButton.setEnabled(True)
self.resubmitButton.setEnabled(False)
self.removeButton.setEnabled(False)
self.autoFetchButton.setEnabled(False)
return
if self._rerun_state == "cancelling":
@@ -705,19 +730,22 @@ class PDDTaskPage(QWidget):
self.rerunButton.setAccessibleName("正在停止重新采集")
self.rerunButton.setEnabled(False)
self.resubmitButton.setEnabled(False)
self.removeButton.setEnabled(False)
self.autoFetchButton.setEnabled(False)
return
self.rerunButton.setText("重新执行")
self.rerunButton.setAccessibleName("重新执行当前选中的采集任务")
self.autoFetchButton.setEnabled(not self._resubmit_running)
idle = not self._resubmit_running and not self._remove_running
self.autoFetchButton.setEnabled(idle)
self.rerunButton.setEnabled(
not self._resubmit_running
and len(self.taskModel.checked_task_ids()) == 1
idle and len(self.taskModel.checked_task_ids()) == 1
)
self.resubmitButton.setEnabled(
not self._resubmit_running
and bool(self.taskModel.checked_task_ids())
idle and bool(self.taskModel.checked_task_ids())
)
self.removeButton.setEnabled(
idle and bool(self.taskModel.checked_task_ids())
)
def _update_checked_actions(self, task_ids=None) -> None:
@@ -728,7 +756,9 @@ class PDDTaskPage(QWidget):
self.resubmitButton.setText(
"正在上报…" if self._resubmit_running else "重新上报"
)
self.resubmitButton.setEnabled(bool(checked) and not self._resubmit_running)
self.removeButton.setText(
"正在删除…" if self._remove_running else "删除"
)
self._update_rerun_button()
def _apply_filters(self) -> None:
client/src/pdd_ui_event.py
+158 -5
View File
@@ -59,7 +59,11 @@ from .task_models import (
TaskSummary,
TaskType,
)
from .task_repository import CollectRerunError, TaskRepository
from .task_repository import (
CollectRerunError,
TaskRemovalError,
TaskRepository,
)
from .task_dispatcher import (
TaskDispatcher,
admin_task_to_new_claimed_task,
@@ -322,6 +326,36 @@ class ResultResubmitWorker(QObject):
self.completed.emit()
class TaskRemoveWorker(QObject):
"""在后台原子校验并软移除勾选任务，不访问任何界面控件。"""
succeeded = pyqtSignal(object)
failed = pyqtSignal(str)
completed = pyqtSignal()
def __init__(
self,
repository: TaskRepository,
task_ids: tuple[str, ...],
) -> None:
super().__init__()
self._repository = repository
self._task_ids = task_ids
@pyqtSlot()
def run(self) -> None:
try:
self._repository.remove_tasks_from_list(self._task_ids)
except TaskRemovalError as exc:
self.failed.emit(str(exc))
except Exception:
self.failed.emit("删除本地任务失败，请检查数据库后重试。")
else:
self.succeeded.emit(self._task_ids)
finally:
self.completed.emit()
class PDDTaskPageEvent(QObject):
"""把 PDD 页面只读操作连接到本地任务 Repository。"""
@@ -359,6 +393,9 @@ class PDDTaskPageEvent(QObject):
self._resubmit_busy = False
self._resubmit_thread: Optional[QThread] = None
self._resubmit_worker: Optional[ResultResubmitWorker] = None
self._remove_busy = False
self._remove_thread: Optional[QThread] = None
self._remove_worker: Optional[TaskRemoveWorker] = None
self._stop_requested = False
self._cycle_next_delay_ms: Optional[int] = None
self._stop_status = "自动获取：已停止 · 当前没有执行中的任务"
@@ -417,6 +454,7 @@ class PDDTaskPageEvent(QObject):
page.rerunRequested.connect(self.request_rerun)
page.rerunCancelRequested.connect(self.request_cancel_rerun)
page.resubmitRequested.connect(self.request_resubmit)
page.removeRequested.connect(self.request_remove)
page.autoFetchRequested.connect(self._request_claim_task)
page.detailRequested.connect(self.show_task_detail)
page.taskModel.loadMoreRequested.connect(self._load_page)
@@ -452,7 +490,12 @@ class PDDTaskPageEvent(QObject):
if self._closing or not task_id:
return
if self._auto_fetch_running or self._claim_busy or self._resubmit_busy:
if (
self._auto_fetch_running
or self._claim_busy
or self._resubmit_busy
or self._remove_busy
):
self._show_rerun_warning(
"暂时不能重新执行",
"自动获取或其他采集正在运行，请停止并等待当前任务结束。",
@@ -517,7 +560,12 @@ class PDDTaskPageEvent(QObject):
)
if self._closing or not stable_ids:
return
if self._auto_fetch_running or self._claim_busy or self._resubmit_busy:
if (
self._auto_fetch_running
or self._claim_busy
or self._resubmit_busy
or self._remove_busy
):
self._show_rerun_warning(
"暂时不能重新上报",
"自动获取、重新执行或另一批上报正在运行，请等待当前操作结束。",
@@ -545,6 +593,98 @@ class PDDTaskPageEvent(QObject):
self._start_resubmit_worker(stable_ids)
@pyqtSlot(object)
def request_remove(self, task_ids) -> None:
"""确认后在后台从普通列表软移除勾选的本地任务。"""
stable_ids = tuple(
dict.fromkeys(
str(value).strip()
for value in task_ids
if value is not None and str(value).strip()
)
)
if self._closing or not stable_ids:
return
if (
self._auto_fetch_running
or self._claim_busy
or self._resubmit_busy
or self._remove_busy
):
self._show_rerun_warning(
"暂时不能删除",
"自动获取、重新执行、重新上报或另一批删除正在运行，"
"请等待当前操作结束。",
)
return
count = len(stable_ids)
dialog = MessageBox(
f"从列表删除 {count} 条任务？",
"只会从普通任务列表中隐藏这些本地记录，不会删除 Admin 任务；"
"执行记录和上报历史仍会永久保留。\n\n"
"只有已结束、没有待上报数据且没有进入不可逆阶段的任务才允许删除。",
self._page.window(),
)
dialog.yesButton.setText(f"删除 {count} 条")
dialog.cancelButton.setText("暂不删除")
dialog.cancelButton.setFocus()
if not dialog.exec():
return
self._start_remove_worker(stable_ids)
def _start_remove_worker(self, task_ids: tuple[str, ...]) -> None:
"""启动本地任务软删除工作线程。"""
self._remove_busy = True
self._page.set_remove_running(True)
self._page.set_engine_status(
f"正在检查并删除 {len(task_ids)} 条本地任务…"
)
thread = QThread(self)
worker = TaskRemoveWorker(self._repository, task_ids)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.succeeded.connect(self._on_remove_succeeded)
worker.failed.connect(self._on_remove_failed)
worker.completed.connect(thread.quit)
worker.completed.connect(worker.deleteLater)
thread.finished.connect(thread.deleteLater)
thread.finished.connect(self._on_remove_thread_finished)
self._remove_thread = thread
self._remove_worker = worker
thread.start()
@pyqtSlot(object)
def _on_remove_succeeded(self, task_ids) -> None:
if self._closing:
return
stable_ids = tuple(task_ids)
self._page.taskModel.uncheck_tasks(stable_ids)
self._page.set_engine_status(
f"已从列表删除 {len(stable_ids)} 条本地任务；"
"执行记录和上报历史仍会永久保留。"
)
self._reload()
@pyqtSlot(str)
def _on_remove_failed(self, message: str) -> None:
if self._closing:
return
self._page.set_engine_status(f"删除未完成：{message}")
self._show_claim_error("不能删除所选任务", message)
@pyqtSlot()
def _on_remove_thread_finished(self) -> None:
self._remove_worker = None
self._remove_thread = None
self._remove_busy = False
if not self._closing:
self._page.set_remove_running(False)
def _start_resubmit_worker(self, task_ids: tuple[str, ...]) -> None:
"""启动只处理指定任务 Outbox 的工作线程。"""
@@ -774,10 +914,10 @@ class PDDTaskPageEvent(QObject):
if self._closing:
return
if self._resubmit_busy:
if self._resubmit_busy or self._remove_busy:
self._show_rerun_warning(
"暂时不能获取任务",
"任务结果正在重新上报，请等待当前操作结束。",
"本地任务正在重新上报或删除，请等待当前操作结束。",
)
return
if self._auto_fetch_running:
@@ -1283,6 +1423,19 @@ class PDDTaskPageEvent(QObject):
resubmit_thread.quit()
resubmit_thread.wait(60_000)
remove_worker = self._remove_worker
remove_thread = self._remove_thread
if remove_worker is not None:
try:
remove_worker.succeeded.disconnect(self._on_remove_succeeded)
remove_worker.failed.disconnect(self._on_remove_failed)
except (TypeError, RuntimeError):
pass
if remove_thread is not None and remove_thread.isRunning():
# 删除只包含一个很短的 SQLite 事务，不中途取消，避免部分更新。
remove_thread.quit()
remove_thread.wait(60_000)
def summary_to_row(summary: TaskSummary) -> TaskRow:
"""把领域摘要转换成只供表格显示的轻量行。"""
client/src/task_repository.py
+98 -2
View File
@@ -7,7 +7,7 @@ import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Union
from typing import Dict, Iterable, List, Optional, Tuple, Union
from uuid import uuid4
from .db import initialize_database, open_database
@@ -39,6 +39,10 @@ class CollectRerunError(ValueError):
"""当前任务不满足重新采集条件。"""
class TaskRemovalError(ValueError):
"""勾选任务不满足从普通列表移除的安全条件。"""
def utc_now_iso() -> str:
"""返回精确到秒的 UTC ISO 8601 时间。"""
@@ -145,6 +149,98 @@ class TaskRepository:
connection.close()
return int(row[0])
def remove_tasks_from_list(self, remote_task_ids: Iterable[str]) -> int:
"""把符合安全条件的终态任务从普通列表中软移除。
全部任务会在同一个事务中完成校验和更新；任一任务不安全时，
所有任务都保持原样。执行记录和 Outbox 不会被删除或修改。
"""
task_ids = tuple(
dict.fromkeys(
str(value).strip()
for value in remote_task_ids
if value is not None and str(value).strip()
)
)
if not task_ids:
raise TaskRemovalError("没有可删除的任务，请重新勾选。")
placeholders = ", ".join("?" for _ in task_ids)
terminal_statuses = {
TaskStatus.SUCCEEDED.value,
TaskStatus.FAILED.value,
TaskStatus.CANCELLED.value,
}
now = utc_now_iso()
connection = open_database(self._db_path)
try:
with connection:
rows = connection.execute(
"SELECT id, remote_task_id, status, removed_at FROM pdd_tasks"
f" WHERE remote_task_id IN ({placeholders})",
task_ids,
).fetchall()
found = {row["remote_task_id"]: row for row in rows}
missing = [task_id for task_id in task_ids if task_id not in found]
if missing:
raise TaskRemovalError(
f"本地找不到任务 {missing[0]}，请刷新列表后重试。"
)
removed = next(
(row for row in rows if row["removed_at"] is not None), None
)
if removed is not None:
raise TaskRemovalError(
f"任务 {removed['remote_task_id']} 已不在普通列表，请刷新后重试。"
)
non_terminal = next(
(row for row in rows if row["status"] not in terminal_statuses),
None,
)
if non_terminal is not None:
raise TaskRemovalError(
f"任务 {non_terminal['remote_task_id']} 尚未结束，不能删除。"
)
irreversible = connection.execute(
"SELECT t.remote_task_id FROM task_runs r"
" JOIN pdd_tasks t ON t.id = r.task_id"
f" WHERE t.remote_task_id IN ({placeholders})"
" AND r.irreversible_action_at IS NOT NULL LIMIT 1",
task_ids,
).fetchone()
if irreversible is not None:
raise TaskRemovalError(
f"任务 {irreversible['remote_task_id']} 已进入不可逆阶段，"
"必须保留在列表中核对订单。"
)
unsent = connection.execute(
"SELECT t.remote_task_id, o.status FROM outbox_events o"
" JOIN pdd_tasks t ON t.id = o.task_id"
f" WHERE t.remote_task_id IN ({placeholders})"
" AND o.status <> 'sent' LIMIT 1",
task_ids,
).fetchone()
if unsent is not None:
raise TaskRemovalError(
f"任务 {unsent['remote_task_id']} 还有未发送完成的数据，"
"请先重新上报。"
)
cursor = connection.execute(
"UPDATE pdd_tasks SET removed_at = ?, updated_at = ?"
f" WHERE remote_task_id IN ({placeholders})"
" AND removed_at IS NULL",
(now, now, *task_ids),
)
if cursor.rowcount != len(task_ids):
raise TaskRemovalError("任务列表已发生变化，请刷新后重试。")
finally:
connection.close()
return len(task_ids)
def get_task(self, remote_task_id: str) -> Optional[TaskDetail]:
"""按稳定远程编号读取完整任务；不存在时返回 None。"""
@@ -1594,7 +1690,7 @@ class TaskRepository:
@staticmethod
def _build_where(filters: TaskFilters) -> Tuple[str, List[object]]:
clauses = []
clauses = ["removed_at IS NULL"]
parameters: List[object] = []
if filters.task_type is not None:
clauses.append("task_type = ?")
client/test/test_db.py
+47 -4
View File
@@ -6,7 +6,7 @@ import unittest
from pathlib import Path
from src.db import DatabaseVersionError, initialize_database, open_database
from src.db_schema import MIGRATION_1, MIGRATION_2
from src.db_schema import MIGRATION_1, MIGRATION_2, MIGRATION_3
EXPECTED_TABLES = {
@@ -22,6 +22,7 @@ EXPECTED_INDEXES = {
"idx_pdd_tasks_goods_id",
"idx_task_runs_task",
"idx_outbox_pending",
"idx_pdd_tasks_visible_list",
}
@@ -63,7 +64,7 @@ class DatabaseInitializationTests(unittest.TestCase):
self.assertTrue(EXPECTED_TABLES.issubset(tables))
self.assertTrue(EXPECTED_INDEXES.issubset(indexes))
self.assertEqual(version, 3)
self.assertEqual(version, 4)
def test_v1_database_is_upgraded_without_losing_task_runs(self) -> None:
connection = open_database(self.db_path)
@@ -106,7 +107,7 @@ class DatabaseInitializationTests(unittest.TestCase):
connection.close()
self.assertIn("result_data", columns)
self.assertEqual(attempt_id, "ATTEMPT-OLD")
self.assertEqual(version, 3)
self.assertEqual(version, 4)
def test_initialize_can_run_twice_without_losing_data(self) -> None:
initialize_database(self.db_path)
@@ -169,7 +170,49 @@ class DatabaseInitializationTests(unittest.TestCase):
finally:
connection.close()
self.assertEqual(row[0], "dry_run")
self.assertEqual(version, 3)
self.assertEqual(version, 4)
def test_v3_database_adds_soft_remove_column_without_losing_tasks(self) -> None:
connection = open_database(self.db_path)
try:
with connection:
for statement in MIGRATION_1 + MIGRATION_2 + MIGRATION_3:
connection.execute(statement)
connection.execute("PRAGMA user_version = 3")
connection.execute(
"INSERT INTO pdd_tasks"
" (remote_task_id, task_type, goods_url, status, received_at,"
" created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
(
"COL-V3",
"collect",
"https://example.test/goods",
"succeeded",
"2026-08-10T00:00:00Z",
"2026-08-10T00:00:00Z",
"2026-08-10T00:00:00Z",
),
)
finally:
connection.close()
initialize_database(self.db_path)
connection = open_database(self.db_path)
try:
columns = {
row[1] for row in connection.execute("PRAGMA table_info(pdd_tasks)")
}
row = connection.execute(
"SELECT remote_task_id, removed_at FROM pdd_tasks"
" WHERE remote_task_id = 'COL-V3'"
).fetchone()
version = connection.execute("PRAGMA user_version").fetchone()[0]
finally:
connection.close()
self.assertIn("removed_at", columns)
self.assertEqual(tuple(row), ("COL-V3", None))
self.assertEqual(version, 4)
def test_new_connection_uses_required_pragmas(self) -> None:
initialize_database(self.db_path)
client/test/test_pdd_ui.py
+23
View File
@@ -63,9 +63,12 @@ class TaskTableModelCheckTests(unittest.TestCase):
refresh_position = layout.getItemPosition(layout.indexOf(page.refreshButton))
rerun_position = layout.getItemPosition(layout.indexOf(page.rerunButton))
resubmit_position = layout.getItemPosition(layout.indexOf(page.resubmitButton))
remove_position = layout.getItemPosition(layout.indexOf(page.removeButton))
self.assertEqual(refresh_position[1], search_position[1] + 1)
self.assertEqual(resubmit_position[1], rerun_position[1] + 1)
self.assertEqual(remove_position[1], resubmit_position[1] + 1)
self.assertEqual(page.checkedCountLabel.text(), "已选 1 条")
self.assertTrue(page.removeButton.isEnabled())
page.resize(1000, 650)
self.app.processEvents()
@@ -78,6 +81,26 @@ class TaskTableModelCheckTests(unittest.TestCase):
self.assertEqual(page.taskModel.checked_task_ids(), ())
self.assertEqual(page.checkedCountLabel.text(), "已选 0 条")
self.assertFalse(page.removeButton.isEnabled())
page.deleteLater()
def test_remove_emits_stable_checked_task_ids(self):
page = PDDTaskPage()
page.set_tasks(
[TaskRow("TASK-1", "采集"), TaskRow("TASK-2", "采集")]
)
received = []
page.removeRequested.connect(received.append)
for row in (0, 1):
page.taskModel.setData(
page.taskModel.index(row, CHECK_COLUMN),
Qt.Checked,
Qt.CheckStateRole,
)
page.removeButton.click()
self.assertEqual(received, [("TASK-1", "TASK-2")])
page.deleteLater()
def test_auto_fetch_text_and_keyword_width(self):
client/test/test_pdd_ui_event.py
+99
View File
@@ -627,6 +627,105 @@ class PDDTaskPageEventTest(unittest.TestCase):
events.shutdown()
page.deleteLater()
def test_remove_confirmation_cancel_keeps_local_task(self):
self._add_task(1)
connection = open_database(self.db_path)
try:
with connection:
connection.execute(
"UPDATE pdd_tasks SET status = 'succeeded'"
" WHERE remote_task_id = 'PDD-001'"
)
finally:
connection.close()
page = PDDTaskPage()
gateway = RecordingClaimGateway()
events = PDDTaskPageEvent(page, self.repository, claim_gateway=gateway)
events.load_initial_tasks()
page.taskModel.setData(
page.taskModel.index(0, 0), Qt.Checked, Qt.CheckStateRole
)
with patch("src.pdd_ui_event.MessageBox") as message_box:
dialog = message_box.return_value
dialog.exec.return_value = False
page.removeButton.click()
message_box.assert_called_once()
self.assertIs(message_box.call_args.args[2], page.window())
dialog.cancelButton.setFocus.assert_called_once()
self.assertEqual(self.repository.count_tasks(), 1)
self.assertFalse(events._remove_busy)
self.assertEqual(gateway.calls, [])
events.shutdown()
page.deleteLater()
def test_remove_terminal_task_refreshes_list_and_keeps_audit_data(self):
self._add_task(1)
connection = open_database(self.db_path)
try:
with connection:
connection.execute(
"UPDATE pdd_tasks SET status = 'failed'"
" WHERE remote_task_id = 'PDD-001'"
)
finally:
connection.close()
page = PDDTaskPage()
gateway = RecordingClaimGateway()
events = PDDTaskPageEvent(page, self.repository, claim_gateway=gateway)
events.load_initial_tasks()
page.taskModel.setData(
page.taskModel.index(0, 0), Qt.Checked, Qt.CheckStateRole
)
with patch("src.pdd_ui_event.MessageBox") as message_box:
message_box.return_value.exec.return_value = True
page.removeButton.click()
self.assertTrue(wait_until(self.app, lambda: not events._remove_busy))
self.assertEqual(page.taskModel.data_row_count(), 0)
self.assertEqual(page.taskModel.checked_task_ids(), ())
self.assertIsNotNone(self.repository.get_task("PDD-001"))
self.assertIn("已从列表删除 1 条", page.statusLabel.text())
self.assertEqual(gateway.calls, [])
self.device_checker.assert_not_called()
events.shutdown()
page.deleteLater()
def test_remove_mixed_batch_reports_reason_and_changes_nothing(self):
self._add_task(1)
self._add_task(2)
connection = open_database(self.db_path)
try:
with connection:
connection.execute(
"UPDATE pdd_tasks SET status = 'cancelled'"
" WHERE remote_task_id = 'PDD-001'"
)
finally:
connection.close()
page = PDDTaskPage()
events = PDDTaskPageEvent(
page, self.repository, claim_gateway=RecordingClaimGateway()
)
events.load_initial_tasks()
for row in range(2):
page.taskModel.setData(
page.taskModel.index(row, 0), Qt.Checked, Qt.CheckStateRole
)
with patch("src.pdd_ui_event.MessageBox") as message_box:
message_box.return_value.exec.return_value = True
page.removeButton.click()
self.assertTrue(wait_until(self.app, lambda: not events._remove_busy))
self.assertEqual(self.repository.count_tasks(), 2)
self.assertEqual(len(page.taskModel.checked_task_ids()), 2)
self.assertIn("尚未结束", page.statusLabel.text())
events.shutdown()
page.deleteLater()
def test_rerun_confirmation_has_clear_safe_cancel_action(self):
self._add_task(1)
first = self.repository.start_collect_run("PDD-001", "USB-001")
client/test/test_task_repository.py
+119 -1
View File
@@ -14,7 +14,12 @@ from src.task_models import (
TaskStatus,
TaskType,
)
from src.task_repository import CollectRerunError, DuplicateTaskError, TaskRepository
from src.task_repository import (
CollectRerunError,
DuplicateTaskError,
TaskRemovalError,
TaskRepository,
)
class TaskRepositoryTests(unittest.TestCase):
@@ -26,6 +31,17 @@ class TaskRepositoryTests(unittest.TestCase):
def tearDown(self) -> None:
self._temporary_directory.cleanup()
def _set_task_status(self, task_id: str, status: TaskStatus) -> None:
connection = open_database(self.db_path)
try:
with connection:
connection.execute(
"UPDATE pdd_tasks SET status = ? WHERE remote_task_id = ?",
(status.value, task_id),
)
finally:
connection.close()
@staticmethod
def _task(
remote_task_id: str,
@@ -516,6 +532,108 @@ class TaskRepositoryTests(unittest.TestCase):
OutboxStatus.PENDING,
)
def test_terminal_task_is_soft_removed_but_audit_record_remains(self):
self.repository.add_claimed_task(self._task("REMOVE-OK"))
self._set_task_status("REMOVE-OK", TaskStatus.SUCCEEDED)
removed_count = self.repository.remove_tasks_from_list(("REMOVE-OK",))
self.assertEqual(removed_count, 1)
self.assertEqual(self.repository.list_tasks(), [])
self.assertEqual(self.repository.count_tasks(), 0)
self.assertIsNotNone(self.repository.get_task("REMOVE-OK"))
connection = open_database(self.db_path)
try:
row = connection.execute(
"SELECT removed_at FROM pdd_tasks"
" WHERE remote_task_id = 'REMOVE-OK'"
).fetchone()
finally:
connection.close()
self.assertIsNotNone(row[0])
with self.assertRaises(DuplicateTaskError):
self.repository.add_claimed_task(self._task("REMOVE-OK"))
def test_batch_remove_is_all_or_nothing_for_non_terminal_task(self):
self.repository.add_claimed_task(self._task("REMOVE-ENDED"))
self.repository.add_claimed_task(self._task("REMOVE-RUNNING"))
self._set_task_status("REMOVE-ENDED", TaskStatus.FAILED)
self._set_task_status("REMOVE-RUNNING", TaskStatus.RUNNING)
with self.assertRaisesRegex(TaskRemovalError, "尚未结束"):
self.repository.remove_tasks_from_list(
("REMOVE-ENDED", "REMOVE-RUNNING")
)
self.assertEqual(self.repository.count_tasks(), 2)
def test_task_with_unsent_outbox_cannot_be_removed(self):
self.repository.add_claimed_task(self._task("REMOVE-UNSENT"))
started = self.repository.start_collect_run("REMOVE-UNSENT", "USB-001")
self.repository.save_collect_failure(
"REMOVE-UNSENT",
started.attempt_id,
TaskStatus.FAILED,
"TEST_FAILURE",
"测试失败",
False,
)
with self.assertRaisesRegex(TaskRemovalError, "未发送完成"):
self.repository.remove_tasks_from_list(("REMOVE-UNSENT",))
self.assertEqual(self.repository.count_tasks(), 1)
def test_irreversible_task_cannot_be_removed_even_when_terminal(self):
self.repository.add_claimed_task(
self._task(
"REMOVE-IRREVERSIBLE",
TaskType.PURCHASE,
execution_mode="live",
)
)
started = self.repository.start_purchase_run(
"REMOVE-IRREVERSIBLE", "USB-001"
)
self.repository.mark_purchase_irreversible(
"REMOVE-IRREVERSIBLE", started.attempt_id
)
self._set_task_status("REMOVE-IRREVERSIBLE", TaskStatus.SUCCEEDED)
with self.assertRaisesRegex(TaskRemovalError, "不可逆阶段"):
self.repository.remove_tasks_from_list(("REMOVE-IRREVERSIBLE",))
self.assertEqual(self.repository.count_tasks(), 1)
def test_sent_outbox_does_not_block_soft_remove(self):
self.repository.add_claimed_task(self._task("REMOVE-SENT"))
started = self.repository.start_collect_run("REMOVE-SENT", "USB-001")
event = self.repository.save_collect_result(
"REMOVE-SENT",
started.attempt_id,
{"goods_id": "10001", "title": "测试商品", "skus": []},
)
self.repository.mark_outbox_sent(event.id)
self.assertEqual(
self.repository.remove_tasks_from_list(("REMOVE-SENT",)), 1
)
self.assertEqual(self.repository.count_tasks(), 0)
connection = open_database(self.db_path)
try:
run_count = connection.execute(
"SELECT COUNT(*) FROM task_runs r JOIN pdd_tasks t"
" ON t.id = r.task_id WHERE t.remote_task_id = 'REMOVE-SENT'"
).fetchone()[0]
outbox_row = connection.execute(
"SELECT o.status FROM outbox_events o JOIN pdd_tasks t"
" ON t.id = o.task_id WHERE t.remote_task_id = 'REMOVE-SENT'"
).fetchone()
finally:
connection.close()
self.assertEqual(run_count, 1)
self.assertEqual(outbox_row[0], OutboxStatus.SENT.value)
if __name__ == "__main__":
unittest.main()