1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
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_non_async() -> 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, thread=True)
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,
]
async def test_run_worker_thread_async() -> None:
"""Check self.run_worker"""
worker_events: list[Worker.StateChanged] = []
work_result: str = ""
class WorkerWidget(Widget):
async 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, thread=True)
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,
]
|
