From ea22067c52ec335ea4d3a616cb2107d5fdbdf7b6 Mon Sep 17 00:00:00 2001 From: blasty Date: Fri, 21 Aug 2026 12:12:53 +0200 Subject: tests: repair the two seams the ida-nexus port left behind Two scenario monkeypatch sites still hooked client.invoke -- the rename to call() updated the call sites but not the patches, so reprime_is_free and split_view crashed instead of counting. call() takes the remote_ops declaration itself now, so both count by its __name__. remote_ops imported RemoteModule inside _bindings(), which made test_nexus_client (NEEDS_IDA = False) unrunnable under a stdlib-only python3 -- the house rule tests/run.py --fast depends on. The import moves to the same eagerly-if-present, bound-to-None-so-patchable contract nexus_client uses, and the test injects FakeRemoteModule/FakeRemoteError exactly like its other fakes: real names when the library is installed, strict fakes when it is not. --- tests/test_nexus_client.py | 60 ++++++++++++++++++++++++++++++++++++++++++++++ tests/test_scenarios.py | 22 +++++++++-------- 2 files changed, 72 insertions(+), 10 deletions(-) (limited to 'tests') 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, -- cgit v1.3.1-sl0p