diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/_fixtures.py | 51 | ||||
| -rw-r--r-- | tests/test_blob_ui.py | 16 | ||||
| -rw-r--r-- | tests/test_project_ui.py | 30 | ||||
| -rw-r--r-- | tests/test_scenarios.py | 157 | ||||
| -rw-r--r-- | tests/test_thumb_ui.py | 21 | ||||
| -rw-r--r-- | tests/test_trace_ui.py | 46 |
6 files changed, 261 insertions, 60 deletions
diff --git a/tests/_fixtures.py b/tests/_fixtures.py index 0b72856..16192fa 100644 --- a/tests/_fixtures.py +++ b/tests/_fixtures.py @@ -22,12 +22,63 @@ doesn't silently test the previous one. `.pristine.i64` is gitignored. """ from __future__ import annotations +import asyncio import contextlib import os import shutil import tempfile +def fast_keys() -> None: + """Make a simulated keypress cost ~2ms instead of ~85ms. Call before the app. + + **The problem.** Textual sends a key and then calls ``wait_for_idle`` + *twice*, and that helper sleeps in 20ms granules until *process* time stops + advancing -- a CPU-load heuristic standing in for "the state is predictable + now", which takes even more granules on a loaded box. Measured here: 84ms + per keypress, which was 23s of the pilot suite's 43s. + + **The fix, and why it is not just deletion.** Removing the heuristic alone + broke nine checks that read state straight after a keypress -- so it *was* + doing a job, badly. This replaces it with the real gate the rest of the + suite already uses: send the keys, then ``settle`` (message pump drained, + threaded workers finished). That is strictly stronger than "the CPU looks + idle", and it is ~2ms. + + **What it still cannot see:** anything driven by a TIMER rather than a + worker -- the function-list filter's 80ms debounce, and Textual's own frame + timer (so a widget's ``region``/``size`` is not laid out just because the + app settled). Those need a wait on the effect: ``wait(lambda: rows < full)``, + ``wait(lambda: inp.region.height >= 1)``. Every such site in this repo is + commented; if a check that reads geometry or a debounced view starts + flaking, that is the reason. + + Verified equivalent, not just faster: pressing j 40 times moves 40 rows with + and without the patch, and the full suite passes with identical counts. + """ + import textual.app + from textual.pilot import Pilot + + from idatui._sync import settle + + if not hasattr(textual.app, "wait_for_idle"): # pragma: no cover + raise RuntimeError( + "textual.app.wait_for_idle is gone -- tests/_fixtures.fast_keys " + "needs updating for this Textual version") + + async def _yield_instead_of_sleeping(min_sleep: float = 0.0, + max_sleep: float = 1.0) -> None: + await asyncio.sleep(0) + + async def _press(self, *keys: str) -> None: + if keys: + await self._app._press_keys(keys) + await settle(self._app, timeout=5.0) + + textual.app.wait_for_idle = _yield_instead_of_sleeping + Pilot.press = _press + + def cache_path(binary: str) -> str: return binary + ".pristine.i64" diff --git a/tests/test_blob_ui.py b/tests/test_blob_ui.py index 1055e5a..d4740d0 100644 --- a/tests/test_blob_ui.py +++ b/tests/test_blob_ui.py @@ -22,7 +22,9 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from textual.widgets import Input, Static # noqa: E402 from idatui.app import ConfirmScreen, IdaTui, ListingView # noqa: E402 -from _fixtures import staged, synthetic # noqa: E402 +from _fixtures import fast_keys, staged, synthetic # noqa: E402 + +fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys from idatui._sync import settle # noqa: E402 PASS = FAIL = 0 @@ -135,7 +137,7 @@ async def run() -> int: lst.focus() await pilot.press("down") await pilot.press("down") - await pilot.pause(0.3) + await settle(app) status2 = str(app.query_one("#status", Static).render()) check("the hint survives navigating", "no functions" in status2, status2[:90]) @@ -167,7 +169,7 @@ async def run() -> int: target = 0x4000 + PLANTED # a NOP we put there ourselves lst.cursor = m.index_of_ea(target) lst._scroll_cursor_into_view() - await pilot.pause(0.1) + await settle(app, lambda: lst._cursor_ea() == target) check("the cursor sits on the byte we aimed at", lst._cursor_ea() == target, f"{lst._cursor_ea():#x} want {target:#x}") @@ -219,7 +221,7 @@ async def run() -> int: # suite it can mutate the database freely. lst.cursor = m.index_of_ea(0x4200) lst._scroll_cursor_into_view() - await pilot.pause(0.4) + await settle(app, lambda: lst._cursor_ea() == 0x4200) ctop = lst.model.get(round(lst.scroll_offset.y)).ea ccur = lst._cursor_ea() old = lst.model @@ -254,7 +256,7 @@ async def run() -> int: far = 0x4000 + 0x600 lst.cursor = lst.model.index_of_ea(far) lst._scroll_cursor_into_view() - await pilot.pause(0.4) + await settle(app, lambda: lst._cursor_ea() == far) top_before = lst.model.get(round(lst.scroll_offset.y)).ea cur_before = lst._cursor_ea() check("scrolled somewhere with rows above us", @@ -286,7 +288,7 @@ async def run() -> int: f"n={len(app._func_index)}") lst.cursor = lst.model.index_of_ea(target) lst._scroll_cursor_into_view() - await pilot.pause(0.3) + await settle(app, lambda: lst._cursor_ea() == target) mp = lst.model await pilot.press("p") await wait(lambda: lst.model is not mp and lst.model is not None, @@ -313,7 +315,7 @@ async def run() -> int: "1 function," in note and "nothing is lost" not in note, note[:80]) await pilot.press("escape") - await pilot.pause(0.3) + await settle(app, lambda: not isinstance(app.screen, ConfirmScreen)) check("declining leaves the binary open", not isinstance(app.screen, ConfirmScreen) and app._cur is not None) diff --git a/tests/test_project_ui.py b/tests/test_project_ui.py index 7e93c81..863b98a 100644 --- a/tests/test_project_ui.py +++ b/tests/test_project_ui.py @@ -19,7 +19,11 @@ import sys import tempfile sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from idatui._sync import wait_for # noqa: E402 +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from _fixtures import fast_keys # noqa: E402 +from idatui._sync import settle as quiesce, wait_for # noqa: E402 + +fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys from idatui.app import IdaTui, ProjectPalette # noqa: E402 from idatui.project import Project # noqa: E402 from textual.widgets import Input, OptionList, Static # noqa: E402 @@ -113,7 +117,7 @@ async def run(bins): # -- switch to the second binary -------------------------------- # pal.query_one(Input).value = second - await pilot.pause(0.2) + await settle(lambda: bool(pal._results), 20) await pilot.press("enter") switched = await settle( lambda: app._binary == second and app.program is not None @@ -141,7 +145,7 @@ async def run(bins): if not reopened: return app.screen.query_one(Input).value = first - await pilot.pause(0.2) + await settle(lambda: bool(app.screen._results), 20) await pilot.press("enter") back = await settle(lambda: app._binary == first and app._func_index is not None @@ -160,7 +164,7 @@ async def run(bins): "switcher did not reopen") return app.screen.query_one(Input).value = second - await pilot.pause(0.2) + await settle(lambda: bool(app.screen._results), 20) await pilot.press("enter") again = await settle(lambda: app._binary == second and app._cur is not None, 120) @@ -181,15 +185,19 @@ async def run(bins): if await settle(lambda: isinstance(app.screen, SymbolPalette), 20): pal = app.screen pal.query_one(Input).value = "main" - await pilot.pause(0.3) + await settle(lambda: bool(pal._results), 20) + # The palette re-applies inside the key handler, so the gate is + # quiescence -- NOT "a foreign binary appeared", which is the + # thing under test and would sit out its whole timeout on the + # day it breaks. await pilot.press("f2") # widen to the whole project - await pilot.pause(0.4) + await quiesce(app) names = [(b, n) for b, _, n in pal._results] check("project scope finds a name shared by both binaries", len({b for b, n in names if n == "main"}) == 2, f"{names[:6]}") await pilot.press("escape") - await pilot.pause(0.2) + await settle(lambda: not isinstance(app.screen, SymbolPalette), 20) # -- a cross-binary jump is not a one-way door ----------------- # # Nav history is per-binary, so arriving in another binary lands you @@ -217,7 +225,7 @@ async def run(bins): if not app._hops or app._binary != there: break await pilot.press("escape") - await pilot.pause(0.6) + await quiesce(app) returned = await settle(lambda: app._binary == here, 180) check("Esc crosses back to the binary the jump came from", returned, f"binary={app._binary} want={here} hops={app._hops}") @@ -269,17 +277,17 @@ async def run(bins): if await settle(lambda: isinstance(app.screen, StringsPalette), 30): pal = app.screen pal.query_one(Input).value = "usage" - await pilot.pause(0.3) + await settle(lambda: bool(pal._results), 20) local = {b for b, _, _ in pal._results} await pilot.press("f2") - await pilot.pause(0.4) + await quiesce(app) wide = {b for b, _, _ in pal._results} check("strings: local scope is this binary only", local == {None}, f"{local}") check("strings: F2 widens across the project", len(wide) >= 2 and None not in wide, f"{wide}") await pilot.press("escape") - await pilot.pause(0.2) + await settle(lambda: not isinstance(app.screen, StringsPalette), 20) # -- the promise: nothing was written next to the sources ---------- # left = sorted(os.listdir(src)) diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py index d7c1f14..6c3177a 100644 --- a/tests/test_scenarios.py +++ b/tests/test_scenarios.py @@ -29,7 +29,9 @@ import traceback sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) -from _fixtures import staged # noqa: E402 +from _fixtures import fast_keys, staged # noqa: E402 + +fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys from idatui.app import ( # noqa: E402 ConfirmScreen, DecompView, FunctionsPanel, GraphView, HexView, IdaTui, HelpScreen, ListingView, QuitScreen, StringsPalette, StructEditor, @@ -40,13 +42,68 @@ from textual.widgets import ( # noqa: E402 DataTable, Input, OptionList, Static, TextArea, ) from rich.text import Text # noqa: E402 -from idatui._sync import wait_for # noqa: E402 +from idatui._sync import settle, wait_for # noqa: E402 PASS = FAIL = 0 STOP_AFTER = None SCENARIOS: list[tuple[str, object]] = [] +class _Profile: + """Where the suite's wall clock goes, per scenario. + + Two numbers matter and neither is visible from a scenario's total: seconds + spent in FIXED pauses (a guess about a state we could have observed), and + waits that ran out their timeout -- those cost the full timeout AND let the + check after them pass vacuously. ``--profile`` prints both. + """ + + def __init__(self) -> None: + self.enabled = False + self.paused: dict[str, float] = {} + self.waited: dict[str, float] = {} + self.pressed: dict[str, float] = {} + self.expired: list[tuple[str, float]] = [] + + def pause(self, scenario: str, secs: float) -> None: + if self.enabled: + self.paused[scenario] = self.paused.get(scenario, 0.0) + secs + + def press(self, scenario: str, secs: float) -> None: + if self.enabled: + self.pressed[scenario] = self.pressed.get(scenario, 0.0) + secs + + def wait(self, scenario: str, secs: float, ok: bool, line: int = 0) -> None: + if not self.enabled: + return + self.waited[scenario] = self.waited.get(scenario, 0.0) + secs + if not ok: + self.expired.append((f"{scenario} (line {line})", secs)) + + def report(self) -> None: + if not self.enabled: + return + tp, tw = sum(self.paused.values()), sum(self.waited.values()) + tk = sum(self.pressed.values()) + print(f"\nprofile: {tp:.1f}s settling, {tw:.1f}s in waits, " + f"{tk:.1f}s in keystrokes") + rows = sorted(self.paused.items(), key=lambda kv: -kv[1])[:8] + for name, secs in rows: + print(f" pause {secs:5.2f}s {name}") + rows = sorted(self.waited.items(), key=lambda kv: -kv[1])[:8] + for name, secs in rows: + print(f" wait {secs:5.2f}s {name}") + rows = sorted(self.pressed.items(), key=lambda kv: -kv[1])[:8] + for name, secs in rows: + print(f" keys {secs:5.2f}s {name}") + for name, secs in self.expired: + print(f" EXPIRED wait {secs:5.2f}s in {name} " + f"(the check after it may have passed vacuously)") + + +PROFILE = _Profile() + + class _StopSuite(Exception): pass @@ -89,14 +146,52 @@ class Ctx: raise _StopSuite async def wait(self, pred, t=20.0, step=0.02): - return await wait_for(pred, self.pilot.pause, t, step) + t0 = asyncio.get_event_loop().time() + ok = await wait_for(pred, self.pilot.pause, t, step) + PROFILE.wait(self.scenario, asyncio.get_event_loop().time() - t0, ok, + sys._getframe(1).f_lineno) + return ok async def press(self, *keys): + """Send keys, then wait for the app to finish reacting to them. + + The wait is ours, deliberately. Textual's own ``press`` ends with two + ``wait_for_idle`` sleeps per key (~85ms), which is a CPU-load heuristic + standing in for a gate -- and scenarios came to lean on it, so removing + it alone broke nine checks that read state straight after a keypress. + ``settle`` is the real thing (pump drained, workers finished) at ~2ms, + and it holds on a loaded box where the heuristic is exactly as likely to + return early. + """ + t0 = asyncio.get_event_loop().time() for k in keys: await self.pilot.press(k) + await settle(self.app, timeout=5.0) + PROFILE.press(self.scenario, asyncio.get_event_loop().time() - t0) async def pause(self, d=0.05): + """Yield until the app has finished reacting to what we just did. + + This used to be a flat ``asyncio.sleep(d)``, and across ~140 call sites + that was 20s of the suite's 62s spent asleep -- a guess about a state we + can observe directly. ``settle`` drains the message pump and waits for + the threaded workers, so it returns the moment the app is quiescent + (single-digit ms in the common case) and, unlike a sleep, it does not + silently pass when the machine is loaded and the work took longer than + the guess. ``d`` survives as the upper BOUND, not as the cost. + + Use :meth:`sleep` for the rare thing that is genuinely gated on a timer. + """ + t0 = asyncio.get_event_loop().time() + await settle(self.app, timeout=max(d, 2.0)) + PROFILE.pause(self.scenario, asyncio.get_event_loop().time() - t0) + + async def sleep(self, d): + """A real wall-clock sleep -- only for what a TIMER drives (throttles, + debounces, blink), where there is no worker to wait for.""" + t0 = asyncio.get_event_loop().time() await self.pilot.pause(d) + PROFILE.pause(self.scenario, asyncio.get_event_loop().time() - t0) async def type(self, text): for ch in text: @@ -1390,7 +1485,9 @@ async def s_search(c: Ctx): si = app.query_one("#search", Input) status = app.query_one("#status", Static) await c.press("slash") - await c.pause(0.1) + # display flips synchronously; the REGION only exists once Textual has laid + # the prompt out, which is a frame, not a worker. + await c.wait(lambda: si.display and si.region.height >= 1, 5) c.check("search bar visible, status hidden (no overlap)", si.display and not status.display, f"si={si.display} status={status.display}") c.check("search input owns the bottom row (nothing overlaps it)", @@ -1416,22 +1513,23 @@ async def s_incr_filter(c: Ctx): table.focus() full = table.row_count await c.press("slash") - await c.pause(0.1) for ch in "sub_": await c.press(ch) - await c.pause(0.05) - await c.pause(0.1) + # The filter is DEBOUNCED (set_timer(0.08)) -- a timer, not a worker, so + # settling can't see it. Wait for the effect instead of guessing at the + # debounce: it returns the moment the rows are rebuilt. + await c.wait(lambda: 0 < table.row_count < full, 5) c.check("filter narrows incrementally as you type", 0 < table.row_count < full, f"{table.row_count}/{full}") cell = table.get_row_at(0)[1] c.check("filter highlights matched substring in name", isinstance(cell, Text) and any(s.style for s in cell.spans), repr(str(cell))) await c.press("enter") - await c.pause(0.1) + await c.wait(lambda: isinstance(app.focused, DataTable), 5) c.check("Enter keeps filter + focuses table", isinstance(app.focused, DataTable) and table.row_count < full) await c.press("escape") - await c.pause(0.15) + await c.wait(lambda: table.row_count == full, 5) c.check("Esc on the list clears the filter", table.row_count == full, f"{table.row_count}/{full}") @@ -1811,8 +1909,12 @@ async def s_rename(c: Ctx): f"display={ci.display}") ci.value = cnote await c.press("enter") - await c.wait(lambda: dec.loaded_ea == app._cur.ea - and any(cnote in t for t in dec._texts), 25) + # Gate on the comment showing up, and ONLY that: the extra + # `dec.loaded_ea == app._cur.ea` conjunct this used to carry is not a + # signal the re-decompile ever sets, so on some orderings the wait sat + # out its full 25s (9s of wall clock) and then the check below passed + # vacuously anyway. + await c.wait(lambda: any(cnote in t for t in dec._texts), 25) c.check("comment appears in the pseudocode after ';'", any(cnote in t for t in dec._texts), "comment not shown") app.program.client.invoke("set_comments", items=[{"addr": hex(cea), "comment": ""}]) @@ -2021,7 +2123,10 @@ async def s_scroll_restore(c: Ctx): await c.press("escape") await c.wait(lambda: app._cur.ea == fa.addr, 20) await c.wait(lambda: dis.total > 40, 20) - await c.pause(0.25) + # A repaint is driven by Textual's frame timer, so settling does not imply + # one happened. Wait for the paint we are actually asserting about (each + # poll ticks the screen, so this is ~one frame, not a quarter second). + await c.wait(lambda: bool(renders) and renders[-1] == want_sy, 5) c.check("disasm scroll + cursor restored on back (mid-viewport)", round(dis.scroll_offset.y) == want_sy and dis.cursor == want_cur and want_rel > 0, f"scroll={round(dis.scroll_offset.y)} (want {want_sy}) " @@ -2202,8 +2307,12 @@ async def s_listing_view(c: Ctx): return await c.goto_ui(hex(data_ea)) + # `total > 0` is set from the segment's size before a single page has + # materialised, so waiting on it and then reading rows was the suite's + # one known flake (it failed roughly one run in three). Wait for a ROW. await c.wait(lambda: app._cur is not None and app._active == "listing" - and c.lst.total > 0, 25) + and c.lst.total > 0 and c.lst.model is not None + and any(h.kind == "data" for h in c.lst.model.window(0, 40)), 25) c.check("navigating to a data segment opens the listing view", app._active == "listing" and c.lst.display and c.lst.total > 0, f"active={app._active} total={c.lst.total}") @@ -3110,8 +3219,14 @@ async def s_graph_render(c: Ctx): if gv.lay is None: c.check("graph loaded", False) return - rows = [gv.render_line(y).text for y in range(gv.size.height)] - blob = "\n".join(rows) + # The layout being ready (`gv.lay`) is not the same as the view having a + # SIZE to render into -- that needs a laid-out frame, and reading glyphs + # before one lands scrapes an empty canvas. Gate on the paint itself. + def _blob(): + return "\n".join(gv.render_line(y).text for y in range(gv.size.height)) + + await c.wait(lambda: gv.size.height > 0 and "\u250c" in _blob(), 10) + blob = _blob() c.check("boxes are drawn", blob.count("\u250c") >= 1 and blob.count("\u2502") > 4, f"corners={blob.count(chr(0x250c))} verts={blob.count(chr(0x2502))}") c.check("edges are drawn", any(ch in blob for ch in "\u25bc\u2570\u256d\u256e\u256f"), @@ -3126,11 +3241,11 @@ async def s_graph_render(c: Ctx): # minimap on/off actually changes the picture before = blob await c.press("m") - await c.pause(0.1) - after = "\n".join(gv.render_line(y).text for y in range(gv.size.height)) + await c.wait(lambda: _blob() != before, 5) + after = _blob() c.check("m toggles the minimap", after != before and not gv._show_minimap) await c.press("m") - await c.pause(0.1) + await c.wait(lambda: gv._show_minimap, 5) c.check("and toggles it back", gv._show_minimap) # a row query must never paint inside a box (that is what dummies buy us) bad = 0 @@ -3180,6 +3295,9 @@ async def s_graph_minimap(c: Ctx): if gv.lay is None: c.check("graph loaded", False) return + # The minimap's rect is derived from the view's SIZE, so it needs a laid-out + # frame -- not just a settled app. + await c.wait(lambda: gv._minimap_rect() is not None, 5) rect = gv._minimap_rect() c.check("the minimap has a hit-box while it's shown", rect is not None, f"size={gv.size} shown={gv._show_minimap}") @@ -3392,6 +3510,8 @@ def main(argv): STOP_AFTER = next(it) elif a in ("--worker", "--binary"): binary = os.path.abspath(os.path.expanduser(next(it))) + elif a == "--profile": + PROFILE.enabled = True elif a == "--list": for name, _ in SCENARIOS: print(name) @@ -3406,6 +3526,7 @@ def main(argv): asyncio.run(run(binary, only)) except _StopSuite: print(f" … stopped after '{STOP_AFTER}'") + PROFILE.report() print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0 diff --git a/tests/test_thumb_ui.py b/tests/test_thumb_ui.py index fed08a7..f12fbc4 100644 --- a/tests/test_thumb_ui.py +++ b/tests/test_thumb_ui.py @@ -20,10 +20,14 @@ import sys import tempfile sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from textual.widgets import Static # noqa: E402 +from _fixtures import fast_keys # noqa: E402 from idatui._sync import settle # noqa: E402 + +fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys from idatui.app import DecompView, IdaTui, ListingView # noqa: E402 PASS = FAIL = 0 @@ -95,7 +99,7 @@ async def run() -> int: lst.focus() lst.cursor = lst.model.index_of_ea(0) lst._scroll_cursor_into_view() - await pilot.pause(0.3) + await settle(app) check("starts undefined at the entry", lst.model.get(lst.cursor).kind == "unknown", f"{lst.model.get(lst.cursor).text!r}") @@ -104,7 +108,10 @@ async def run() -> int: # the Thumb prologue. m0 = lst.model await pilot.press("c") - await pilot.pause(2.5) + # A refusal produces no new signal to wait for, so the gate is "the app + # finished reacting" -- 2.5s of hoping bought nothing a drained worker + # queue doesn't say better. + await settle(app) h = lst.model.get(lst.model.index_of_ea(0)) check("`c` alone does not produce the Thumb prologue", h is None or h.kind != "code" or "PUSH" not in h.text.upper(), @@ -167,7 +174,7 @@ async def run() -> int: lst.focus() lst.cursor = lst.model.index_of_ea(0) lst._scroll_cursor_into_view() - await pilot.pause(0.3) + await settle(app) m = lst.model await pilot.press("t") await settle(app, lambda: "64-bit" in status_of(app), timeout=60) @@ -181,7 +188,7 @@ async def run() -> int: # the reason, which is the only part that tells you what to do — with it # missing, F5 doing nothing is indistinguishable from a bug in the TUI. lst.cursor = lst.model.index_of_ea(0) - await pilot.pause(0.2) + await settle(app) mp = lst.model await pilot.press("p") # The function appearing in the index IS the signal; the model identity @@ -214,7 +221,7 @@ async def run() -> int: check("a 32-bit ARM database finds functions by itself", len(app._func_index) > 5, f"n={len(app._func_index)}") await pilot.press("escape") - await pilot.pause(0.5) + await settle(app, lambda: type(app.screen).__name__ == "Screen") f = app._func_index.all_loaded()[0] app._goto_ea(f.addr, push=True) await wait(lambda: app._cur is not None @@ -248,14 +255,14 @@ async def run() -> int: len(app._func_index) == 0, f"n={len(app._func_index)}") if type(app.screen).__name__ != "Screen": await pilot.press("escape") - await pilot.pause(0.5) + await settle(app, lambda: type(app.screen).__name__ == "Screen") app._goto_ea(0, push=True) lst = app.query_one(ListingView) await wait(lambda: lst.model is not None, pilot, 60) lst.focus() lst.cursor = lst.model.index_of_ea(0) lst._scroll_cursor_into_view() - await pilot.pause(0.3) + await settle(app) await pilot.press("T") await wait(lambda: app._func_index is not None and len(app._func_index) >= 3, pilot, 90) diff --git a/tests/test_trace_ui.py b/tests/test_trace_ui.py index 5844c90..855ce4d 100644 --- a/tests/test_trace_ui.py +++ b/tests/test_trace_ui.py @@ -19,7 +19,9 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from textual.widgets import Input, OptionList, Static # noqa: E402 -from _fixtures import staged # noqa: E402 +from _fixtures import fast_keys, staged # noqa: E402 + +fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys from idatui._sync import settle # noqa: E402 from idatui.app import (DecompView, IdaTui, ListingView, # noqa: E402 RegWriteScreen, TraceDock) @@ -108,7 +110,7 @@ async def run() -> int: # -- stepping --------------------------------------------------- # lst = app.query_one(ListingView) lst.focus() - await pilot.pause(0.4) + await settle(app) await pilot.press("]") # `app._t` is assigned the moment the key is handled, so it is NOT a # signal that the VIEW has followed -- the navigation it kicks off @@ -126,7 +128,10 @@ async def run() -> int: timeout=20) check("[ steps backward", app._t == 0, f"t={app._t}") await pilot.press("[") - await pilot.pause(0.4) + # Nothing should happen, so there is no signal to wait FOR: the + # honest gate is "the app finished reacting" (workers drained), not + # a sleep long enough that a bug would have shown by now. + await settle(app) check("and stops at the start of the trace", app._t == 0) # -- step over -------------------------------------------------- # @@ -252,12 +257,14 @@ async def run() -> int: # opens the LISTING unless the decompiler is preferred — so # stepping through C used to drop you into disassembly on the # first keypress. Found by watching a demo, not by a test. + was = app._t await pilot.press("]") - await pilot.pause(1.2) + await settle(app, lambda: app._t != was) check("stepping in pseudocode stays in pseudocode", app._active == "decomp", f"active={app._active}") + was = app._t await pilot.press("[") - await pilot.pause(1.2) + await settle(app, lambda: app._t != was) check("and so does stepping backward", app._active == "decomp", f"active={app._active}") @@ -336,7 +343,10 @@ async def run() -> int: app._seek(s0) await wait(lambda: lst._cursor_ea() == dbaddr, pilot, 60) app._seek(s1) - await pilot.pause(3.0) # long enough for a stale one to land + # The stale navigation this guards against is a WORKER, so wait + # for the workers to drain rather than for three seconds and a + # hope: same question, ~50ms instead of 3s. + await settle(app) check("a stale navigation doesn't drag the cursor away", lst._cursor_ea() == dbaddr and app._cur.ea == dbaddr, f"cursor={lst._cursor_ea():#x} cur={app._cur.ea:#x} " @@ -354,47 +364,49 @@ async def run() -> int: else: if app._split: app.action_toggle_split() - await pilot.pause(1.0) + await settle(app, lambda: not app._split) if app._active != "listing": # focus() does NOT make a view active outside split mode; # Tab is what switches which one is showing. await pilot.press("tab") await wait(lambda: app._active == "listing", pilot, 60) app._seek(stamps[0]) - await pilot.pause(1.2) + await settle(app, lambda: app._t == stamps[0]) lst.focus() row = lst.model.index_of_ea(db) lst.cursor = row lst._scroll_cursor_into_view() - await pilot.pause(0.4) + await settle(app, lambda: lst._cursor_ea() == db) check("cursor is on the repeated instruction", lst._cursor_ea() == db, f"{lst._cursor_ea():#x} vs {db:#x}") await pilot.press(">") - await pilot.pause(1.0) + await settle(app, lambda: app._t == stamps[1]) check("> seeks to the next execution of it", app._t == stamps[1], f"t={app._t}, expected {stamps[1]}") status = str(app.query_one("#status", Static).render()) check("and says which execution this is", f"2 of {len(stamps)}" in status, status[:80]) lst.cursor = row - await pilot.pause(0.2) + await settle(app) await pilot.press("<") - await pilot.pause(1.0) + await settle(app, lambda: app._t == stamps[0]) check("< seeks back to the previous one", app._t == stamps[0], f"t={app._t}, expected {stamps[0]}") # An edge must SAY it's an edge rather than silently doing # nothing, which is indistinguishable from a broken key. lst.cursor = row - await pilot.pause(0.2) + await settle(app) await pilot.press("<") - await pilot.pause(1.0) + await settle(app, lambda: "first" in str( + app.query_one("#status", Static).render())) status = str(app.query_one("#status", Static).render()) check("and the first execution says so instead of moving", app._t == stamps[0] and "first" in status, status[:80]) # -- "which instruction set this register?" ---------------------- # - app._seek(min(200, t.length - 1)) - await pilot.pause(1.0) + want_t = min(200, t.length - 1) + app._seek(want_t) + await settle(app, lambda: app._t == want_t) lst.focus() await pilot.press("W") opened = await wait(lambda: isinstance(app.screen, RegWriteScreen), @@ -411,7 +423,7 @@ async def run() -> int: else: name, _v, last, _n = sc._rows[pick] sc.query_one(OptionList).highlighted = pick - await pilot.pause(0.3) + await settle(app) await pilot.press("enter") await wait(lambda: app._t == last, pilot, 30) check("choosing one seeks to the write that set it", |