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
|
"""Settle/wait helpers shared by the pilot tests and the live RPC driver.
The single source of truth for "the UI has finished reacting to what I just did".
Both the headless test harness and the in-process RPC server need the exact same
guarantee before they read state back, so it lives here once.
Two yield strategies feed the same poll loop: under a Pilot (tests) we yield with
``pilot.pause`` (which also drains the screen); live (RPC) we yield with
``asyncio.sleep`` and drain explicitly via a throwaway ``Pilot(app)``.
"""
from __future__ import annotations
import asyncio
from typing import Awaitable, Callable, Optional
from textual.pilot import Pilot
async def wait_for(
pred: Callable[[], bool],
tick: Callable[[float], Awaitable[None]],
timeout: float = 20.0,
step: float = 0.02,
) -> bool:
"""Poll ``pred`` until true or ``timeout`` elapses.
``tick(step)`` yields control between polls: ``pilot.pause`` in tests,
``asyncio.sleep`` in the live app. Returns whether ``pred`` became true.
"""
waited = 0.0
while waited < timeout:
if pred():
return True
await tick(step)
waited += step
return False
async def drain(app, timeout: float = 15.0) -> None:
"""Wait for the message pump and every widget to process queued events.
Reuses Textual's own ``Pilot._wait_for_screen`` (it only needs ``app``), so
it works against a live app with no real Pilot attached.
"""
try:
await Pilot(app)._wait_for_screen(timeout=timeout)
except Exception: # noqa: BLE001 — best-effort; never let settle explode
pass
async def workers_idle(app, timeout: float = 20.0) -> None:
"""Wait (bounded) for all threaded ``@work`` workers to finish.
Safe because the app has no perpetual Textual workers — the keepalive is a
client-side thread, not a worker. A timeout guards against a wedged worker.
"""
try:
await asyncio.wait_for(app.workers.wait_for_complete(), timeout)
except Exception: # noqa: BLE001 — timeout or manager churn; fall through
pass
async def settle(
app,
pred: Optional[Callable[[], bool]] = None,
*,
timeout: float = 20.0,
step: float = 0.02,
rounds: int = 3,
) -> bool:
"""Block until the app is quiescent, then optionally until ``pred`` holds.
A round = drain the pump, wait for workers, repeat — because a completing
worker can post a message that spawns the next worker (nav -> decompile).
With a ``pred`` (the reliable signal, e.g. ``dec.loaded_ea == ea``) we return
as soon as it holds; without one we return once no workers remain.
"""
for _ in range(max(1, rounds)):
await drain(app, min(timeout, 15.0))
await workers_idle(app, timeout)
if pred is not None:
if pred():
await drain(app, min(timeout, 15.0))
return True
elif len(app.workers) == 0:
await drain(app, min(timeout, 15.0))
return True
await drain(app, min(timeout, 15.0))
if pred is not None:
return await wait_for(pred, asyncio.sleep, timeout, step)
return True
|
