SXXXXXXX_PyUCC/tests/test_worker.py

87 lines
2.0 KiB
Python

import time
from pyucc.core.worker import WorkerManager
def test_submit_and_dispatch():
wm = WorkerManager()
result = {}
def on_done(payload):
result["done"] = payload
task_id = wm.submit(lambda: "ok", on_done=on_done)
# collect messages from ui_queue and dispatch them
got_done = False
start_time = time.time()
while time.time() - start_time < 5:
try:
msg = wm.ui_queue.get(timeout=1)
except Exception:
break
wm.dispatch_message(msg)
if msg[0] == "done":
got_done = True
break
assert got_done
assert "done" in result
def test_map_iterable_progress_and_done():
wm = WorkerManager()
progresses = []
done_res = {}
def on_progress(payload):
progresses.append(payload)
def on_done(payload):
done_res["done"] = payload
items = [1, 2, 3]
task_id = wm.map_iterable(
lambda x: x * 2, items, on_progress=on_progress, on_done=on_done
)
# drain ui_queue and dispatch until done
start = time.time()
saw_done = False
while time.time() - start < 5:
try:
msg = wm.ui_queue.get(timeout=1)
except Exception:
break
wm.dispatch_message(msg)
if msg[0] == "done":
saw_done = True
break
assert saw_done
# progress callbacks should have been invoked
assert len(progresses) == 3
assert "done" in done_res
def test_cancel_emits_cancelled():
wm = WorkerManager()
def slow():
time.sleep(0.5)
return "done"
tid = wm.submit(slow)
# request cancel
cancelled = wm.cancel(tid)
# ui_queue should include a cancelled message
saw_cancelled = False
start = time.time()
while time.time() - start < 3:
try:
msg = wm.ui_queue.get(timeout=1)
except Exception:
break
if msg[0] == "cancelled":
saw_cancelled = True
break
assert saw_cancelled