diff options
Diffstat (limited to '')
| -rw-r--r-- | idatui/remote_ops.py | 19 | ||||
| -rw-r--r-- | tests/test_nexus_client.py | 60 | ||||
| -rw-r--r-- | tests/test_scenarios.py | 22 |
3 files changed, 89 insertions, 12 deletions
diff --git a/idatui/remote_ops.py b/idatui/remote_ops.py index a67b2b7..cbd1d49 100644 --- a/idatui/remote_ops.py +++ b/idatui/remote_ops.py @@ -11,6 +11,16 @@ from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from ida_domain import Database +# Same contract as nexus_client: ida_nexus is imported eagerly-if-present but +# never at hard import cost, and the name is bound to None rather than left +# undefined so it stays PATCHABLE -- the offline contract tests inject a fake +# RemoteModule here and run this module's binding logic under a stdlib-only +# python3 (tests/run.py --fast). +try: + from ida_nexus import RemoteModule +except ImportError: # library absent: bindings fail actionably on first use + RemoteModule = None # type: ignore[assignment,misc] + def operation_label() -> str: """Display attribution for the current call; ready for per-user context.""" @@ -1655,8 +1665,13 @@ def _bindings() -> dict[Callable[..., Any], Any]: with _BIND_LOCK: if _BOUND is not None: return _BOUND - from ida_nexus import RemoteModule - + # Gated on the binding, not a fresh import, so an injected fake is + # honoured (see the module docstring on the guarded import above). + if RemoteModule is None: + raise ImportError( + "The 'ida-nexus' package is required to execute remote " + "operations but is not installed in this interpreter." + ) operations_module = RemoteModule( Path(__file__), operation_label=operation_label, codec="json" ) diff --git a/tests/test_nexus_client.py b/tests/test_nexus_client.py index 4ebf753..ae2bfa7 100644 --- a/tests/test_nexus_client.py +++ b/tests/test_nexus_client.py @@ -177,6 +177,54 @@ class FakeDisconnected(Exception): """Stand-in for DatabaseDisconnectedError in stdlib-only runs.""" +class FakeRemoteError(Exception): + """Stand-in for RemoteError: (code, message, status, details).""" + + def __init__(self, code, message, status=500, details=None): + super().__init__(message) + self.code = code + self.status = status + self.details = details or {} + + +class FakeRemoteModule: + """Stand-in for ida_nexus.RemoteModule in stdlib-only runs. + + Speaks the same two-step wire contract FakeHandle.execute_python answers: + install the module's real source once, then send each call as a snippet + that looks the function up in the installed module registry (the + ``.modules.get(`` marker the fake keys on), carrying the operation label. + """ + + def __init__(self, path, *, operation_label=None, codec="json"): + with open(path) as file: + self._source = file.read() + self._label = operation_label + self._installed = False + + def function(self, declaration, timeout=None): + name = getattr(declaration, "__name__", str(declaration)) + + def remote(handle, **args): + label = self._label() if callable(self._label) else self._label + if not self._installed: + handle.execute_python(self._source, operation_label=label) + self._installed = True + response = handle.execute_python( + f"__mod = __registry.modules.get(...) # call {name}", + operation_label=label, + ) + result = response["result"] + if ( + isinstance(result, dict) + and result.get("__remote_ida_status__") == "ok" + ): + return result.get("__remote_ida_value__") + raise module.RemoteError(name, f"remote call failed: {result!r}", 500) + + return remote + + def _open_kwargs_are_real(sent: dict): """(ok, detail) for the kwargs the adapter passes to DatabaseHandle.open. @@ -230,9 +278,17 @@ def main() -> int: original_options = module.DatabaseOpenOptions original_busy = module.DatabaseBusyError original_disconnected = module.DatabaseDisconnectedError + original_remote_error = module.RemoteError module.DatabaseOpenOptions = original_options or FakeOpenOptions module.DatabaseBusyError = original_busy or FakeBusy module.DatabaseDisconnectedError = original_disconnected or FakeDisconnected + module.RemoteError = original_remote_error or FakeRemoteError + # The binding seam in remote_ops, same rule: the real RemoteModule when the + # library is installed, this file's fake otherwise -- and the lazy binding + # cache reset around it so this run binds through whichever is active. + original_remote_module = remote_ops.RemoteModule + remote_ops.RemoteModule = original_remote_module or FakeRemoteModule + remote_ops._BOUND = None try: with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "sample.bin") @@ -416,6 +472,7 @@ def main() -> int: module.DatabaseOpenOptions = original_options module.DatabaseBusyError = original_busy module.DatabaseDisconnectedError = original_disconnected + module.RemoteError = original_remote_error client = NexusClient(__file__) @@ -431,6 +488,9 @@ def main() -> int: ) else: check("unknown adapter operations are explicit", False) + finally: + remote_ops.RemoteModule = original_remote_module + remote_ops._BOUND = None print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0 diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py index 190777c..30d6087 100644 --- a/tests/test_scenarios.py +++ b/tests/test_scenarios.py @@ -585,14 +585,15 @@ async def s_reprime_is_free(c: Ctx): await c.wait(lambda: lv.model.complete, 30) client = app.program.client - original = type(client).invoke + original = type(client).call seen: list[str] = [] def counting(self, operation, *a, **kw): - seen.append(operation) + # call() takes the remote_ops declaration itself; count by its name. + seen.append(getattr(operation, "__name__", str(operation))) return original(self, operation, *a, **kw) - type(client).invoke = counting + type(client).call = counting try: for _ in range(3): # decomp and back, three times await c.press("tab") @@ -600,7 +601,7 @@ async def s_reprime_is_free(c: Ctx): await c.press("tab") await c.pause(0.05) finally: - type(client).invoke = original + type(client).call = original rebuilds = seen.count("segment_index") c.check( @@ -1321,18 +1322,19 @@ async def s_split_view(c: Ctx): # 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 + _orig_call = c.prog.client.call - def _counting(name, *a, **kw): - if name == "lookup_funcs": + def _counting(operation, *a, **kw): + # call() takes the remote_ops declaration itself; match by its name. + if getattr(operation, "__name__", "") == "lookup_funcs": _lookups["n"] += 1 - return _orig_call(name, *a, **kw) + return _orig_call(operation, *a, **kw) - c.prog.client.invoke = _counting + c.prog.client.call = _counting try: await _split_view_body(c, app, lst, dec) finally: - c.prog.client.invoke = _orig_call + c.prog.client.call = _orig_call c.check( "split view doesn't storm the worker with function lookups", _lookups["n"] < 500, |