From f4c1d9b5497fd0d38137b6b345e8171b307c1117 Mon Sep 17 00:00:00 2001 From: Duncan Ogilvie Date: Thu, 20 Aug 2026 23:55:52 +0200 Subject: Switch to ida-nexus --- tests/_fixtures.py | 2 +- tests/run.py | 4 +- tests/test_codemode_client.py | 440 ------------------------------------------ tests/test_kittygfx.py | 2 +- tests/test_launch.py | 6 +- tests/test_nexus_client.py | 440 ++++++++++++++++++++++++++++++++++++++++++ tests/test_pool.py | 4 +- tests/test_project.py | 2 +- tests/test_scenarios.py | 2 +- tests/test_thumb_ui.py | 2 +- tests/test_trace_ui.py | 2 +- 11 files changed, 453 insertions(+), 453 deletions(-) delete mode 100644 tests/test_codemode_client.py create mode 100644 tests/test_nexus_client.py (limited to 'tests') diff --git a/tests/_fixtures.py b/tests/_fixtures.py index 16192fa..bc10235 100644 --- a/tests/_fixtures.py +++ b/tests/_fixtures.py @@ -133,7 +133,7 @@ async def build_pristine(binary: str, cache: str, app_factory) -> None: break app.program.client.save_database() # Textual's headless run_test context does not reliably emit App.Unmount on - # every platform/version; release the Code Mode lease explicitly. + # every platform/version; release the IDA Nexus lease explicitly. if app.program is not None: app.program.close() if app.client is not None: diff --git a/tests/run.py b/tests/run.py index b38a707..5d552bb 100755 --- a/tests/run.py +++ b/tests/run.py @@ -51,8 +51,8 @@ import time ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) TESTS = os.path.join(ROOT, "tests") -#: The IDA-capable interpreter. The pilot tests need textual AND the Code Mode -#: library in one python; the database process is Code Mode's to place. +#: The IDA-capable interpreter. The pilot tests need textual AND the IDA Nexus +#: library in one python; the database process is IDA Nexus's to place. DEFAULT_PY = os.path.expanduser("~/ida-venv/bin/python") #: Both shapes the suites print: "N passed, M failed" and "N checks, M failed". diff --git a/tests/test_codemode_client.py b/tests/test_codemode_client.py deleted file mode 100644 index dd0ae08..0000000 --- a/tests/test_codemode_client.py +++ /dev/null @@ -1,440 +0,0 @@ -"""IDA-free contract tests for the Code Mode client adapter.""" - -from __future__ import annotations - -import os -import queue -import sys -import threading -import time -import tempfile -from dataclasses import dataclass - -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from idatui.errors import IDAConnectionError, IDAToolError # noqa: E402 -from idatui import remote_ops # noqa: E402 -import idatui.codemode_client as module # noqa: E402 -from idatui.codemode_client import CodeModeClient, _parse_load_args # noqa: E402 - -#: Pure: fakes the DatabaseHandle, never touches IDA or the Code Mode library. -NEEDS_IDA = False - -PASS = FAIL = 0 - - -def check(name: str, condition: bool, detail="") -> None: - global PASS, FAIL - if condition: - PASS += 1 - print(f" ok {name}") - else: - FAIL += 1 - print(f" FAIL {name} {detail}") - - -@dataclass(frozen=True) -class FakeEntry: - pid: int = 123 - backend: str = "gui" - record_id: str = "123-abcdef" - exe_path: str = "" - idb_path: str = "" - managed: bool = False - - -_CLOSED = object() - - -class FakeSubscription: - def __init__(self) -> None: - self._queue: queue.Queue = queue.Queue() - self.closed = False - - def __iter__(self): - return self - - def __next__(self): - item = self._queue.get() - if isinstance(item, BaseException): - raise item - if item is _CLOSED: - raise StopIteration - return item - - def emit(self, event: dict) -> None: - self._queue.put(event) - - def close(self) -> None: - if not self.closed: - self.closed = True - self._queue.put(_CLOSED) - - -class FakeHandle: - def __init__(self, path: str) -> None: - self.connected = True - self.instance = FakeEntry(exe_path=path, idb_path=path + ".i64") - self.waited = None - self.saved = 0 - self.closed = False - self.code = "" - self.codes = [] - self.code_timeout = None - self.operation_label = None - self.event_origin_id = "fake-handle-origin" - self.owns_checks = 0 - self.subscription = FakeSubscription() - self.shutdown_calls = [] - self.shutdown_error = None - - def wait_autoanalysis(self, timeout=None): - self.waited = timeout - return {"complete": True, "status": "complete"} - - def execute_python( - self, - code, - timeout=None, - *, - operation_id=None, - operation_label=None, - persist_globals=False, - filename=None, - ): - self.code = code - self.codes.append(code) - self.code_timeout = timeout - self.operation_label = operation_label - result = ( - { - "__remote_ida_status__": "ok", - "__remote_ida_value__": {"sentinel": 7}, - } - if ".modules.get(" in code - else True - ) - return {"result": result, "stdout": "", "stderr": ""} - - def subscribe_idb_events(self): - return self.subscription - - def owns_event(self, event): - self.owns_checks += 1 - return event.get("origin_id") == self.event_origin_id - - def save_database(self): - self.saved += 1 - return {"saved": True, "idb_path": self.instance.idb_path} - - def shutdown_database(self, *, save=True): - self.shutdown_calls.append(save) - if self.shutdown_error is not None: - raise module.RemoteError(self.shutdown_error, self.shutdown_error, 409) - return {"shutting_down": True, "save": save} - - def close(self): - self.subscription.close() - self.connected = False - self.closed = True - - -class FakeDatabaseHandle: - opened = None - opens = 0 - kwargs = None - - @classmethod - def open(cls, path, **kwargs): - cls.opens += 1 - cls.opened = path - cls.kwargs = kwargs - return FakeHandle(path) - - -@dataclass(frozen=True) -class FakeOpenOptions: - """Stand-in for DatabaseOpenOptions when the library is not installed. - - Deliberately STRICT (no **kwargs): an option the adapter invents would - raise here, and `_option_fields_are_real` checks the surviving names - against the real dataclass wherever it is importable. - """ - - spawn: bool = True - startup_timeout: float = 120.0 - output_database: str | None = None - processor: str | None = None - image_base: int | None = None - file_type: str | None = None - new_database: bool = False - - -class FakeBusy(Exception): - """Stand-in for DatabaseBusyError: `except None` is a TypeError.""" - - -class FakeDisconnected(Exception): - """Stand-in for DatabaseDisconnectedError in stdlib-only runs.""" - - -def _open_kwargs_are_real(sent: dict): - """(ok, detail) for the kwargs the adapter passes to DatabaseHandle.open. - - Skips (passes) when ida_codemode is not installed, so the file stays pure. - """ - try: - import inspect - from ida_codemode import DatabaseHandle as Real - except ImportError: - return True, "ida_codemode not installed - signature not checked" - accepted = set(inspect.signature(Real.open).parameters) - unknown = sorted(set(sent) - accepted) - return not unknown, f"open() rejects {unknown}" - - -def _option_fields_are_real(options): - """(ok, detail) for the option names the adapter fills in. - - The open() signature no longer names the loader options -- they moved - inside DatabaseOpenOptions -- so the `loading_address` class of bug now - hides there instead. Check it in the same way. - """ - try: - import dataclasses - from ida_codemode import DatabaseOpenOptions as Real - except ImportError: - return True, "ida_codemode not installed - fields not checked" - accepted = {field.name for field in dataclasses.fields(Real)} - unknown = sorted({f.name for f in dataclasses.fields(options)} - accepted) - return not unknown, f"DatabaseOpenOptions rejects {unknown}" - - -def main() -> int: - proc, base, file_type = _parse_load_args("-parm:ARMv7-M -b800000 -TRaw") - check( - "legacy switches map to typed Code Mode options", - (proc, base, file_type) == ("arm:ARMv7-M", 0x8000000, "Raw"), - (proc, base, file_type), - ) - try: - _parse_load_args("-parm -zcustom") - except ValueError as exc: - check("arbitrary IDA switches fail loudly", "cannot represent" in str(exc), exc) - else: - check("arbitrary IDA switches fail loudly", False) - - original = module.DatabaseHandle - module.DatabaseHandle = FakeDatabaseHandle - # The library's own names when it is installed; strict fakes when it is not - # (this file must keep running under a stdlib-only python3). - original_options = module.DatabaseOpenOptions - original_busy = module.DatabaseBusyError - original_disconnected = module.DatabaseDisconnectedError - module.DatabaseOpenOptions = original_options or FakeOpenOptions - module.DatabaseBusyError = original_busy or FakeBusy - module.DatabaseDisconnectedError = original_disconnected or FakeDisconnected - try: - with tempfile.TemporaryDirectory() as tmp: - path = os.path.join(tmp, "sample.bin") - with open(path, "wb") as file: - file.write(b"sample") - client = CodeModeClient(path, load_args="-parm:ARMv7-A -b100") - notes = [] - client.connect(timeout=42, progress=notes.append) - handle = client._handle - check( - "connect delegates database discovery to DatabaseHandle.open", - FakeDatabaseHandle.opened == path and handle is not None, - ) - options = FakeDatabaseHandle.kwargs["options"] - check( - "typed loader options cross the dependency boundary", - options.processor == "arm:ARMv7-A" and options.image_base == 0x1000, - options, - ) - check( - "every open option exists in the real library", - *_option_fields_are_real(options), - ) - # A fake that swallows **kwargs cannot catch a keyword the real - # library does not have -- which is exactly how this port shipped - # `loading_address` (the real name is `image_base`) and would have - # raised TypeError on the very first connect. Check the names we - # send against the real signature whenever it is importable. - check( - "every open() keyword exists in the real library", - *_open_kwargs_are_real(FakeDatabaseHandle.kwargs), - ) - check( - "connect waits for Code Mode autoanalysis", - handle.waited == 42, - getattr(handle, "waited", None), - ) - check( - "progress distinguishes discovery and backend attachment", - len(notes) == 2 and "gui" in notes[-1], - notes, - ) - result = client.call( - remote_ops.list_funcs, queries=[{"offset": 0, "count": 2}] - ) - check( - "remote operation returns its JSON result", - result == {"sentinel": 7}, - result, - ) - check( - "operation source is real Python installed through ida-domain", - any("db.functions.get_all()" in code for code in handle.codes), - handle.codes[0][:200], - ) - check( - "remote operations attribute IDB events to IDA TUI", - handle.operation_label == "IDA TUI", - handle.operation_label, - ) - batches = [] - delivered = threading.Event() - - def changed(batch): - batches.append(batch) - delivered.set() - - watcher = client.watch_idb_events(changed, debounce=0.05) - handle.subscription.emit({"event_name": "renamed", "origin_id": "peer-1"}) - handle.subscription.emit( - {"event_name": "cmt_changed", "origin_id": "peer-2"} - ) - check( - "event bursts produce one debounced refresh", - delivered.wait(1) and len(batches) == 1 and len(batches[0]) == 2, - batches, - ) - delivered.clear() - handle.subscription.emit( - {"event_name": "renamed", "origin_id": handle.event_origin_id} - ) - time.sleep(0.1) - check( - "the listener uses handle ownership to ignore its own events", - not delivered.is_set() - and len(batches) == 1 - and handle.owns_checks >= 3, - (batches, handle.owns_checks), - ) - handle.subscription.emit( - {"event_name": "byte_patched", "origin_id": "peer-3"} - ) - watcher.close() - time.sleep(0.1) - check( - "closing drops a pending debounced refresh", - not delivered.is_set() and len(batches) == 1, - batches, - ) - check( - "health exposes registry identity", - client.health()["record_id"] == "123-abcdef", - ) - client.save_database() - check("save uses the public Code Mode save route", handle.saved == 1) - check( - "GUI leases transfer rather than claiming discard", - client.discard_database() is False and handle.shutdown_calls == [], - handle.shutdown_calls, - ) - handle.instance = FakeEntry( - backend="idalib", managed=True, exe_path=path, idb_path=path + ".i64" - ) - check( - "a final managed lease discards without saving", - client.discard_database() is True and handle.shutdown_calls == [False], - handle.shutdown_calls, - ) - handle.shutdown_error = "instance_shared" - check( - "a shared managed lease transfers finalization", - client.discard_database() is False, - handle.shutdown_calls, - ) - handle.shutdown_error = "instance_busy" - try: - client.discard_database(timeout=0) - except IDAToolError as exc: - check( - "a busy final lease never silently saves", - exc.tool == "shutdown_database", - exc, - ) - else: - check("a busy final lease never silently saves", False) - handle.shutdown_error = None - handle.instance = FakeEntry(exe_path=path, idb_path=path + ".i64") - opens = FakeDatabaseHandle.opens - handle.connected = False - try: - client.health() - except IDAConnectionError as exc: - check( - "a disconnected handle requires explicit rediscovery", - "explicit rediscovery" in str(exc) - and FakeDatabaseHandle.opens == opens, - (exc, FakeDatabaseHandle.opens, opens), - ) - else: - check("a disconnected handle requires explicit rediscovery", False) - client.close() - check("close releases only the handle lease", handle.closed) - check( - "GUI lifetime is never claimed by the client", - client.wait_released(0) is False, - ) - disconnected = CodeModeClient(path).connect() - stream_errors = [] - stream_failed = threading.Event() - - def failed(error): - stream_errors.append(error) - stream_failed.set() - - stream_watch = disconnected.watch_idb_events( - lambda _batch: None, on_error=failed, debounce=0 - ) - disconnected._handle.subscription.emit( - module.DatabaseDisconnectedError("GUI database closed") - ) - check( - "stream disconnects become application connection errors", - stream_failed.wait(1) - and isinstance(stream_errors[0], IDAConnectionError), - stream_errors, - ) - stream_watch.close() - disconnected.close() - finally: - module.DatabaseHandle = original - module.DatabaseOpenOptions = original_options - module.DatabaseBusyError = original_busy - module.DatabaseDisconnectedError = original_disconnected - - client = CodeModeClient(__file__) - - def unknown_operation(): - pass - - try: - client.call(unknown_operation) - except IDAToolError as exc: - check( - "unknown adapter operations are explicit", - exc.tool == "unknown_operation", - ) - else: - check("unknown adapter operations are explicit", False) - - print(f"\n{PASS} passed, {FAIL} failed") - return 1 if FAIL else 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/tests/test_kittygfx.py b/tests/test_kittygfx.py index 74b6bb7..cfc6512 100644 --- a/tests/test_kittygfx.py +++ b/tests/test_kittygfx.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Cross-platform checks for the optional kitty-graphics startup splash. -Pure: stdlib only, no terminal, Textual, Code Mode, or IDA. +Pure: stdlib only, no terminal, Textual, IDA Nexus, or IDA. """ from __future__ import annotations diff --git a/tests/test_launch.py b/tests/test_launch.py index d57ee5f..35a0b8d 100644 --- a/tests/test_launch.py +++ b/tests/test_launch.py @@ -3,12 +3,12 @@ The old `_sweep_locks` deleted `.id0/.id1/.id2/.nam/.til` next to the user's binary when a database failed to open. That was only defensible while the TUI -exclusively owned a private worker; under Code Mode a GUI or another client may +exclusively owned a private worker; under IDA Nexus a GUI or another client may own the database, so the sweep is gone. Its tests are replaced by one that keeps it gone -- deleting a shared database's working files is unrecoverable, and this is the cheapest guard against someone reintroducing the "helpful" cleanup. -Pure: no IDA, no Code Mode library, no Textual. +Pure: no IDA, no IDA Nexus library, no Textual. """ from __future__ import annotations @@ -47,7 +47,7 @@ def touch(*paths): def t_no_lock_sweeping(): """The launcher must not delete database working files any more. - Code Mode's registry locks, health probes and IDA itself arbitrate database + IDA Nexus's registry locks, health probes and IDA itself arbitrate database ownership now. A sweep here would delete files out from under a live GUI. """ check("_sweep_locks is gone", not hasattr(launch, "_sweep_locks")) diff --git a/tests/test_nexus_client.py b/tests/test_nexus_client.py new file mode 100644 index 0000000..4ebf753 --- /dev/null +++ b/tests/test_nexus_client.py @@ -0,0 +1,440 @@ +"""IDA-free contract tests for the IDA Nexus client adapter.""" + +from __future__ import annotations + +import os +import queue +import sys +import threading +import time +import tempfile +from dataclasses import dataclass + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from idatui.errors import IDAConnectionError, IDAToolError # noqa: E402 +from idatui import remote_ops # noqa: E402 +import idatui.nexus_client as module # noqa: E402 +from idatui.nexus_client import NexusClient, _parse_load_args # noqa: E402 + +#: Pure: fakes the DatabaseHandle, never touches IDA or the IDA Nexus library. +NEEDS_IDA = False + +PASS = FAIL = 0 + + +def check(name: str, condition: bool, detail="") -> None: + global PASS, FAIL + if condition: + PASS += 1 + print(f" ok {name}") + else: + FAIL += 1 + print(f" FAIL {name} {detail}") + + +@dataclass(frozen=True) +class FakeEntry: + pid: int = 123 + backend: str = "gui" + record_id: str = "123-abcdef" + exe_path: str = "" + idb_path: str = "" + managed: bool = False + + +_CLOSED = object() + + +class FakeSubscription: + def __init__(self) -> None: + self._queue: queue.Queue = queue.Queue() + self.closed = False + + def __iter__(self): + return self + + def __next__(self): + item = self._queue.get() + if isinstance(item, BaseException): + raise item + if item is _CLOSED: + raise StopIteration + return item + + def emit(self, event: dict) -> None: + self._queue.put(event) + + def close(self) -> None: + if not self.closed: + self.closed = True + self._queue.put(_CLOSED) + + +class FakeHandle: + def __init__(self, path: str) -> None: + self.connected = True + self.instance = FakeEntry(exe_path=path, idb_path=path + ".i64") + self.waited = None + self.saved = 0 + self.closed = False + self.code = "" + self.codes = [] + self.code_timeout = None + self.operation_label = None + self.event_origin_id = "fake-handle-origin" + self.owns_checks = 0 + self.subscription = FakeSubscription() + self.shutdown_calls = [] + self.shutdown_error = None + + def wait_autoanalysis(self, timeout=None): + self.waited = timeout + return {"complete": True, "status": "complete"} + + def execute_python( + self, + code, + timeout=None, + *, + operation_id=None, + operation_label=None, + persist_globals=False, + filename=None, + ): + self.code = code + self.codes.append(code) + self.code_timeout = timeout + self.operation_label = operation_label + result = ( + { + "__remote_ida_status__": "ok", + "__remote_ida_value__": {"sentinel": 7}, + } + if ".modules.get(" in code + else True + ) + return {"result": result, "stdout": "", "stderr": ""} + + def subscribe_idb_events(self): + return self.subscription + + def owns_event(self, event): + self.owns_checks += 1 + return event.get("origin_id") == self.event_origin_id + + def save_database(self): + self.saved += 1 + return {"saved": True, "idb_path": self.instance.idb_path} + + def shutdown_database(self, *, save=True): + self.shutdown_calls.append(save) + if self.shutdown_error is not None: + raise module.RemoteError(self.shutdown_error, self.shutdown_error, 409) + return {"shutting_down": True, "save": save} + + def close(self): + self.subscription.close() + self.connected = False + self.closed = True + + +class FakeDatabaseHandle: + opened = None + opens = 0 + kwargs = None + + @classmethod + def open(cls, path, **kwargs): + cls.opens += 1 + cls.opened = path + cls.kwargs = kwargs + return FakeHandle(path) + + +@dataclass(frozen=True) +class FakeOpenOptions: + """Stand-in for DatabaseOpenOptions when the library is not installed. + + Deliberately STRICT (no **kwargs): an option the adapter invents would + raise here, and `_option_fields_are_real` checks the surviving names + against the real dataclass wherever it is importable. + """ + + spawn: bool = True + startup_timeout: float = 120.0 + output_database: str | None = None + processor: str | None = None + image_base: int | None = None + file_type: str | None = None + new_database: bool = False + + +class FakeBusy(Exception): + """Stand-in for DatabaseBusyError: `except None` is a TypeError.""" + + +class FakeDisconnected(Exception): + """Stand-in for DatabaseDisconnectedError in stdlib-only runs.""" + + +def _open_kwargs_are_real(sent: dict): + """(ok, detail) for the kwargs the adapter passes to DatabaseHandle.open. + + Skips (passes) when ida_nexus is not installed, so the file stays pure. + """ + try: + import inspect + from ida_nexus import DatabaseHandle as Real + except ImportError: + return True, "ida_nexus not installed - signature not checked" + accepted = set(inspect.signature(Real.open).parameters) + unknown = sorted(set(sent) - accepted) + return not unknown, f"open() rejects {unknown}" + + +def _option_fields_are_real(options): + """(ok, detail) for the option names the adapter fills in. + + The open() signature no longer names the loader options -- they moved + inside DatabaseOpenOptions -- so the `loading_address` class of bug now + hides there instead. Check it in the same way. + """ + try: + import dataclasses + from ida_nexus import DatabaseOpenOptions as Real + except ImportError: + return True, "ida_nexus not installed - fields not checked" + accepted = {field.name for field in dataclasses.fields(Real)} + unknown = sorted({f.name for f in dataclasses.fields(options)} - accepted) + return not unknown, f"DatabaseOpenOptions rejects {unknown}" + + +def main() -> int: + proc, base, file_type = _parse_load_args("-parm:ARMv7-M -b800000 -TRaw") + check( + "legacy switches map to typed IDA Nexus options", + (proc, base, file_type) == ("arm:ARMv7-M", 0x8000000, "Raw"), + (proc, base, file_type), + ) + try: + _parse_load_args("-parm -zcustom") + except ValueError as exc: + check("arbitrary IDA switches fail loudly", "cannot represent" in str(exc), exc) + else: + check("arbitrary IDA switches fail loudly", False) + + original = module.DatabaseHandle + module.DatabaseHandle = FakeDatabaseHandle + # The library's own names when it is installed; strict fakes when it is not + # (this file must keep running under a stdlib-only python3). + original_options = module.DatabaseOpenOptions + original_busy = module.DatabaseBusyError + original_disconnected = module.DatabaseDisconnectedError + module.DatabaseOpenOptions = original_options or FakeOpenOptions + module.DatabaseBusyError = original_busy or FakeBusy + module.DatabaseDisconnectedError = original_disconnected or FakeDisconnected + try: + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "sample.bin") + with open(path, "wb") as file: + file.write(b"sample") + client = NexusClient(path, load_args="-parm:ARMv7-A -b100") + notes = [] + client.connect(timeout=42, progress=notes.append) + handle = client._handle + check( + "connect delegates database discovery to DatabaseHandle.open", + FakeDatabaseHandle.opened == path and handle is not None, + ) + options = FakeDatabaseHandle.kwargs["options"] + check( + "typed loader options cross the dependency boundary", + options.processor == "arm:ARMv7-A" and options.image_base == 0x1000, + options, + ) + check( + "every open option exists in the real library", + *_option_fields_are_real(options), + ) + # A fake that swallows **kwargs cannot catch a keyword the real + # library does not have -- which is exactly how this port shipped + # `loading_address` (the real name is `image_base`) and would have + # raised TypeError on the very first connect. Check the names we + # send against the real signature whenever it is importable. + check( + "every open() keyword exists in the real library", + *_open_kwargs_are_real(FakeDatabaseHandle.kwargs), + ) + check( + "connect waits for IDA Nexus autoanalysis", + handle.waited == 42, + getattr(handle, "waited", None), + ) + check( + "progress distinguishes discovery and backend attachment", + len(notes) == 2 and "gui" in notes[-1], + notes, + ) + result = client.call( + remote_ops.list_funcs, queries=[{"offset": 0, "count": 2}] + ) + check( + "remote operation returns its JSON result", + result == {"sentinel": 7}, + result, + ) + check( + "operation source is real Python installed through ida-domain", + any("db.functions.get_all()" in code for code in handle.codes), + handle.codes[0][:200], + ) + check( + "remote operations attribute IDB events to IDA TUI", + handle.operation_label == "IDA TUI", + handle.operation_label, + ) + batches = [] + delivered = threading.Event() + + def changed(batch): + batches.append(batch) + delivered.set() + + watcher = client.watch_idb_events(changed, debounce=0.05) + handle.subscription.emit({"event_name": "renamed", "origin_id": "peer-1"}) + handle.subscription.emit( + {"event_name": "cmt_changed", "origin_id": "peer-2"} + ) + check( + "event bursts produce one debounced refresh", + delivered.wait(1) and len(batches) == 1 and len(batches[0]) == 2, + batches, + ) + delivered.clear() + handle.subscription.emit( + {"event_name": "renamed", "origin_id": handle.event_origin_id} + ) + time.sleep(0.1) + check( + "the listener uses handle ownership to ignore its own events", + not delivered.is_set() + and len(batches) == 1 + and handle.owns_checks >= 3, + (batches, handle.owns_checks), + ) + handle.subscription.emit( + {"event_name": "byte_patched", "origin_id": "peer-3"} + ) + watcher.close() + time.sleep(0.1) + check( + "closing drops a pending debounced refresh", + not delivered.is_set() and len(batches) == 1, + batches, + ) + check( + "health exposes registry identity", + client.health()["record_id"] == "123-abcdef", + ) + client.save_database() + check("save uses the public IDA Nexus save route", handle.saved == 1) + check( + "GUI leases transfer rather than claiming discard", + client.discard_database() is False and handle.shutdown_calls == [], + handle.shutdown_calls, + ) + handle.instance = FakeEntry( + backend="idalib", managed=True, exe_path=path, idb_path=path + ".i64" + ) + check( + "a final managed lease discards without saving", + client.discard_database() is True and handle.shutdown_calls == [False], + handle.shutdown_calls, + ) + handle.shutdown_error = "instance_shared" + check( + "a shared managed lease transfers finalization", + client.discard_database() is False, + handle.shutdown_calls, + ) + handle.shutdown_error = "instance_busy" + try: + client.discard_database(timeout=0) + except IDAToolError as exc: + check( + "a busy final lease never silently saves", + exc.tool == "shutdown_database", + exc, + ) + else: + check("a busy final lease never silently saves", False) + handle.shutdown_error = None + handle.instance = FakeEntry(exe_path=path, idb_path=path + ".i64") + opens = FakeDatabaseHandle.opens + handle.connected = False + try: + client.health() + except IDAConnectionError as exc: + check( + "a disconnected handle requires explicit rediscovery", + "explicit rediscovery" in str(exc) + and FakeDatabaseHandle.opens == opens, + (exc, FakeDatabaseHandle.opens, opens), + ) + else: + check("a disconnected handle requires explicit rediscovery", False) + client.close() + check("close releases only the handle lease", handle.closed) + check( + "GUI lifetime is never claimed by the client", + client.wait_released(0) is False, + ) + disconnected = NexusClient(path).connect() + stream_errors = [] + stream_failed = threading.Event() + + def failed(error): + stream_errors.append(error) + stream_failed.set() + + stream_watch = disconnected.watch_idb_events( + lambda _batch: None, on_error=failed, debounce=0 + ) + disconnected._handle.subscription.emit( + module.DatabaseDisconnectedError("GUI database closed") + ) + check( + "stream disconnects become application connection errors", + stream_failed.wait(1) + and isinstance(stream_errors[0], IDAConnectionError), + stream_errors, + ) + stream_watch.close() + disconnected.close() + finally: + module.DatabaseHandle = original + module.DatabaseOpenOptions = original_options + module.DatabaseBusyError = original_busy + module.DatabaseDisconnectedError = original_disconnected + + client = NexusClient(__file__) + + def unknown_operation(): + pass + + try: + client.call(unknown_operation) + except IDAToolError as exc: + check( + "unknown adapter operations are explicit", + exc.tool == "unknown_operation", + ) + else: + check("unknown adapter operations are explicit", False) + + print(f"\n{PASS} passed, {FAIL} failed") + return 1 if FAIL else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_pool.py b/tests/test_pool.py index 6309e13..37058bf 100644 --- a/tests/test_pool.py +++ b/tests/test_pool.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Unit tests for idatui.pool (Code Mode lease residency and LRU budget). +"""Unit tests for idatui.pool (IDA Nexus lease residency and LRU budget). A fake client keeps the policy testable without IDA or Textual. @@ -31,7 +31,7 @@ def check(name, cond, detail=""): class FakeClient: - """Stands in for a CodeModeClient lease and records saves/closes.""" + """Stands in for a NexusClient lease and records saves/closes.""" def __init__(self, ref, mem=100, backend="idalib", discardable=True): diff --git a/tests/test_project.py b/tests/test_project.py index 690f360..bae0802 100644 --- a/tests/test_project.py +++ b/tests/test_project.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Unit tests for idatui.project (the multi-binary project model + staging). -IDA-free: exercises staging plus Code Mode ownership checks without opening a database. +IDA-free: exercises staging plus IDA Nexus ownership checks without opening a database. python tests/test_project.py """ diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py index 45ed51b..190777c 100644 --- a/tests/test_scenarios.py +++ b/tests/test_scenarios.py @@ -5272,7 +5272,7 @@ async def run(binary, only=None): async def _run_on(binary, only=None): - # Code Mode attaches a registered GUI or starts/reuses a managed worker. + # IDA Nexus 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) diff --git a/tests/test_thumb_ui.py b/tests/test_thumb_ui.py index f12fbc4..1b4c438 100644 --- a/tests/test_thumb_ui.py +++ b/tests/test_thumb_ui.py @@ -50,7 +50,7 @@ def check(name, ok, detail=""): #: #: This suite used to delete .i64 and reopen the SAME path for each phase. #: That was safe when the TUI owned a private worker that died with it; under -#: Code Mode the database is leased and the previous phase's worker can still +#: IDA Nexus the database is leased and the previous phase's worker can still #: hold it through its lease grace, so the delete raced a live owner and the #: next open never produced a listing (the crash this fixed). Separate paths #: cannot collide, and nothing has to wait for anyone else to let go. diff --git a/tests/test_trace_ui.py b/tests/test_trace_ui.py index 855ce4d..586ce2e 100644 --- a/tests/test_trace_ui.py +++ b/tests/test_trace_ui.py @@ -115,7 +115,7 @@ async def run() -> int: # `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 # runs in a worker. Waiting on it and then reading the cursor was a - # race that the (slower) Code Mode backend loses. Gate on the thing + # race that the (slower) IDA Nexus backend loses. Gate on the thing # the check is about. await settle(app, lambda: app._t == 1 and lst._cursor_ea() == t.ip(1), timeout=20) -- cgit v1.3.1-sl0p