diff options
| author | blasty <peter@haxx.in> | 2026-07-24 14:53:56 +0200 |
|---|---|---|
| committer | blasty <peter@haxx.in> | 2026-07-24 14:53:56 +0200 |
| commit | 3a28b97cb355822b2d380f9507b203b79cb0e4e9 (patch) | |
| tree | 35565378585b06abf03aa772994d9f9e8b3b2f6b /tests/stress_paging.py | |
| parent | mcp: collapse app + launcher to worker-only (diff) | |
| download | ida-tui-3a28b97cb355822b2d380f9507b203b79cb0e4e9.tar.gz ida-tui-3a28b97cb355822b2d380f9507b203b79cb0e4e9.tar.xz ida-tui-3a28b97cb355822b2d380f9507b203b79cb0e4e9.zip | |
mcp: delete the ida-pro-mcp transport, supervisor, and mcp-only tests
The idalib worker is the only backend now, so remove the dead HTTP/supervisor
surface entirely (~2200 lines):
* deleted idatui/client.py (the IDAClient HTTP/JSON-RPC transport + session
manager), idatui/tui.py (the old mcp TUI entry, superseded by launch.py),
spawn.sh, and systemd/ (the supervisor unit).
* deleted the mcp-only tests (stress_client, smoke_client, test_keepalive,
stress_paging, rpc_smoke, serverctl.sh, pane_smoke, test_domain) -- the worker
pilot (tests/test_scenarios.py) supersedes them.
* migrated the tmux RPC harness (idatui/pane.py) to the worker: it spawns
`idatui.launch <binary> --rpc <sock>` instead of the mcp `idatui.tui`, drops
the supervisor auto-start/ensure machinery, and reaps our own worker
(idatui/worker.py) instead of ida_pro_mcp.idalib_server. --db/--url/--no-
ensure-server are gone; --open is required.
* __init__ / __main__ / domain no longer import client (exceptions come from
errors.py, the domain client hint is WorkerClient); pyproject points both
console scripts at idatui.launch; README + ida-tui header describe the
worker-only flow.
What stays (by design): the ida_pro_mcp *package* (the worker reuses its @tool
functions in-process) and server/patch_server.py (the worker injects its custom
tools on startup). Verified: whole package imports + IdaTui constructs + pilot
lists 31 scenarios. The worker pilot (134 pass / 2 known flakes) is the E2E gate.
Diffstat (limited to 'tests/stress_paging.py')
| -rw-r--r-- | tests/stress_paging.py | 150 |
1 files changed, 0 insertions, 150 deletions
diff --git a/tests/stress_paging.py b/tests/stress_paging.py deleted file mode 100644 index b821354..0000000 --- a/tests/stress_paging.py +++ /dev/null @@ -1,150 +0,0 @@ -#!/usr/bin/env python3 -"""Stress the 'many lines of text' problem: paging huge listings/functions. - -Validates the windowing strategy the TUI will use so we never hand a widget more -than a viewport-sized slice. Prints concise stats only. - - python3 tests/stress_paging.py --db <session_id> -""" -import os -import statistics -import sys -import time - -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from idatui.client import IDAClient # noqa: E402 - -URL = "http://127.0.0.1:8745/mcp" -# Verified server cap: list_* honors count up to ~700, then silently collapses -# to a 10-item default. Clamp with margin. -SAFE_PAGE = 500 - - -def ms(fn): - t = time.time() - r = fn() - return (time.time() - t) * 1e3, r - - -def q0(payload): - """Unwrap a *_query/list_* payload: {'result':[{'data':[...],'next_offset':N}]}.""" - res = payload.get("result", payload) - if isinstance(res, list): - res = res[0] if res else {} - return res.get("data", []), res.get("next_offset") - - -def page_all_funcs(c, page=SAFE_PAGE): - """Enumerate the entire function list correctly: advance by len(data), NOT by - next_offset (which is just offset+count and skips over the per-call cap).""" - funcs, offset, pages, t0 = [], 0, 0, time.time() - while True: - payload = c.call("list_funcs", queries=[{"offset": offset, "count": page}]) - data, _ = q0(payload) - funcs.extend(data) - pages += 1 - if len(data) < page: # short page => reached the end - break - offset += len(data) - return funcs, pages, (time.time() - t0) * 1e3 - - -def disasm_meta(c, addr): - # total_instructions is a TOP-LEVEL field, not under 'asm'. - payload = c.call("disasm", addr=addr, max_instructions=1, include_total=True) - return payload.get("total_instructions", payload.get("instruction_count")) - - -def main(argv): - db = None - it = iter(argv) - for a in it: - if a == "--db": - db = next(it) - c = IDAClient(URL, db=db) - c.connect() - if db is None: - c.set_db(c.resolve_db()) - print(f"db={c.db} ({c.health().get('module')})") - - # ---- 1. Full function list via cursor pagination ------------------- # - print("\n[1] page entire function list (cursor)") - funcs, pages, total_ms = page_all_funcs(c, page=SAFE_PAGE) - n = len(funcs) - print(f" {n} functions in {pages} pages, {total_ms:.0f}ms " - f"({total_ms / max(n,1):.3f}ms/func)") - - def sz(f): - s = f["size"] - return int(s, 16) if isinstance(s, str) else s - biggest = sorted(funcs, key=sz, reverse=True)[:5] - print(" biggest by bytes:") - for f in biggest: - print(f" {f['addr']:>12} {sz(f):#8x} {f['name']}") - - # ---- 2. Instruction counts of the biggest -------------------------- # - print("\n[2] instruction totals of biggest funcs") - fattest = None - fattest_n = 0 - for f in biggest: - dt, total = ms(lambda f=f: disasm_meta(c, f["addr"])) - print(f" {f['name']:<28} {str(total):>8} insns (meta {dt:.1f}ms)") - if isinstance(total, int) and total > fattest_n: - fattest, fattest_n = f, total - - # ---- 3. Windowed paging INTO the fattest function ------------------ # - print(f"\n[3] window-scroll fattest func {fattest['name']} ({fattest_n} insns)") - WIN = 60 # a viewport - offsets = list(range(0, max(fattest_n - WIN, 1), max((fattest_n // 12), 1))) - times = [] - for off in offsets: - dt, payload = ms(lambda off=off: c.call( - "disasm", addr=fattest["addr"], offset=off, max_instructions=WIN)) - lines = payload.get("asm", {}).get("lines", []) - times.append(dt) - assert len(lines) <= WIN, f"got {len(lines)} > window {WIN}" - print(f" {len(offsets)} windowed reads @win={WIN}: " - f"min={min(times):.1f} med={statistics.median(times):.1f} " - f"max={max(times):.1f} ms (payload capped to <= {WIN} lines)") - - # ---- 3b. O(offset) latency curve (the key hazard) ------------------ # - print("\n[3b] disasm offset-latency curve (offset paging is O(offset))") - for off in [0, 1000, 5000, 20000, min(50000, fattest_n - WIN)]: - if off < 0: - continue - dt, _ = ms(lambda off=off: c.call( - "disasm", addr=fattest["addr"], offset=off, max_instructions=WIN)) - print(f" offset={off:>6}: {dt:6.1f}ms") - - # ---- 4. Simulated 'hold page-down' throughput ---------------------- # - print("\n[4] rapid sequential scroll (hold page-down)") - reads, t0, off = 0, time.time(), 0 - while time.time() - t0 < 2.0: - c.call("disasm", addr=fattest["addr"], offset=off, max_instructions=WIN) - off = (off + WIN) % max(fattest_n - WIN, 1) - reads += 1 - dur = time.time() - t0 - print(f" {reads} windowed reads in {dur:.1f}s = {reads / dur:.0f} reads/s " - f"(~{reads / dur * WIN:.0f} lines/s)") - - # ---- 5. Decompile: truncation AND hard-failure on monsters --------- # - print("\n[5] decompile on fattest func (may fail on huge funcs)") - dt, payload = ms(lambda: c.call("decompile", addr=fattest["addr"])) - code = payload.get("code") if isinstance(payload, dict) else None - if not code: - err = payload.get("error") if isinstance(payload, dict) else payload - print(f" decompile {dt:.0f}ms -> FAILED (soft error, not raised): {err}") - else: - marker = "chars total]" in code[-40:] - print(f" decompile {dt:.0f}ms, returned {len(code)} chars, " - f"server-truncated={marker}") - if marker: - print(f" tail: ...{code[-60:]!r}") - - c.close() - print("\nOK") - return 0 - - -if __name__ == "__main__": - raise SystemExit(main(sys.argv[1:])) |
