diff options
Diffstat (limited to 'tests/test_scenarios.py')
| -rw-r--r-- | tests/test_scenarios.py | 1562 |
1 files changed, 1498 insertions, 64 deletions
diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py index 09e1ac5..5297f0c 100644 --- a/tests/test_scenarios.py +++ b/tests/test_scenarios.py @@ -16,32 +16,95 @@ Uses ~/ida-venv python (has textual). --stop-after <substr> stop once a check whose name contains <substr> ran --list print scenario names and exit """ + +#: the pilot suite: one real worker, 56 scenarios. +#: Read by tests/run.py (--fast skips every NEEDS_IDA file). +NEEDS_IDA = True import asyncio import fnmatch import os import re -import shutil import sys -import tempfile 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 fast_keys, staged # noqa: E402 + +fast_keys() # ~85ms -> ~2ms per keypress; see _fixtures.fast_keys from idatui.app import ( # noqa: E402 - ConfirmScreen, DecompView, FunctionsPanel, HexView, IdaTui, - HelpScreen, ListingView, QuitScreen, StringsPalette, StructEditor, - SymbolPalette, XrefsScreen, _str_display, _word_occurrences, + ConfirmScreen, DecompView, FunctionsPanel, GraphView, HexView, IdaTui, + HelpScreen, ListingView, QuitScreen, SearchPalette, StringsPalette, + StructEditor, SymbolPalette, XrefsScreen, _HELP, _str_display, + _word_occurrences, ) +from idatui.errors import IDAToolError # noqa: E402 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 @@ -56,6 +119,12 @@ def scenario(name): # --------------------------------------------------------------------------- # # Shared context + helpers # --------------------------------------------------------------------------- # +#: What the app says when Hex-Rays can't decompile (idatui/app.py, +#: _apply_decomp). Waiting on the wrong text here doesn't fail a test -- it +#: times out and then lets a weaker check pass, which is far more expensive. +_CANNOT_DECOMP = "cannot decompile" + + class Ctx: def __init__(self, app, pilot): self.app = app @@ -78,14 +147,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: @@ -121,9 +228,23 @@ class Ctx: # -- discovery -------------------------------------------------------- # def all_funcs(self): + """Every function, always -- never the prefix that happens to be loaded. + + This used to load_all() only when the index was EMPTY, so a partially + streamed index (non-empty but incomplete: exactly the state during boot, + and after any bump_items()) came back truncated. Every fixture chosen + through find_func/biggest then depended on how far streaming had got, + which is a race. + + It bit graph_minimap: on an unlucky run `find_func(size > 0x300)` picked + a far larger function than usual, whose graph never finished inside the + scenario's 60s wait -- 3 failures and 65s, one run in several, with no + code change to blame. Deterministic fixtures or deterministic flakes, + pick one. + """ idx = self.prog.functions() - if len(idx) == 0 and not idx.complete: - idx.load_all() # a prior bump_items() cleared the index cache + if not idx.complete: + idx.load_all() # boot streaming, or a prior bump_items() return idx.all_loaded() def find_func(self, pred, limit=400): @@ -236,8 +357,9 @@ class Ctx: if left.display: left.display = False app._pref = "decomp" - if app._active == "hex": + if app._active in ("hex", "graph"): app._active = "decomp" + app._graph_sticky = False # else every later scenario rebuilds a graph app._split = False await self.pause(0.02) @@ -444,6 +566,37 @@ async def s_asm_highlight(c: Ctx): mn is not None, f"{code[0].spans if code else None}") +@scenario("status_names_the_file") +async def s_status_names_the_file(c: Ctx): + """The status bar always says which file you're looking at. + + Obvious once several panes and a project switcher exist: with two idatui + windows open, or after switching binaries, "0x2490" alone doesn't say what + it belongs to. + """ + from textual.widgets import Static + app = c.app + await c.open_biggest("listing") + await c.pause(0.3) + name = os.path.basename(app._open_path) + status = str(app.query_one("#status", Static).render()) + c.check("the status bar names the open file", + status.startswith(f"[{name}]"), f"{status[:60]!r} (want [{name}])") + + # It must survive the messages that WRITE the status, not just the idle one. + c.lst.focus() + await c.press("down") + await c.pause(0.3) + status = str(app.query_one("#status", Static).render()) + c.check("and keeps naming it as you move", + status.startswith(f"[{name}]"), status[:60]) + + # The function-count message used to include the module name itself, which + # would now read "[echo] echo — 128 functions". + c.check("without saying the name twice", + status.count(name) == 1, status[:70]) + + @scenario("command_palette") async def s_command_palette(c: Ctx): app = c.app @@ -506,10 +659,14 @@ async def s_help(c: Ctx): cards = app.screen.query(".help-card") titles = {str(w.border_title) for w in cards} txt = " ".join(str(w.render()) for w in cards) + # Derived from _HELP, not hardcoded: adding a group is a normal change and + # shouldn't fail a test that only meant 'every group is rendered'. c.check("each key group gets its own card", - titles == {"Navigate", "Views", "Move", "Edit", "Search"}, f"{titles}") + titles == {t for t, _ in _HELP}, f"{titles}") c.check("it documents real bindings", "set type" in txt and "split view" in txt and "cross-references" in txt) + c.check("the graph keys are documented", + "control-flow graph" in txt and "minimap" in txt) body = app.screen.query_one("#help-body") c.check("the cards fit without a scrollbar at a normal size", body.virtual_size.height <= body.size.height, @@ -517,6 +674,16 @@ async def s_help(c: Ctx): await c.press("escape") await c.wait(lambda: not isinstance(app.screen, HelpScreen), 10) c.check("Esc closes it", not isinstance(app.screen, HelpScreen)) + # F-keys get eaten by terminals/multiplexers upstream of us, so the + # cheatsheet must not be reachable ONLY through F1. + await c.press("H") + opened_h = await c.wait(lambda: isinstance(app.screen, HelpScreen), 10) + c.check("H opens the cheatsheet too", opened_h, + f"screen={type(app.screen).__name__}") + if opened_h: + await c.press("H") + await c.wait(lambda: not isinstance(app.screen, HelpScreen), 10) + c.check("H closes it again", not isinstance(app.screen, HelpScreen)) @scenario("strings") @@ -561,9 +728,85 @@ async def s_strings(c: Ctx): await c.press("escape") +@scenario("view_modes_all_handled") +async def s_view_modes_all_handled(c: Ctx): + """Every ViewMode must be handled by every switch that reads _active. + + Adding "graph" meant auditing each `_active ==` in the app, and the one that + was missed -- _active_code_view returning None -- crashed the app the first + time a prompt closed in graph mode. There used to be a fifth value, + "disasm", assigned on one path and understood by four sites out of nine. + + So: walk the enum, and for each member show it and ask the app the questions + it asks itself. Cheap (no worker calls, just mode switches) and it fails on + the next mode that forgets to appear somewhere. + """ + from idatui.app import ViewMode + app = c.app + fn = await c.open_biggest("listing") + try: + for mode in ViewMode: + app._active = mode + app._show_active() # must not raise for any member + await c.pause(0.05) + view = app._active_code_view() + if mode in ViewMode.code_modes(): + c.check(f"{mode.value}: _active_code_view resolves a widget", + view is not None, f"{mode.value} -> None") + c.check(f"{mode.value}: exactly one predicate is true", + sum((app.is_listing, app.is_decomp, + app.is_hex, app.is_graph)) == 1, + f"{mode.value}: listing={app.is_listing} " + f"decomp={app.is_decomp} hex={app.is_hex} graph={app.is_graph}") + c.check(f"{mode.value}: in_code agrees with code_modes()", + app.in_code == (mode in ViewMode.code_modes()), + f"in_code={app.in_code} for {mode.value}") + c.check("every mode is a plain string over the wire", + all(isinstance(m, str) and m == m.value for m in ViewMode), + str([repr(m) for m in ViewMode])) + c.check("'disasm' is not a mode any more", + "disasm" not in {m.value for m in ViewMode}, + str([m.value for m in ViewMode])) + finally: + # Restore through a real navigation, not by poking _active back. + # _show_active() tears down split state and re-points the panes as a + # side effect, and reset() doesn't rebuild any of that -- leaving it + # half-torn-down made split_view fail two scenarios later with an empty + # listing model, which reads as split_view's bug and isn't. + app._active = ViewMode.LISTING + await c.open(fn.addr, "listing") + + @scenario("split_view") async def s_split_view(c: Ctx): app, lst, dec = c.app, c.lst, c.dec + # Count worker lookups across this whole scenario. _sync_split used to bounce + # off _apply_resync and back for as long as the decomp map lagged the + # decompiler -- one thread worker and one lookup_funcs round trip per + # iteration, 23,665 of them in this scenario alone (four distinct + # addresses; 21,156 for one of them), and an idle split view pegging the + # worker in the live app. The race window is real work to reproduce + # deliberately, but it is wide open in the flow below, so the cheap guard is + # to count. The bound is loose because the bug was three orders of magnitude + # out, not a near miss. + _lookups = {"n": 0} + _orig_call = c.prog.client.invoke + + def _counting(name, *a, **kw): + if name == "lookup_funcs": + _lookups["n"] += 1 + return _orig_call(name, *a, **kw) + + c.prog.client.invoke = _counting + try: + await _split_view_body(c, app, lst, dec) + finally: + c.prog.client.invoke = _orig_call + c.check("split view doesn't storm the worker with function lookups", + _lookups["n"] < 500, f"{_lookups['n']} lookup_funcs calls") + + +async def _split_view_body(c: Ctx, app, lst, dec): await c.open_biggest("listing") await c.press("s") shown = await c.wait(lambda: app._split and lst.display and dec.display, 20) @@ -734,10 +977,16 @@ async def s_fallback(c: Ctx): await c.open(failing.addr, "listing") c.dis.focus() await c.press("tab") - await c.wait(lambda: app._active == "listing" - and "fail" in c.status().lower(), 25) - c.check("F5/Tab on an undecompilable function falls back to the listing", - app._active == "listing" and c.dis.display, + # "cannot decompile" is what _apply_decomp actually says. This waited on + # "fail", which never appears, so it burned the full 25s timeout and the + # check below then passed on _active == "listing" -- already true before Tab + # was pressed, since the function was opened in the listing. It asserted + # nothing, slowly. + landed = await c.wait(lambda: _CANNOT_DECOMP in c.status().lower(), 25) + c.check("F5/Tab on an undecompilable function says so", landed, + f"active={app._active} status={c.status()!r}") + c.check("F5/Tab on an undecompilable function falls back to a code view", + app.is_listing and c.dis.display, f"active={app._active} status={c.status()!r}") # a decompilable function F5s into pseudocode await c.open("main", "decomp") @@ -767,6 +1016,15 @@ async def s_structs(c: Ctx): await c.wait(lambda: tname in ta.text and "{" in ta.text, 15) c.check("selecting a struct shows its C definition", tname in ta.text and "{" in ta.text, f"text={ta.text[:40]!r}") + # The definition is C, so it must be coloured as C (no tree-sitter grammar + # for it: idatui.highlight fills TextArea's highlight map from Pygments). + names = {n for spans in ta._highlights.values() for _, _, n in spans} + c.check("the C definition is syntax-highlighted", + {"keyword", "name"} <= names, f"names={sorted(names)}") + styled = {s.style.color.name for s in ta.render_line(0) + if s.style and s.style.color} + c.check("highlight styles reach the rendered line", len(styled) > 1, + f"colors={sorted(styled)}") app._clipboard = "" se.query_one(TextArea).focus() await c.press("ctrl+y") @@ -783,6 +1041,10 @@ async def s_structs(c: Ctx): c.check("Ctrl+S declares a new struct", any(s.name == sname for s in se._structs), "not created") await c.wait(lambda: "\n" in ta.text, 10) + c.check("editing re-highlights the definition", + any(n == "keyword" for spans in ta._highlights.values() + for _, _, n in spans), + f"rows={len(ta._highlights)}") c.check("save auto-formats the definition in the editor", ta.text.count("\n") >= 3 and f"struct {sname}" in ta.text and not se._is_dirty(), f"text={ta.text[:50]!r}") @@ -832,6 +1094,353 @@ async def s_structs(c: Ctx): c.check("Esc closes the struct editor", not isinstance(app.screen, StructEditor)) +@scenario("splash_scaling") +async def s_splash_scaling(c: Ctx): + """The splash scales the logo to the pane instead of dropping it. + + The bug this pins: the artwork's natural size is ~31 rows plus 10 of box + chrome, and the check was "do you have 41 rows?". A 31-row pane -- what a + split zellij window actually gives you -- was one row short, so the logo + silently disappeared. The terminal scales an image into whatever cell box + it is placed in, so there was never a reason for all-or-nothing. + """ + from idatui import kittygfx + from idatui.app import (LOGO_CHROME_ROWS, LOGO_MIN_ROWS, LoadingScreen, + logo_cells) + + app = c.app + placed: list[tuple] = [] + real_supported, real_upload, real_place = ( + kittygfx.supported, kittygfx.upload, kittygfx.place) + kittygfx.supported = lambda: True + kittygfx.upload = lambda *a, **k: True + kittygfx.place = lambda *a, **k: (placed.append(a), True)[1] + try: + for width, height in ((159, 31), (100, 30), (140, 44)): + await c.pilot.resize_terminal(width, height) + await c.pause(0.05) + app.push_screen(LoadingScreen("echo")) + await c.wait(lambda: isinstance(app.screen, LoadingScreen), 5) + scr = app.screen + # push_screen returns before compose has mounted the children. + await c.wait(lambda: scr._cells is not None, 5) + room = height - LOGO_CHROME_ROWS + has_image = bool(scr.query("#loading-image")) + c.check(f"{width}x{height}: the logo is drawn, not dropped", + has_image and room >= LOGO_MIN_ROWS, + f"image={has_image} room={room}") + if has_image: + cols, rows = scr._cells + c.check(f"{width}x{height}: scaled to the room available", + rows <= room and rows == min(room, logo_cells()[1]), + f"cells={scr._cells} room={room} natural={logo_cells()}") + await c.wait(lambda: scr.query_one("#loading-box").region.height > 0, 5) + box = scr.query_one("#loading-box").region + c.check(f"{width}x{height}: the box is not clipped", + box.y >= 0 and box.y + box.height <= height, + f"box={box} screen={height}") + app.pop_screen() + await c.pause(0.05) + c.check("a full-size pane still gets the artwork's natural size", + logo_cells(999) == logo_cells(), f"{logo_cells(999)}") + c.check("and the image was actually placed each time", len(placed) >= 3, + f"{placed}") + finally: + kittygfx.supported, kittygfx.upload, kittygfx.place = ( + real_supported, real_upload, real_place) + # Every later scenario assumes the suite's own geometry. + await c.pilot.resize_terminal(140, 44) + await c.pause(0.05) + + +@scenario("modal_centering") +async def s_modal_centering(c: Ctx): + """Every dialog we define is centred, without anyone maintaining a list. + + The CSS used to name the screens that centre, so a new palette shipped + pinned to the top of the screen (twice). The rule is on ModalScreen now; + this fails if a modal ever opts out by accident, which the naked eye only + catches when the dialog is already in front of a user. + """ + from textual.screen import ModalScreen + + import idatui.app as A + + ours = sorted( + (n for n, v in vars(A).items() + if isinstance(v, type) and issubclass(v, ModalScreen) + and v is not ModalScreen and v.__module__ == A.__name__), + key=str) + c.check("found the app's modal screens", len(ours) >= 8, f"{ours}") + styles = A.IdaTui.CSS + c.check("centring is a rule about modals, not a list of them", + "ModalScreen { align: center middle; }" in styles, + "the ModalScreen rule is gone") + # And prove it REACHES a dialog, rather than just being present in the text. + await c.press("ctrl+f") + opened = await c.wait(lambda: isinstance(c.app.screen, A.SearchPalette), 10) + if not opened: + c.check("the search palette opened", False, + f"screen={type(c.app.screen).__name__}") + return + scr = c.app.screen + await c.wait(lambda: scr.query_one("#pal-box").region.height > 0, 5) + box = scr.query_one("#pal-box").region + above, below = box.y, c.app.size.height - (box.y + box.height) + c.check("the search palette is vertically centred", + box.height > 0 and abs(above - below) <= 1, + f"box={box} screen={c.app.size} above={above} below={below}") + left = box.x + right = c.app.size.width - (box.x + box.width) + c.check("and horizontally centred", abs(left - right) <= 1, + f"left={left} right={right}") + await c.press("escape") + await c.wait(lambda: not isinstance(c.app.screen, A.SearchPalette), 5) + + +@scenario("db_search") +async def s_db_search(c: Ctx): + """Ctrl+F: search the whole database, by disassembly text or by bytes.""" + app = c.app + await c.open("main", "listing") + await c.press("ctrl+f") + opened = await c.wait(lambda: isinstance(app.screen, SearchPalette), 10) + c.check("Ctrl+F opens the search palette", opened, + f"screen={type(app.screen).__name__}") + if not opened: + return + pal = app.screen + inp = pal.query_one("#pal-input", Input) + + # -- text: a mnemonic every x86-64 function starts with ------------------ # + inp.value = "endbr64" + await c.press("enter") + await c.wait(lambda: bool(pal._hits), 30) + c.check("a text search finds instructions", len(pal._hits) > 1, + f"n={len(pal._hits)}") + c.check("and it was classified as text", + pal._searched and pal._searched[0] == "text", f"{pal._searched}") + c.check("hits carry the line they matched", + all("endbr64" in h.line for h in pal._hits[:5]), + [h.line for h in pal._hits[:3]]) + + # -- text with padding: match what is SEEN, not IDA's column spacing ----- # + inp.value = "call cs:" + await c.press("enter") + found = await c.wait(lambda: pal._searched == ("text", "call cs:"), 30) + c.check("a query spanning IDA's column padding still matches", + found and len(pal._hits) > 0, f"n={len(pal._hits)}") + + # -- bytes: the same endbr64, as a pattern ------------------------------- # + inp.value = "f3 0f 1e fa" + await c.press("enter") + await c.wait(lambda: pal._searched and pal._searched[0] == "bytes", 30) + c.check("a hex query is classified as bytes", + pal._searched and pal._searched[0] == "bytes", f"{pal._searched}") + c.check("and finds the same instruction", len(pal._hits) > 1, + f"n={len(pal._hits)}") + + # -- wildcards ----------------------------------------------------------- # + inp.value = "f3 0f ?? fa" + await c.press("enter") + await c.wait(lambda: pal._searched == ("bytes", "f3 0f ?? fa"), 30) + c.check("a wildcard byte matches", len(pal._hits) > 1, f"n={len(pal._hits)}") + + # -- a bad pattern must SAY so, not answer "no matches" ------------------ # + inp.value = "48 zz c3" + await c.press("enter") + await c.pause(0.1) + title = str(app.screen.query_one("#pal-box").border_title) + c.check("a malformed byte pattern is refused with a reason", + "not a byte" in title, f"title={title!r}") + + # -- F2 pins the mode against the guess ---------------------------------- # + inp.value = "dead" + await c.pause(0.05) + c.check("a hex-looking WORD still searches text", + pal._mode_query()[0] == "text", f"{pal._mode_query()}") + await c.press("f2") + c.check("F2 forces it to bytes", pal._mode_query()[0] == "bytes", + f"{pal._mode_query()}") + + # -- Enter on a result navigates ----------------------------------------- # + inp.value = "endbr64" + await c.press("f2") # back to text + await c.press("enter") + await c.wait(lambda: bool(pal._hits) and pal._searched + and pal._searched[0] == "text", 30) + target = pal._hits[1] if len(pal._hits) > 1 else pal._hits[0] + pal.query_one(OptionList).highlighted = 1 if len(pal._hits) > 1 else 0 + await c.press("enter") + closed = await c.wait(lambda: not isinstance(app.screen, SearchPalette), 10) + c.check("Enter on a hit closes the palette", closed, + f"screen={type(app.screen).__name__}") + landed = await c.wait( + lambda: app._cur is not None and c.lst._cursor_ea() == target.head, 30) + c.check("and lands the cursor on it", landed, + f"cursor={c.lst._cursor_ea()} want={target.head:#x}") + + +@scenario("export_findings") +async def s_export_findings(c: Ctx): + """Ctrl+E writes a markdown report of what this session worked out. + + Deliberately end-to-end: the interesting failure is not the formatting (that + is covered offline in test_findings.py) but whether a comment and a rename + made through the UI come back out of the database and into the file. + """ + import tempfile + + from idatui.findings import default_path + + app = c.app + fn = c.biggest() + tag = os.getpid() + newname, note = f"exp_{tag}", f"found_it_{tag}" + old = fn.name + await c.open(fn.addr, "listing") + + # Make something to find: a rename and a comment, through the real paths. + app.program.client.invoke( + "rename", batch={"func": {"addr": hex(fn.addr), "name": newname}}) + app.program.bump_names() + app.program.set_comment(fn.addr, note) + app.program.invalidate(fn.addr) + # ...and tell the journal, exactly as the edit controller would. The + # database cannot say who wrote a comment (IDA's own analyzer uses the same + # call), so the journal is what makes this MY finding rather than noise. + app.journal.record("rename", fn.addr, f"{old} → {newname}") + app.journal.record("comment", fn.addr, note) + + out = os.path.join(tempfile.gettempdir(), f"idatui-findings-{tag}.md") + try: + await c.press("ctrl+e") + inp = app.query_one("#export", Input) + opened = await c.wait(lambda: inp.display, 5) + c.check("Ctrl+E opens the export prompt", opened, f"display={inp.display}") + c.check("the prompt is prefilled with a path beside the binary", + inp.value == default_path(app._open_path), f"value={inp.value!r}") + inp.value = out + await c.press("enter") + written = await c.wait(lambda: os.path.exists(out), 30) + c.check("Enter writes the report", written, f"no {out}") + if not written: + return + doc = open(out, encoding="utf-8").read() + c.check("the report is markdown with the expected sections", + doc.startswith("# Findings") and "## Comments" in doc + and "## Named functions" in doc, doc[:60]) + c.check("a comment written this session is in it", note in doc, + doc[:200]) + c.check("and the function it belongs to is named", newname in doc, + doc[:200]) + c.check("the report is sourced from the journal, not a scan", + "idatui's edit journal" in doc, + [l for l in doc.splitlines() if "**source**" in l]) + c.check("the analyzer's own comments stay out of it", + "switch jump" not in doc and "jumptable" not in doc, + [l for l in doc.splitlines() if "switch" in l][:2]) + c.check("the status line says where it went", + out in c.status(), c.status()) + # The journal has to survive the database, or a report is only ever + # about the session that happened to be open. + from idatui.journal import Journal + + app.journal.flush(app.program) + reloaded = Journal() + reloaded.load(app.program) + c.check("the journal round-trips through the .i64", + fn.addr in reloaded.addresses(), + f"{len(reloaded)} entries, {sorted(reloaded.addresses())[:3]}") + finally: + # Idempotent: hand the database back exactly as we found it. + app.program.set_comment(fn.addr, "") + app.program.client.invoke( + "rename", batch={"func": {"addr": hex(fn.addr), "name": old}}) + app.program.bump_names() + app.program.invalidate(fn.addr) + if os.path.exists(out): + os.remove(out) + + +@scenario("struct_filter") +async def s_struct_filter(c: Ctx): + app = c.app + await c.press("ctrl+t") + if not await c.wait(lambda: isinstance(app.screen, StructEditor), 10): + c.check("Ctrl+T opens the struct editor", False, + f"screen={type(app.screen).__name__}") + return + se = app.screen + await c.wait(lambda: bool(se._structs), 15) + total = len(se._structs) + ol = se.query_one("#se-list", OptionList) + inp = se.query_one("#se-filter", Input) + # Something past the first row, so a filter that "works" by doing nothing + # can't pass: its own name must survive and the others must not. + target = se._structs[min(2, total - 1)].name + q = "".join(ch for ch in target.lower() if ch.isalnum())[:3] + + ol.focus() + await c.press("slash") + opened = await c.wait(lambda: inp.display and app.focused is inp, 5) + c.check("'/' from the list opens the struct filter", opened, + f"display={inp.display} focus={getattr(app.focused, 'id', None)}") + for ch in q: + await c.press(ch) + await c.wait(lambda: len(se._structs) < total, 5) + c.check("typing fuzzy-filters the struct list", + 0 < len(se._structs) < total and + any(s.name == target for s in se._structs), + f"q={q!r} {len(se._structs)}/{total}") + cap = str(se.query_one("#se-title", Static).render()) + c.check("the caption counts what the filter kept", + f"{len(se._structs)}/{total}" in cap, f"caption={cap!r}") + + # 'd' is the delete binding on this screen: in the prompt it must be a + # character, not a destructive verb aimed at the highlighted struct. + await c.press("d") + await c.pause(0.05) + c.check("'d' in the filter types instead of deleting", + isinstance(app.screen, StructEditor) and inp.value == q + "d", + f"screen={type(app.screen).__name__} value={inp.value!r}") + await c.press("backspace") + await c.wait(lambda: inp.value == q, 5) + + # Arrows drive the list while the prompt keeps focus (symbol-palette feel). + before = ol.highlighted + await c.press("down") + await c.pause(0.05) + c.check("arrows move the list while the filter has focus", + app.focused is inp and (ol.highlighted != before + or ol.option_count == 1), + f"{before} -> {ol.highlighted} of {ol.option_count}") + + sel = se._structs[ol.highlighted or 0].name + ta = se.query_one(TextArea) + ta.text = "" + await c.press("enter") + loaded = await c.wait(lambda: sel in ta.text, 15) + c.check("Enter in the filter loads the highlighted struct", loaded, + f"want {sel!r} in {ta.text[:40]!r}") + + # Esc backs out one level at a time: definition -> filter -> dialog. + await c.press("escape") + await c.wait(lambda: app.focused is ol, 5) + c.check("Esc leaves the definition for the list", app.focused is ol, + f"focus={getattr(app.focused, 'id', None)}") + await c.press("escape") + cleared = await c.wait(lambda: len(se._structs) == total, 5) + c.check("Esc clears the filter instead of closing", + cleared and not inp.display and isinstance(app.screen, StructEditor), + f"n={len(se._structs)}/{total} display={inp.display}") + await c.press("escape") + await c.wait(lambda: not isinstance(app.screen, StructEditor), 10) + c.check("a third Esc closes the editor", + not isinstance(app.screen, StructEditor), + f"screen={type(app.screen).__name__}") + + @scenario("open_default_view") async def s_open(c: Ctx): app = c.app @@ -1146,7 +1755,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)", @@ -1172,22 +1783,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}") @@ -1209,11 +1821,16 @@ async def s_follow_xrefs(c: Ctx): orig_name = app._cur.name depth = len(app._nav) await c.press("enter") - await c.wait(lambda: len(app._nav) > depth, 25) + # Wait for exactly what the check asserts. Waiting only on the nav depth let + # the check run while _cur was still the function we jumped FROM -- the + # follow pushes the source entry before it opens the target -- so this + # failed about one run in ten with cur == orig, at full speed, looking like + # a code regression. + await c.wait(lambda: len(app._nav) > depth and app._cur.ea != orig, 25) c.check("Enter follows the call into another function", app._cur.ea != orig and len(app._nav) > depth, f"cur={app._cur.ea:#x}") await c.press("escape") - await c.pause(0.1) + await c.wait(lambda: app._cur.ea == orig, 15) c.check("Esc returns from the follow", app._cur.ea == orig, f"cur={app._cur.ea:#x}") dis.cursor = call_idx dis.refresh() @@ -1255,7 +1872,8 @@ async def s_follow_xrefs(c: Ctx): await c.press("tab") landed = await c.wait( lambda: (app._active == "decomp" and dec.loaded_ea == xref.fn_addr) - or (app._active == "listing" and "fail" in c.status().lower()), 25) + or (app.is_listing + and _CANNOT_DECOMP in c.status().lower()), 25) if app._active == "decomp": c.check("F5 at the xref site decompiles the referencing function", dec.loaded_ea == xref.fn_addr, f"loaded={dec.loaded_ea}") @@ -1427,14 +2045,14 @@ async def s_decomp_nav(c: Ctx): old_ea = dec._line_ea(drow) if old_ea is not None and dsym in old_line: tmp = f"stale_{os.getpid()}" - app.program.client.call("rename", batch={"func": {"addr": hex(dstale), "name": tmp}}) + app.program.client.invoke("rename", batch={"func": {"addr": hex(dstale), "name": tmp}}) app.program.bump_names() d2 = len(app._nav) app._follow_decomp(old_line, dsym, old_ea) await c.wait(lambda: len(app._nav) > d2, 25) c.check("decomp follow works with a stale name (ea-marker fallback)", app._cur.ea == dstale, f"cur={app._cur.ea:#x} want={dstale:#x}") - app.program.client.call("rename", batch={"func": {"addr": hex(dstale), "name": dsym}}) + app.program.client.invoke("rename", batch={"func": {"addr": hex(dstale), "name": dsym}}) app.program.bump_names() @@ -1515,7 +2133,7 @@ async def s_rename(c: Ctx): c.check("rename updates the function name", app._func_index.by_addr(dtarget).name == newname, app._func_index.by_addr(dtarget).name) - rr = app.program.client.call("rename", batch={"func": {"addr": hex(dtarget), "name": dsym}}) + rr = app.program.client.invoke("rename", batch={"func": {"addr": hex(dtarget), "name": dsym}}) c.check("rename reverted cleanly", rr.get("summary", {}).get("ok", 0) == 1, str(rr.get("summary"))) # goto label refuse @@ -1561,11 +2179,15 @@ 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.call("set_comments", items=[{"addr": hex(cea), "comment": ""}]) + app.program.client.invoke("set_comments", items=[{"addr": hex(cea), "comment": ""}]) else: c.check("found a pseudocode line to comment", False, "no marker line") @@ -1607,7 +2229,7 @@ async def s_comment_func(c: Ctx): la is not None and lb is not None and lb > la and a not in dec._texts[lb], f"la={la} lb={lb}") - app.program.client.call("set_comments", items=[{"addr": hex(fn.addr), "comment": ""}]) + app.program.client.invoke("set_comments", items=[{"addr": hex(fn.addr), "comment": ""}]) @scenario("retype") @@ -1771,7 +2393,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}) " @@ -1872,7 +2497,7 @@ async def s_rename_history(c: Ctx): await c.pause(0.15) c.check("caller pseudocode shows renamed callee after 'back'", any(hnew in tx for tx in dec._texts), "pseudocode still stale") - app.program.client.call("rename", batch={"func": {"addr": hex(htarget), "name": hsym}}) + app.program.client.invoke("rename", batch={"func": {"addr": hex(htarget), "name": hsym}}) @scenario("region_define") @@ -1952,8 +2577,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}") @@ -2091,7 +2720,7 @@ async def s_listing_name_addr(c: Ctx): finally: # revert: drop the label and restore raw bytes at A try: - c.prog.client.call("rename", batch={"data": {"addr": hex(A + 1), "new": ""}}) + c.prog.client.invoke("rename", batch={"data": {"addr": hex(A + 1), "new": ""}}) except Exception: # noqa: BLE001 pass c.prog.undefine(A, size=8) @@ -2168,7 +2797,7 @@ async def s_listing_struct_expand(c: Ctx): c.check("found a data address for the struct test", False) return try: - c.prog.client.call( + c.prog.client.invoke( "declare_type", decls=["struct TuiExpandS { int a; char b[4]; short c; };"]) c.prog.make_data(A, "TuiExpandS") @@ -2230,7 +2859,8 @@ async def s_continuous_view(c: Ctx): c.lst.focus() await c.press("tab") await c.wait(lambda: (app._active == "decomp" and c.dec.loaded_ea == fn_ea) - or (app._active == "listing" and "fail" in c.status().lower()), 25) + or (app.is_listing + and _CANNOT_DECOMP in c.status().lower()), 25) if app._active == "decomp": c.check("F5/Tab decompiles the function under the cursor", c.dec.loaded_ea == fn_ea, f"loaded={c.dec.loaded_ea}") @@ -2269,28 +2899,828 @@ async def s_func_banners(c: Ctx): c.check("the proc header names the function", hdr is not None, f"fn={fn.name}") -# --------------------------------------------------------------------------- # -# Runner -# --------------------------------------------------------------------------- # -async def _build_pristine(binary, cache): - """Analyse ``binary`` once and keep the resulting database as a golden copy. +def _find_literal(c, start=0, limit=150): + """(head, show) for a listing row at/after ``start`` whose literal is worth + reformatting: a value over 9, so hex and decimal actually LOOK different. - Costs one full analysis, then every later run starts from it instead of - re-analysing. + Scans FORWARD FROM THE CURSOR rather than from row 0: the listing is + continuous over the whole segment, so row 0 is nowhere near the function + that was opened. """ - print(f" (building pristine database for {os.path.basename(binary)}\u2026)") - app = IdaTui(open_path=binary, keepalive=False) - async with app.run_test(size=(140, 44)) as pilot: - for _ in range(6000): - await pilot.pause(0.05) - if app._func_index is not None and app._func_index.complete: + for i in range(start, min(start + limit, len(c.lst.model))): + h = c.lst.model.get(i) + if h is None or h.kind != "code": + continue + try: + show = c.prog.op_format(h.ea, mode="show") + except Exception: # noqa: BLE001 -- no literal on this line + continue + v = show.get("value") + if v and int(v, 16) > 9 and {"hex", "dec"} <= set(show.get("choices", [])): + return h, show + return None + + +async def _park_on(c, ea, tries=25): + """Put the listing cursor on ``ea`` and make sure it STAYS there. + + An open that is still settling lands its own cursor when its worker + finishes, which silently moves a cursor a test set by hand — and then the + keypress under test edits somewhere else entirely. + """ + held = 0 + for _ in range(tries): + if c.lst._cursor_ea() == ea: + held += 1 + if held >= 3: + return True + else: + held = 0 + i = c.lst.model.index_of_ea(ea) + if i >= 0: + c.lst.cursor = i + c.lst.cursor_x = c.lst._insn_col(i) + await c.pause(0.1) + return False + + +@scenario("opfmt_listing") +async def s_opfmt_listing(c: Ctx): + """'o' cycles how the literal under the cursor is DISPLAYED (IDA's 'o'): + hex -> dec -> bin -> ... -> default, 'O' the other way, and the listing + re-renders in place.""" + app = c.app + await c.open_biggest("listing") + c.lst.model.load_all() + found = _find_literal(c, start=c.lst.cursor) + if found is None: + c.check("found a listing literal to reformat", False) + return + head, show = found + ea = head.ea + c.lst.focus() + parked = await _park_on(c, ea) + c.check("the cursor is on the literal's line", parked, + f"want {ea:#x}, cursor at {c.lst._cursor_ea():#x}") + before = head.text + try: + await c.press("o") + await c.wait(lambda: c.lst.model.index_of_ea(ea) >= 0 + and (c.lst.model.get(c.lst.model.index_of_ea(ea)) or head).text + != before, 25) + i = c.lst.model.index_of_ea(ea) + after = c.lst.model.get(i).text if i >= 0 else before + c.check("'o' re-renders the literal", after != before, + f"{before!r} -> {after!r} ea={ea:#x} show={show}") + c.check("the status names the format it moved to", + any(f in c.status() for f in show["choices"]), + f"status={c.status()!r} choices={show['choices']}") + c.check("the change is marked unsaved", app._dirty) + # 'O' walks the ring the other way: back to where we started. + await c.press("O") + await c.wait(lambda: c.lst.model.index_of_ea(ea) >= 0 + and (c.lst.model.get(c.lst.model.index_of_ea(ea)) or head).text + == before, 25) + j = c.lst.model.index_of_ea(ea) + c.check("'O' cycles back", j >= 0 and c.lst.model.get(j).text == before, + f"{c.lst.model.get(j).text if j >= 0 else None!r} want {before!r}") + # An explicit format by name (what the palette/RPC use). + r = c.prog.op_format(ea, mode="dec") + c.check("an explicit format renders decimal", + r["format"] == "dec" and str(int(show["value"], 16)) in r["text"], + str(r)) + finally: + try: + c.prog.op_format(ea, mode="default", n=show.get("n", -1)) + except Exception: # noqa: BLE001 + pass + c.prog.bump_names() + + +@scenario("opfmt_no_literal") +async def s_opfmt_no_literal(c: Ctx): + """A line with nothing to reformat says so instead of picking something.""" + await c.open_biggest("listing") + c.lst.model.load_all() + + def _banner(i): + h = c.lst.model.get(i) + return h is not None and h.kind in ("sep", "funchdr", "label") + + row = next((i for i in range(c.lst.cursor, min(c.lst.cursor + 400, + len(c.lst.model))) + if _banner(i)), None) + if row is None: + c.check("found a banner row", False) + return + c.lst.focus() + for _ in range(20): # hold it against a late-landing open + c.lst.cursor = row + await c.pause(0.05) + if c.lst.cursor == row: + break + c.check("the cursor is on a banner row", _banner(c.lst.cursor), + f"row={c.lst.cursor}") + await c.press("o") + await c.wait(lambda: "reformat" in c.status() or "format" in c.status(), 15) + c.check("'o' on a line with no literal explains itself", + "reformat" in c.status(), f"status={c.status()!r}") + + +@scenario("opfmt_refusal_is_not_swallowed") +async def s_opfmt_refusal_visible(c: Ctx): + """A refusal right after a successful format still reaches the status bar. + + A result written without priority loses to the PREVIOUS result's flash for + 8 seconds, so a refusal left the last success on the bar — and the RPC + snapshot reads that same bar, so a driver saw text that looked like the + edit had worked. + + Driven the way the RPC verb drives it: the action called directly, with no + keystroke. A keypress clears the flash on its way in, which is why pressing + 'o' hides this bug entirely and only a driver ever saw it. + """ + await c.open_biggest("listing") + c.lst.model.load_all() + found = _find_literal(c, start=c.lst.cursor) + if found is None: + c.check("found a listing literal to reformat", False) + return + head, show = found + c.lst.focus() + if not await _park_on(c, head.ea): + c.check("the cursor is on the literal's line", False) + return + try: + await c.press("o") # a success: sets the flash + await c.wait(lambda: "\u2192" in c.status(), 25) + good = c.status() + + # ... and, within the flash window, 'o' where there is nothing to do. + # The edit rebuilt the segment model, so this is a different (empty) + # one than the load_all above filled. + c.lst.model.load_all() + + def _banner(i): + h = c.lst.model.get(i) + return h is not None and h.kind in ("sep", "funchdr", "label") + + row = next((i for i in range(c.lst.cursor, + min(c.lst.cursor + 400, len(c.lst.model))) + if _banner(i)), None) + if row is None: + c.check("found a banner row to refuse on", False) + return + for _ in range(20): + c.lst.cursor = row + await c.pause(0.05) + if c.lst.cursor == row: break - app.program.client.call("idb_save", timeout=600.0) - db = binary + ".i64" - if os.path.exists(db): - shutil.copy2(db, cache) + c.lst.action_op_format("cycle") # no keypress: as the RPC does it + await c.wait(lambda: c.status() != good, 15) + c.check("a refusal replaces the previous success on the status bar", + c.status() != good and "reformat" in c.status(), + f"still showing {c.status()!r}") + finally: + try: + c.prog.op_format(head.ea, mode="default", n=show.get("n", -1)) + except Exception: # noqa: BLE001 + pass + c.prog.bump_names() + + +def _styled_cols(strip, style_attr, want): + """Columns of ``strip`` whose rendered style has ``style_attr`` == want.""" + cols, x = [], 0 + for seg in strip: + st = seg.style + if st is not None and getattr(st, style_attr, None) is not None \ + and str(getattr(st, style_attr)) == want: + cols.extend(range(x, x + len(seg.text))) + x += len(seg.text) + return cols + + +@scenario("opfmt_highlight") +async def s_opfmt_highlight(c: Ctx): + """The literal 'o' would reformat is MARKED on screen before you press it. + + A line can carry several literals and the cursor picks one; without showing + which, you find out by pressing and reading the status. The marker must also + survive the cursor-line decoration, which paints the word under the cursor + (usually the same characters) and used to win. + """ + from idatui.app import _S_OPERAND + await c.open_biggest("listing") + c.lst.model.load_all() + lst = c.lst + lst.focus() + # A row with two operands, so "which one" is a real question. + row = next((i for i in range(lst.cursor, min(lst.cursor + 400, len(lst.model))) + if lst.model.get(i) is not None + and (lst.model.get(i).ops or ()) and len(lst.model.get(i).ops) >= 2), + None) + if row is None: + c.check("found a row with two operands", False) + return + h = lst.model.get(row) + if not await _park_on(c, h.ea): + c.check("parked on the two-operand row", False) + return + row = lst.model.index_of_ea(h.ea) + h = lst.model.get(row) + base = lst._insn_col(row) + want_bg = str(_S_OPERAND.bgcolor) + seen = [] + for lo, hi, n in h.ops: + lst.cursor_x = base + lo + await c.pause(0.05) + strip = lst.render_line(row - round(lst.scroll_offset.y)) + cols = _styled_cols(strip, "bgcolor", want_bg) + seen.append((n, min(cols) if cols else None, max(cols) + 1 if cols else None)) + c.check(f"operand {n} ({h.text[lo:hi]!r}) is marked when the cursor is on it", + cols and min(cols) == base + lo and max(cols) + 1 == base + hi, + f"marked={min(cols) if cols else None}.." + f"{max(cols)+1 if cols else None} want={base+lo}..{base+hi}") + c.check("the mark MOVES between the operands (it isn't the whole line)", + len({s[1] for s in seen}) == len(seen), str(seen)) + # ... and the marked operand is the one the edit acts on — either it gets + # reformatted, or the refusal names that same operand. What must never + # happen is a different operand quietly changing. + for lo, hi, n in h.ops: + lst.cursor_x = base + lo + try: + r = c.prog.op_format(h.ea, mode="show", col=lst.op_col()) + got, why = r.get("n"), "" + except IDAToolError as e: # "operand N (rsp) has no format" + m = re.search(r"operand (\d+)", e.message) + got, why = (int(m.group(1)) if m else None), e.message + c.check(f"marked operand {n} is the one acted on (or refused)", + got == n, f"marked op{n}, worker said op{got} {why}") + + +@scenario("opfmt_sticks_to_its_literal") +async def s_opfmt_sticks(c: Ctx): + """Pressing 'o' twice cycles the SAME literal, even when the line reflows. + + A reformat changes the printed width (``48`` -> ``0x30``), which moves every + literal to its right. Holding the cursor column meant the second press + landed on a different literal — you cycle one number and a neighbour + changes. + """ + app = c.app + pick = None + for fn in c.all_funcs()[:60]: + d = c.prog.decompile(fn.addr) + if d.failed or not d.code: + continue + nums = c.prog.pc_nums(fn.addr) + for line, recs in sorted(nums.items()): + # The second literal must be one whose printed WIDTH changes as it + # cycles (0x36u vs 54), or the cursor never falls off it and the + # test proves nothing. + if len(recs) >= 2 and recs[0][1] < recs[1][0] \ + and int(recs[1][2], 16) >= 16: + pick = (fn, line, recs) + break + if pick: + break + if pick is None: + c.check("found a pseudocode line with two literals", False) + return + fn, line, recs = pick + first, second = recs[0], recs[1] + target = (second[3], second[4]) # (ea, opnum) of the literal we mean + other = (first[3], first[4]) + try: + # Make it WIDE first (0x30, four characters). The cursor then sits on a + # column that stops existing when the literal is printed short (48) -- + # which is the whole failure: the next press finds no literal under the + # cursor and quietly falls back to the first one on the line. + c.prog.pc_num_format(fn.addr, mode="hex", line=line, col=second[0]) + c.prog.bump_names() + await c.open(fn.addr, "decomp") + if app._active != "decomp": + c.check("pseudocode view opened", False, f"active={app._active}") + return + dec = c.dec + dec.focus() + wide = next((r for r in dec._nums.get(line, ()) + if (r[3], r[4]) == target), None) + if wide is None or wide[1] - wide[0] < 3: + c.check("the literal is now printed wide", False, + f"nums={dec._nums.get(line)}") + return + dec.cursor, dec.cursor_x = line, wide[1] - 1 # its LAST character + dec.refresh() + await c.pause(0.05) + seen = [] + for _ in range(3): + before = dec._texts[line] + await c.press("o") + await c.wait(lambda: dec.loaded_ea == fn.addr + and line < len(dec._texts) + and dec._texts[line] != before, 30) + cur = next(((r[3], r[4]) for r in dec._nums.get(line, ()) + if r[0] <= dec.cursor_x < r[1]), None) + seen.append(cur) + c.check("every press stays on the literal we started on", + all(s == target for s in seen), + f"target={target} other={other} landed={seen} " + f"line={dec._texts[line].strip()!r}") + finally: + try: + c.prog.pc_num_format(fn.addr, mode="default", line=line, + col=second[0]) + except Exception: # noqa: BLE001 + pass + c.prog.bump_names() + + +@scenario("cursor_on_stays_visible") +async def s_cursor_on_visible(c: Ctx): + """``cursor_on`` lands somewhere you can SEE, near where you are. + + It used to scan from row 0 of the whole segment and move the cursor without + scrolling, so `drive fmt dec 18h` in main reformatted an ``18h`` 170 rows + away, off screen: the driver reported success and the pane showed a line + that hadn't changed. + """ + from idatui.rpc import cursor_on + app = c.app + fn = await c.open_biggest("listing") + c.lst.model.load_all() + lst = c.lst + lst.focus() + await c.pause(0.1) + top = round(lst.scroll_offset.y) + # A token that occurs both before the viewport and inside it. + here = next((t for t in ("rax", "rsp", "eax", "rbp", "rdi") + if any(t in (lst._line_plain(i) or "") + for i in range(top, min(top + 20, lst.total))) + and any(t in (lst._line_plain(i) or "") for i in range(0, top))), + None) + if here is None: + c.check("found a token both above and inside the viewport", False, + f"top={top}") + return + found = cursor_on(app, here) + await c.pause(0.1) + c.check(f"cursor_on({here!r}) found it", found) + vis = round(lst.scroll_offset.y) + c.check("it lands inside the viewport, not thousands of rows above", + vis <= lst.cursor < vis + lst._visible_height(), + f"cursor={lst.cursor} viewport={vis}..{vis + lst._visible_height()}") + c.check("and it searched from the viewport, not from row 0", + lst.cursor >= top, f"cursor={lst.cursor} was top={top}") + # An explicit line still wins, and lands visibly. + far = next((i for i in range(0, min(top, lst.total)) + if here in (lst._line_plain(i) or "")), None) + if far is not None: + cursor_on(app, here, line=far) + await c.pause(0.1) + v2 = round(lst.scroll_offset.y) + c.check("an explicit line is honoured AND scrolled into view", + lst.cursor == far and v2 <= far < v2 + lst._visible_height(), + f"cursor={lst.cursor} want={far} viewport={v2}") + # The `cursor` verb has the same duty: a driver that parks the cursor for + # an edit must leave it where the edit can be watched. + from idatui.rpc import place_cursor + deep = min(lst.total - 1, 900) + place_cursor(lst, deep, 0) + await c.pause(0.1) + v3 = round(lst.scroll_offset.y) + c.check("`cursor line=` scrolls to what it selected", + v3 <= deep < v3 + lst._visible_height(), + f"cursor={lst.cursor} viewport={v3}..{v3 + lst._visible_height()}") + + +@scenario("opfmt_decomp") +async def s_opfmt_decomp(c: Ctx): + """'o' in the pseudocode reformats the C literal under the cursor. Hex-Rays + keeps its own number formats, so this is a different edit from the + listing's — and the decompilation re-renders.""" + app = c.app + pick = None + for fn in c.all_funcs()[:60]: + d = c.prog.decompile(fn.addr) + if d.failed or not d.code: + continue + for i, txt in enumerate(d.code.split("\n")): + m = re.search(r"[=<>+\-*/(,]\s(\d{2,}|0x[0-9A-Fa-f]{2,})\b", txt) + if m and "//" not in txt[:m.start()]: + pick = (fn, i, m.start(1)) + break + if pick: + break + if pick is None: + c.check("found a pseudocode number to reformat", False) + return + fn, line, col = pick + await c.open(fn.addr, "decomp") + if app._active != "decomp": + c.check("pseudocode view opened", False, f"active={app._active}") + return + dec = c.dec + dec.focus() + dec.cursor, dec.cursor_x = line, col + dec.refresh() + await c.pause(0.05) + before = dec._texts[line] + try: + await c.press("o") + await c.wait(lambda: dec.loaded_ea == fn.addr and line < len(dec._texts) + and dec._texts[line] != before, 30) + c.check("'o' re-renders the pseudocode literal", + line < len(dec._texts) and dec._texts[line] != before, + f"{before!r} -> {dec._texts[line] if line < len(dec._texts) else None!r}") + c.check("the status says which format", + any(f in c.status() for f in ("hex", "dec", "oct", "char", "default")), + f"status={c.status()!r}") + finally: + try: + c.prog.pc_num_format(fn.addr, mode="default", line=line, col=col) + except Exception: # noqa: BLE001 + pass + c.prog.bump_names() + +@scenario("opfmt_opcode_key_moved") +async def s_opfmt_opcode_key_moved(c: Ctx): + """'o' now belongs to the operand format, so the opcode-bytes column moved + to 'B' — and still cycles off/limited/full.""" + await c.open_biggest("listing") + lst = c.lst + lst.focus() + await c.pause(0.05) + modes = [lst._op_mode] + for _ in range(3): + await c.press("B") + await c.pause(0.05) + modes.append(lst._op_mode) + c.check("'B' cycles the opcode-bytes column", len(set(modes)) == 3, str(modes)) + c.check("and returns to where it started", modes[0] == modes[3], str(modes)) + +# --------------------------------------------------------------------------- # +# Graph view +# --------------------------------------------------------------------------- # +async def _open_graph(c: Ctx, fn=None, t=60): + """Open a multi-block function and press Space. Returns (fn, GraphView).""" + app = c.app + if fn is None: + # A function with real branching -- a straight-line one proves nothing + # about layering, and a huge one is slow. + fn = c.find_func(lambda f: 0x200 < f.size < 0x600) or c.biggest() + await c.open(fn.addr, "listing") + c.lst.focus() + await c.pause(0.05) + await c.press("space") + gv = app.query_one(GraphView) + ok = await c.wait(lambda: app._active == "graph" and gv.lay is not None, t) + c.check("the graph opened", ok, + f"active={app._active} sticky={app._graph_sticky} " + f"focus={type(app.focused).__name__} prompt={app._prompt_active()} " + f"status={c.status()!r}") + return fn, gv + + +@scenario("graph_open") +async def s_graph_open(c: Ctx): + app = c.app + fn, gv = await _open_graph(c) + c.check("space opens the graph view", app._active == "graph", + f"active={app._active} status={c.status()}") + if gv.lay is None: + return + c.check("the graph has the function's blocks", + len(gv.lay.nodes) == len(gv.fc.blocks) and len(gv.lay.nodes) > 1, + f"nodes={len(gv.lay.nodes)} blocks={len(gv.fc.blocks) if gv.fc else 0}") + c.check("it is the right function", gv.fc is not None and gv.fc.func_ea == fn.addr, + f"{gv.fc.func_ea if gv.fc else None:#x} want {fn.addr:#x}") + c.check("the cursor starts on a real address", gv._cursor_ea() is not None) + # The invariant the whole dummy-node machinery exists for. + boxes = [(n.x, n.y, n.right, n.y + n.h - 1) for n in gv.lay.nodes] + overlap = any(a[0] <= b[2] and b[0] <= a[2] and a[1] <= b[3] and b[1] <= a[3] + for i, a in enumerate(boxes) for b in boxes[i + 1:]) + c.check("no two blocks overlap", not overlap) + c.check("the status names the graph", "graph" in c.status(), c.status()) + await c.press("space") + await c.wait(lambda: app._active != "graph", 15) + c.check("space returns to the listing", app._active == "listing", + f"active={app._active}") + + +@scenario("graph_nav") +async def s_graph_nav(c: Ctx): + app = c.app + fn, gv = await _open_graph(c) + if gv.lay is None or len(gv.lay.nodes) < 2: + c.check("graph has enough blocks to navigate", False) + return + start_ea = gv._cursor_ea() + await c.press("j") + await c.pause(0.05) + c.check("j moves the cursor within the block", gv._cursor_ea() != start_ea, + f"{start_ea:#x} -> {gv._cursor_ea():#x}") + await c.press("k") + await c.pause(0.05) + c.check("k comes back", gv._cursor_ea() == start_ea) + # J follows an edge to a successor block + b0 = gv.cursor_node + succs = gv.lay.succ.get(b0) or [] + if succs: + await c.press("J") + await c.pause(0.1) + c.check("J follows an edge to a successor block", + gv.cursor_node == succs[0][0], + f"node={gv.cursor_node} want={succs[0][0]}") + await c.press("K") + await c.pause(0.1) + c.check("K goes back up an edge", gv.cursor_node == b0, + f"node={gv.cursor_node} want={b0}") + await c.press("0") + await c.pause(0.1) + c.check("0 returns to the entry block", gv.cursor_node == gv.fc.entry, + f"node={gv.cursor_node} entry={gv.fc.entry}") + # the cursor is always scrolled into view + cell = gv._cursor_cell() + top, left = int(gv.scroll_offset.y), int(gv.scroll_offset.x) + c.check("the cursor block is scrolled into view", + cell is not None and top <= cell[0] < top + gv.size.height + and left <= cell[1] < left + gv.size.width, + f"cell={cell} scroll=({top},{left}) size={gv.size}") + + +@scenario("graph_zoom") +async def s_graph_zoom(c: Ctx): + app = c.app + fn, gv = await _open_graph(c) + if gv.lay is None: + c.check("graph loaded", False) + return + full_h = gv.lay.height + seen = [gv.ZOOMS[gv._zoom]] + for _ in range(2): + await c.press("z") + await c.pause(0.15) + seen.append(gv.ZOOMS[gv._zoom]) + c.check("z cycles the three zoom levels", seen == ["full", "compact", "collapsed"], + str(seen)) + c.check("collapsed is much smaller than full", gv.lay.height < full_h, + f"{gv.lay.height} vs {full_h}") + c.check("the cursor survives a zoom", gv._cursor_ea() is not None) + c.check("the status still names the function", + gv.fc.name in c.status(), c.status()) + await c.press("z") + await c.pause(0.15) + c.check("and wraps back to full", gv.ZOOMS[gv._zoom] == "full") + c.check("canvas is restored", gv.lay.height == full_h, + f"{gv.lay.height} vs {full_h}") + + +@scenario("graph_render") +async def s_graph_render(c: Ctx): + """The drawing itself: boxes, instruction text and edge glyphs must actually + reach the screen. A layout that is right but paints nothing looks identical + to a broken one from the outside.""" + app = c.app + fn, gv = await _open_graph(c) + if gv.lay is None: + c.check("graph loaded", False) + return + # 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"), + "no edge glyphs on screen") + ea = gv._cursor_ea() + head = gv.cur_head() + c.check("the cursor block's instruction text is on screen", + head is not None and head.text.split(" ")[0] in blob, + f"mnem={head.text.split(' ')[0] if head else None}") + c.check("the address gutter renders at full zoom", + ea is not None and f"{ea:08X}" in blob, f"ea={ea:#x}") + # minimap on/off actually changes the picture + before = blob + await c.press("m") + 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.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 + for row in range(min(gv.lay.height, 300)): + cells = gv.lay.painting.cells_at_row(row, 0, gv.lay.width) + if not cells: + continue + for n in gv.lay.nodes_at_row(row): + bad += sum(1 for col in cells if n.inside(row, col)) + c.check("no edge is painted inside a block", bad == 0, f"{bad} cells") + + +@scenario("graph_click") +async def s_graph_click(c: Ctx): + app = c.app + fn, gv = await _open_graph(c) + if gv.lay is None or len(gv.lay.nodes) < 2: + c.check("graph has blocks to click", False) + return + # find a block whose body is on screen right now + top, left = int(gv.scroll_offset.y), int(gv.scroll_offset.x) + target = None + for n in gv.lay.nodes: + if (n.id != gv.cursor_node and top <= n.y + 1 < top + gv.size.height - 1 + and left <= n.x + 2 < left + gv.size.width - 2): + target = n + break + if target is None: + c.check("a second block is visible to click", True, "(skipped: none on screen)") + return + PAD = 1 # GraphView { padding: 0 1 } + await c.pilot.click(GraphView, + offset=(PAD + target.x + 2 - left, target.y + 1 - top)) + await c.pause(0.15) + c.check("clicking a block moves the cursor into it", + gv.cursor_node == target.id, + f"node={gv.cursor_node} want={target.id}") + + +@scenario("graph_minimap") +async def s_graph_minimap(c: Ctx): + """The minimap floats over the canvas, so it must be hit-tested BEFORE the + canvas -- otherwise a click on it reads as canvas coordinates and drops the + cursor into whatever block happens to lie underneath.""" + app = c.app + fn, gv = await _open_graph(c) + 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}") + if rect is None: + return + left, top, mw, mh = rect + c.check("it sits inside the pane, clear of the scrollbar", + left + mw <= gv.size.width - 1, + f"left={left} w={mw} pane={gv.size.width}") + + # a big graph, so the overview actually maps to somewhere far away + big = c.find_func(lambda f: f.size > 0x300) or fn + # _open_graph left graph mode STICKY, and a sticky navigation schedules the + # next function's graph by itself -- so whether the Space below ENTERS the + # graph or LEAVES it depended on whether that async load landed first. This + # scenario is about the minimap, not about sticky mode (graph_sticky covers + # that), so drop stickiness and press Space from a known state. Without + # this the whole scenario passes or fails on a coin toss: make the backend + # fast enough that the reload wins and every minimap click lands on a + # widget that is no longer on screen. + app._graph_sticky = False + await c.open(big.addr, "listing") + await c.wait(lambda: app._active == "listing", 10) + c.lst.focus() + await c.press("space") + await c.wait(lambda: app._active == "graph" and gv.lay is not None, 60) + if gv.lay is None or gv.lay.height < gv.size.height * 2: + c.check("a graph tall enough to scrub", True, "(skipped: too small)") + return + + gv.scroll_to(y=0, x=0, animate=False) + await c.pause(0.1) + # click near the BOTTOM of the minimap -> the view should jump down + PAD = 1 + await c.pilot.click(GraphView, offset=(PAD + left + mw // 2, top + mh - 2)) + await c.pause(0.2) + c.check("clicking low on the minimap scrolls the view down", + gv.scroll_offset.y > 0, f"scroll_y={gv.scroll_offset.y}") + # Most of a graph is padding, so a coordinate-accurate jump would park you + # in empty space with the cursor left behind: every minimap click must land + # on a block and take the cursor with it. + landed = gv.lay.by_id.get(gv.cursor_node) + c.check("it snaps the cursor onto a real block", + landed is not None and landed.block is not None, + f"node={gv.cursor_node}") + c.check("and that block is what the viewport is showing", + landed is not None + and int(gv.scroll_offset.y) <= landed.y + landed.h + and landed.y <= int(gv.scroll_offset.y) + gv.size.height, + f"node.y={landed.y if landed else None} " + f"scroll={gv.scroll_offset.y} h={gv.size.height}") + low_node = gv.cursor_node + + # and the top of the minimap brings it back to a block up there + await c.pilot.click(GraphView, offset=(PAD + left + mw // 2, top + 1)) + await c.pause(0.2) + top_node = gv.lay.by_id.get(gv.cursor_node) + c.check("clicking high on the minimap goes back up", + top_node is not None and gv.cursor_node != low_node + and top_node.y < gv.lay.by_id[low_node].y, + f"top={gv.cursor_node} low={low_node}") + c.check("the cursor still has a real address after a minimap jump", + gv._cursor_ea() is not None) + + # with the minimap hidden the same click is an ordinary canvas click + await c.press("m") + await c.pause(0.1) + c.check("no hit-box once it's hidden", gv._minimap_rect() is None) + await c.press("m") + await c.pause(0.1) + + # Panning into the padding (which is most of the canvas) must not strand + # you on a blank screen with nothing to navigate back by. + gv.scroll_to(y=max(gv.lay.height - 1, 0), x=max(gv.lay.width - 1, 0), + animate=False) + await c.pause(0.1) + c.check("a pan past the graph leaves the viewport empty", + not gv._viewport_has_block() or True) # setup, not an assertion + gv._snap_into_view() + await c.pause(0.1) + c.check("panning into empty padding snaps back to a block", + gv._viewport_has_block(), + f"scroll={gv.scroll_offset} canvas={gv.lay.width}x{gv.lay.height}") + + +@scenario("graph_rename") +async def s_graph_rename(c: Ctx): + """Editing verbs must work from inside a box -- that is the whole point of + reusing the listing's rows rather than rendering our own.""" + app = c.app + fn, gv = await _open_graph(c) + if gv.lay is None: + c.check("graph loaded", False) + return + ea = gv._cursor_ea() + if ea is None: + c.check("cursor has an address", False) + return + new = f"gtest_{os.getpid()}" + await c.press("n") + await c.wait(lambda: app.query_one("#rename", Input).display, 10) + c.check("n opens the rename prompt from the graph", + app.query_one("#rename", Input).display) + inp = app.query_one("#rename", Input) + inp.value = "" + await c.type(new) + await c.press("enter") + await c.wait(lambda: not app.query_one("#rename", Input).display, 15) + await c.pause(0.4) + try: + got = app.program.resolve(new) + except Exception: # noqa: BLE001 + got = None + c.check("the rename reached the database", got == ea, + f"resolve({new}) -> {got if got is None else hex(got)} want {ea:#x}") + # revert, so the suite stays idempotent + if got is not None: + app.program.client.invoke( + "rename", batch={"data": {"addr": hex(ea), "new": ""}}) + app.program.bump_names() + + +@scenario("graph_sticky") +async def s_graph_sticky(c: Ctx): + """Graph mode survives a navigation: jumping to another function should land + in ITS graph, not dump you back into the listing.""" + app = c.app + fn, gv = await _open_graph(c) + if gv.lay is None: + c.check("graph loaded", False) + return + other = c.find_func(lambda f: f.addr != fn.addr and 0x120 < f.size < 0x500) + if other is None: + c.check("a second function exists", True, "(skipped)") + return + app._goto(other.name) + ok = await c.wait(lambda: app._cur is not None and app._cur.ea == other.addr + and app._active == "graph" + and gv.fc is not None and gv.fc.func_ea == other.addr, 60) + c.check("a goto from the graph lands in the next function's graph", ok, + f"active={app._active} sticky={app._graph_sticky} " + f"cur={app._cur.ea if app._cur else None} " + f"fc={gv.fc.func_ea if gv.fc else None} want={other.addr:#x}") + await c.press("space") + await c.wait(lambda: app._active != "graph", 15) + c.check("space still leaves graph mode", app._active == "listing", + f"active={app._active}") + c.check("and it stops being sticky", app._graph_sticky is False) + + +# --------------------------------------------------------------------------- # +# Runner +# --------------------------------------------------------------------------- # async def run(binary, only=None): # The suite EDITS the database — it defines code, undefines items, renames # and comments — and IDA saves those edits. Run that against the tracked @@ -2302,20 +3732,13 @@ async def run(binary, only=None): # # So: work on a scratch copy, seeded from a golden database that nothing # ever writes back to. - with tempfile.TemporaryDirectory(prefix="idatui-pilot-") as scratch: - target = os.path.join(scratch, os.path.basename(binary)) - shutil.copy2(binary, target) - cache = binary + ".pristine.i64" - if not (os.path.exists(cache) - and os.path.getmtime(cache) >= os.path.getmtime(binary)): - await _build_pristine(target, cache) - if os.path.exists(cache): - shutil.copy2(cache, target + ".i64") + async with staged(binary, lambda p: IdaTui(open_path=p, keepalive=False), + prefix="idatui-pilot-") as target: await _run_on(target, only) async def _run_on(binary, only=None): - # Own idalib worker: opens the binary in-process over a unix socket. + # Code Mode attaches a registered GUI or starts/reuses a managed worker. app = IdaTui(open_path=binary, keepalive=False) async with app.run_test(size=(140, 44)) as pilot: c = Ctx(app, pilot) @@ -2335,6 +3758,14 @@ async def _run_on(binary, only=None): print(f"── {name} ({asyncio.get_event_loop().time() - _t0:.1f}s) CRASHED") c.check("scenario did not crash", False, f"{type(e).__name__}: {e}") traceback.print_exc() + # Headless run_test does not reliably emit App.Unmount; explicitly release + # the lease. Then wait through the managed worker's final-lease grace and + # IDB close so Windows can remove this suite's TemporaryDirectory safely. + if app.program is not None: + app.program.close() + if app.client is not None: + app.client.close() + await asyncio.to_thread(app.client.wait_released, 45.0) def main(argv): @@ -2349,6 +3780,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) @@ -2363,6 +3796,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 |