118 lines
3.5 KiB
Python
118 lines
3.5 KiB
Python
|
import os
|
||
|
|
import unittest
|
||
|
|
|
||
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
|
|
|
||
|
|
from _helpers import REPO_ROOT # noqa: F401
|
||
|
|
|
||
|
|
from app import workers
|
||
|
|
|
||
|
|
if workers.QT_IMPORT_ERROR is not None:
|
||
|
|
raise unittest.SkipTest("PySide6 未安装")
|
||
|
|
|
||
|
|
from PySide6.QtCore import QEventLoop, QTimer
|
||
|
|
from PySide6.QtWidgets import QApplication
|
||
|
|
|
||
|
|
from app.workers import BaseWorker, run_worker
|
||
|
|
|
||
|
|
|
||
|
|
class DemoWorker(BaseWorker):
|
||
|
|
def execute(self):
|
||
|
|
self.log.emit("开始")
|
||
|
|
self.progress.emit({"done": 1, "total": 1})
|
||
|
|
self.row_updated.emit(7, {"status": "success"})
|
||
|
|
return {"ok": True, "done": 1}
|
||
|
|
|
||
|
|
|
||
|
|
class CancelAwareWorker(BaseWorker):
|
||
|
|
def execute(self):
|
||
|
|
if self.should_cancel():
|
||
|
|
return {"done": 0}
|
||
|
|
self.cancel()
|
||
|
|
return {"done": 0}
|
||
|
|
|
||
|
|
|
||
|
|
class FailingWorker(BaseWorker):
|
||
|
|
def execute(self):
|
||
|
|
raise RuntimeError("模拟失败")
|
||
|
|
|
||
|
|
|
||
|
|
class WorkerTests(unittest.TestCase):
|
||
|
|
@classmethod
|
||
|
|
def setUpClass(cls):
|
||
|
|
cls.app = QApplication.instance() or QApplication([])
|
||
|
|
|
||
|
|
def start_and_wait(self, thread, timeout_ms=2000):
|
||
|
|
loop = QEventLoop()
|
||
|
|
finished = []
|
||
|
|
|
||
|
|
def on_finished():
|
||
|
|
finished.append(True)
|
||
|
|
loop.quit()
|
||
|
|
|
||
|
|
timer = QTimer()
|
||
|
|
timer.setSingleShot(True)
|
||
|
|
timer.timeout.connect(loop.quit)
|
||
|
|
thread.finished.connect(on_finished)
|
||
|
|
timer.start(timeout_ms)
|
||
|
|
thread.start()
|
||
|
|
loop.exec()
|
||
|
|
|
||
|
|
self.assertTrue(finished, "worker thread did not finish before timeout")
|
||
|
|
|
||
|
|
def test_worker_emits_common_signals_and_finishes(self):
|
||
|
|
worker = DemoWorker()
|
||
|
|
logs = []
|
||
|
|
progress = []
|
||
|
|
rows = []
|
||
|
|
finished = []
|
||
|
|
|
||
|
|
worker.log.connect(logs.append)
|
||
|
|
worker.progress.connect(lambda payload: progress.append(dict(payload)))
|
||
|
|
worker.row_updated.connect(lambda task_id, fields: rows.append((task_id, dict(fields))))
|
||
|
|
worker.finished.connect(lambda payload: finished.append(dict(payload)))
|
||
|
|
|
||
|
|
thread = run_worker(worker, start=False)
|
||
|
|
self.start_and_wait(thread)
|
||
|
|
|
||
|
|
self.assertEqual(["开始"], logs)
|
||
|
|
self.assertEqual([{"done": 1, "total": 1}], progress)
|
||
|
|
self.assertEqual([(7, {"status": "success"})], rows)
|
||
|
|
self.assertEqual([{"ok": True, "done": 1}], finished)
|
||
|
|
|
||
|
|
def test_cancel_flag_emits_cancelled_instead_of_finished(self):
|
||
|
|
worker = CancelAwareWorker()
|
||
|
|
cancelled = []
|
||
|
|
finished = []
|
||
|
|
|
||
|
|
worker.cancelled.connect(lambda payload: cancelled.append(dict(payload)))
|
||
|
|
worker.finished.connect(lambda payload: finished.append(dict(payload)))
|
||
|
|
|
||
|
|
thread = run_worker(worker, start=False)
|
||
|
|
self.start_and_wait(thread)
|
||
|
|
|
||
|
|
self.assertEqual([{"done": 0, "cancelled": True}], cancelled)
|
||
|
|
self.assertEqual([], finished)
|
||
|
|
|
||
|
|
def test_uncaught_exception_emits_failed_and_finished_summary(self):
|
||
|
|
worker = FailingWorker()
|
||
|
|
failed = []
|
||
|
|
finished = []
|
||
|
|
|
||
|
|
worker.failed.connect(lambda task_id, error: failed.append((task_id, error)))
|
||
|
|
worker.finished.connect(lambda payload: finished.append(dict(payload)))
|
||
|
|
|
||
|
|
thread = run_worker(worker, start=False)
|
||
|
|
self.start_and_wait(thread)
|
||
|
|
|
||
|
|
self.assertEqual([(-1, "模拟失败")], failed)
|
||
|
|
self.assertEqual([{"ok": False, "error": "模拟失败"}], finished)
|
||
|
|
|
||
|
|
def test_run_worker_rejects_plain_object(self):
|
||
|
|
with self.assertRaises(TypeError):
|
||
|
|
run_worker(object(), start=False)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|