mirror of
https://github.com/Textualize/textual.git
synced 2025-10-17 02:38:12 +03:00
b5689b1f69
* worker class * worker API tests * tidy * Decorator and more tests * type fix * error order * more tests * remove active message * move worker manager to app * cancel nodes * typing fix * revert change * typing fixes and cleanup * revert typing * test fix * cancel group * Added test for worker * comment * workers docs * Added exit_on_error * changelog * svg * refactor test * remove debug tweaks * docstrings * worker test * fix typing in run * fix 3.7 tests * blog post * fix deadlock test * words * words * words * workers docs * blog post * Apply suggestions from code review Co-authored-by: Dave Pearson <davep@davep.org> * docstring * fix and docstring * Apply suggestions from code review Co-authored-by: Rodrigo Girão Serrão <5621605+rodrigogiraoserrao@users.noreply.github.com> * Update src/textual/widgets/_markdown.py Co-authored-by: Rodrigo Girão Serrão <5621605+rodrigogiraoserrao@users.noreply.github.com> * Apply suggestions from code review Co-authored-by: Rodrigo Girão Serrão <5621605+rodrigogiraoserrao@users.noreply.github.com> * Update src/textual/worker.py Co-authored-by: Rodrigo Girão Serrão <5621605+rodrigogiraoserrao@users.noreply.github.com> * Fix black * docstring * merge * changelog --------- Co-authored-by: Dave Pearson <davep@davep.org> Co-authored-by: Rodrigo Girão Serrão <5621605+rodrigogiraoserrao@users.noreply.github.com>
97 lines
2.5 KiB
Python
97 lines
2.5 KiB
Python
import asyncio
|
|
import time
|
|
|
|
from textual.app import App, ComposeResult
|
|
from textual.widget import Widget
|
|
from textual.worker import Worker, WorkerState
|
|
|
|
|
|
def test_worker_manager_init():
|
|
app = App()
|
|
assert isinstance(repr(app.workers), str)
|
|
assert not bool(app.workers)
|
|
assert len(app.workers) == 0
|
|
assert list(app.workers) == []
|
|
assert list(reversed(app.workers)) == []
|
|
|
|
|
|
async def test_run_worker_async() -> None:
|
|
"""Check self.run_worker"""
|
|
worker_events: list[Worker.StateChanged] = []
|
|
|
|
work_result: str = ""
|
|
|
|
new_worker: Worker
|
|
|
|
class WorkerWidget(Widget):
|
|
async def work(self) -> str:
|
|
nonlocal work_result
|
|
await asyncio.sleep(0.02)
|
|
work_result = "foo"
|
|
return "foo"
|
|
|
|
def on_mount(self):
|
|
nonlocal new_worker
|
|
new_worker = self.run_worker(self.work, start=False)
|
|
|
|
def on_worker_state_changed(self, event) -> None:
|
|
worker_events.append(event)
|
|
|
|
class WorkerApp(App):
|
|
def compose(self) -> ComposeResult:
|
|
yield WorkerWidget()
|
|
|
|
app = WorkerApp()
|
|
async with app.run_test():
|
|
assert new_worker in app.workers
|
|
assert len(app.workers) == 1
|
|
app.workers.start_all()
|
|
await app.workers.wait_for_complete()
|
|
assert len(app.workers) == 0
|
|
|
|
assert work_result == "foo"
|
|
assert isinstance(worker_events[0].worker.node, WorkerWidget)
|
|
states = [event.state for event in worker_events]
|
|
assert states == [
|
|
WorkerState.PENDING,
|
|
WorkerState.RUNNING,
|
|
WorkerState.SUCCESS,
|
|
]
|
|
|
|
|
|
async def test_run_worker_thread() -> None:
|
|
"""Check self.run_worker"""
|
|
worker_events: list[Worker.StateChanged] = []
|
|
|
|
work_result: str = ""
|
|
|
|
class WorkerWidget(Widget):
|
|
def work(self) -> str:
|
|
nonlocal work_result
|
|
time.sleep(0.02)
|
|
work_result = "foo"
|
|
return "foo"
|
|
|
|
def on_mount(self):
|
|
self.run_worker(self.work)
|
|
|
|
def on_worker_state_changed(self, event) -> None:
|
|
worker_events.append(event)
|
|
|
|
class WorkerApp(App):
|
|
def compose(self) -> ComposeResult:
|
|
yield WorkerWidget()
|
|
|
|
app = WorkerApp()
|
|
async with app.run_test():
|
|
await app.workers.wait_for_complete()
|
|
|
|
assert work_result == "foo"
|
|
assert isinstance(worker_events[0].worker.node, WorkerWidget)
|
|
states = [event.state for event in worker_events]
|
|
assert states == [
|
|
WorkerState.PENDING,
|
|
WorkerState.RUNNING,
|
|
WorkerState.SUCCESS,
|
|
]
|