From c87c842e5e4a2062ad07c1acbba6e2c809c61024 Mon Sep 17 00:00:00 2001 From: Jeremy Howard Date: Sun, 30 Aug 2026 20:20:14 +1000 Subject: [PATCH] Make kernel integration tests deterministic --- DEV.md | 6 +- pyproject.toml | 1 + tests/client.py | 34 --- tests/echo_kernel.py | 31 +++ tests/ipython_kernel.py | 20 ++ tests/test_kernel_echo.py | 367 ++++++++++-------------------- tests/test_rust_ipython_kernel.py | 273 ++++++++-------------- tests/test_rust_kernel_echo.py | 96 ++------ 8 files changed, 300 insertions(+), 528 deletions(-) delete mode 100644 tests/client.py create mode 100644 tests/echo_kernel.py create mode 100644 tests/ipython_kernel.py diff --git a/DEV.md b/DEV.md index 155cef2..f32bc0f 100644 --- a/DEV.md +++ b/DEV.md @@ -92,8 +92,10 @@ The Python wrapper may place a standalone kernel in its own process group. On sh - a pure Rust echo language, implemented entirely in the example binary, through the crate API; - an IPython shell through the Python adapter. -Standalone Rust tests cover wire framing and language interruption primitives. A Python integration test drives a real debugpy session through the DAP -transport. ipymini's complete protocol and behavior suite is kernmini's main integration test. +`ConKernelClient` launches each kernel and manages its Jupyter requests, replies, IOPub messages, stdin, and shutdown. Tests use live protocol events to +synchronize concurrent behavior rather than sleeps or hand-written socket draining. Standalone Rust tests cover wire framing and language interruption +primitives. A Python integration test drives a real debugpy session through the DAP transport directly, since that transport is the subject of the test. +ipymini's complete protocol and behavior suite is kernmini's main integration test. ```bash cd ../ipymini diff --git a/pyproject.toml b/pyproject.toml index a41ee95..d390426 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,6 +19,7 @@ dependencies = ["fastcore>=2.2.17"] [project.optional-dependencies] dev = [ + "conkernelclient>=0.0.23", "debugpy>=1.8.21", "fastship>=0.0.11", "ipymini>=0.1.20", diff --git a/tests/client.py b/tests/client.py deleted file mode 100644 index 5c6a3bd..0000000 --- a/tests/client.py +++ /dev/null @@ -1,34 +0,0 @@ -"Test client using jupywire message framing over pyzmq sockets." - -import zmq -from jupywire.session import Session - - -class MiniSession(Session): - "jupywire `Session` with kernel defaults, plus zmq socket `send` and `recv`." - - def __init__(self, key: bytes = b"", signature_scheme: str = "hmac-sha256", username: str = "kernel", - session: str | None = None, digest_history_size: int = 2**16): - super().__init__(key=key, signature_scheme=signature_scheme, username=username, session=session, - digest_history_size=digest_history_size) - - def send(self, stream, msg_or_type, content: dict | None = None, parent: dict | None = None, - ident: bytes | list[bytes] | None = None, buffers: list | None = None, metadata: dict | None = None) -> dict: - "Build (or take) a message, serialize it, and send it with optional raw buffer frames appended." - if isinstance(msg_or_type, dict): - msg = msg_or_type - buffers = buffers or msg.get("buffers", []) - else: msg = self.msg(msg_or_type, content=content, parent=parent, metadata=metadata) - to_send = self.serialize(msg, ident) - to_send.extend(buffers or []) - stream.send_multipart(to_send, copy=True) - return msg - - def recv(self, socket, mode: int = zmq.NOBLOCK, content: bool = True, copy: bool = True): - "Receive and unpack a message; returns (idents, msg), or (None, None) when nothing is waiting." - try: msg_list = socket.recv_multipart(mode, copy=copy) - except zmq.ZMQError as e: - if e.errno == zmq.EAGAIN: return None, None - raise - idents, msg_list = self.feed_identities(msg_list, copy) - return idents, self.deserialize(msg_list, content=content, copy=copy) diff --git a/tests/echo_kernel.py b/tests/echo_kernel.py new file mode 100644 index 0000000..3dc2197 --- /dev/null +++ b/tests/echo_kernel.py @@ -0,0 +1,31 @@ +"The minimal Python language adapter used by kernmini's client-driven tests." + +import sys +from contextlib import contextmanager + +from kernmini import run_kernel + + +class EchoShell: + def __init__(self, request_input=None, **kw): self.execution_count,self._stream = 0,None + def set_stream_sender(self, sender): self._stream = sender + + @contextmanager + def execution_context(self, allow_stdin, silent): yield + + def kernel_info(self): + return dict(implementation="echokernel", implementation_version="0.0.1", banner="echo", + language_info=dict(name="echo", version="1.0", mimetype="text/plain", file_extension=".txt")) + + async def execute(self, code, silent=False, store_history=True, user_expressions=None, allow_stdin=False): + self.execution_count += 1 + if self._stream: self._stream("stdout", f"echo: {code}\n") + if code.startswith("sleep:"): + import asyncio + await asyncio.sleep(float(code[6:])) + if code == "boom": return dict(execution_count=self.execution_count, error=dict(ename="EchoError", evalue=code, traceback=[])) + if code == "bytes": return dict(execution_count=self.execution_count, result={"image/png": b"raw"}) + return dict(execution_count=self.execution_count, result={"text/plain": code.upper()}) + + +if __name__ == "__main__": run_kernel(sys.argv[-1], EchoShell) diff --git a/tests/ipython_kernel.py b/tests/ipython_kernel.py new file mode 100644 index 0000000..d74bf45 --- /dev/null +++ b/tests/ipython_kernel.py @@ -0,0 +1,20 @@ +"The real ipymini language adapter hosted directly by kernmini." + +import asyncio, sys + +from ipymini.shell import MiniShell +from kernmini._native import run_kernel + + +async def main(): + user_ns, first = {}, True + def shell_factory(): + nonlocal first + shell = MiniShell(request_input=lambda *_: "", user_ns=user_ns, use_singleton=first) + first = False + return shell + await run_kernel(sys.argv[-1], shell_factory, asyncio.new_event_loop) + + +if __name__ == "__main__": + with asyncio.Runner() as runner: runner.run(main()) diff --git a/tests/test_kernel_echo.py b/tests/test_kernel_echo.py index d09556c..351a645 100644 --- a/tests/test_kernel_echo.py +++ b/tests/test_kernel_echo.py @@ -1,252 +1,135 @@ "The Python adapter running a trivial shell over the native kernmini engine." -import json, socket, subprocess, sys, time +import asyncio, os, sys +from pathlib import Path -import pytest, zmq +import pytest +from conkernelclient import JmsgQueues, run_kernel +from jupywire.ops import parent_id -from client import MiniSession -RUNNER = ''' -import sys -from contextlib import contextmanager -from kernmini import run_kernel +ROOT = Path(__file__).parents[1] +ECHO_ARGV = [sys.executable, str(ROOT/'tests'/'echo_kernel.py'), "{connection_file}"] -class EchoShell: - "The minimal shell contract: execute, execution_count, execution_context, set_stream_sender." +async def _run(kc, code, **kw): return [m async for m in kc.run(code, timeout=30, **kw)] +def _one(msgs, msg_type): return next(m for m in msgs if m['msg_type'] == msg_type) +def _pubs(msgs): return [m for m in msgs if m['channel'] == 'iopub'] - def __init__(self, request_input=None, **kw): - self.execution_count = 0 - self._stream = None - def set_stream_sender(self, sender): self._stream = sender - - @contextmanager - def execution_context(self, allow_stdin, silent): yield - - def kernel_info(self): - return dict(implementation="echokernel", implementation_version="0.0.1", banner="echo", - language_info=dict(name="echo", version="1.0", mimetype="text/plain", file_extension=".txt")) - - async def execute(self, code, silent=False, store_history=True, user_expressions=None, allow_stdin=False): - self.execution_count += 1 - if self._stream: self._stream("stdout", f"echo: {code}\\n") - if code.startswith("sleep:"): - import asyncio - await asyncio.sleep(float(code[6:])) - if code == "boom": return dict(execution_count=self.execution_count, error=dict(ename="EchoError", evalue=code, traceback=[])) - if code == "bytes": return dict(execution_count=self.execution_count, result={"image/png": b"raw"}) - return dict(execution_count=self.execution_count, result={"text/plain": code.upper()}) - - -run_kernel(sys.argv[-1], EchoShell) -''' - - -def _sock(ctx, typ, port, identity=None): - s = ctx.socket(typ) - s.linger = 0 - if identity is not None: s.setsockopt(zmq.IDENTITY, identity) - if typ == zmq.SUB: s.setsockopt(zmq.SUBSCRIBE, b"") - s.connect(f"tcp://127.0.0.1:{port}") - return s - - -def _ports(n): - socks = [socket.socket() for _ in range(n)] - for s in socks: s.bind(("127.0.0.1", 0)) - ports = [s.getsockname()[1] for s in socks] - for s in socks: s.close() - return ports - - -def _drain_iopub(sub, until_idle=True, timeout=10.0): - "Collect iopub msg dicts until an idle status (skipping the welcome)." - sess, out = _drain_iopub.sess, [] - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - if not sub.poll(200): continue - frames = sub.recv_multipart() - idents, rest = sess.feed_identities(frames) - msg = sess.deserialize(rest) - if msg["msg_type"] == "iopub_welcome": continue +async def _until_stream(run, text): + out = [] + while True: + msg = await anext(run) out.append(msg) - if until_idle and msg["msg_type"] == "status" and msg["content"]["execution_state"] == "idle": return out - raise TimeoutError(f"no idle within {timeout}s; got {[m['msg_type'] for m in out]}") - - -def _await_welcome(sub, timeout=60.0): - "Wait for the JEP 65 iopub_welcome: proof the subscription is live, so no later message can be missed." - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - if not sub.poll(200): continue - _, rest = _drain_iopub.sess.feed_identities(sub.recv_multipart()) - if _drain_iopub.sess.deserialize(rest)["msg_type"] == "iopub_welcome": return - raise TimeoutError("no iopub_welcome") - - -def _request(sock, sess, msg_type, content, timeout=10.0): - sock.send_multipart(sess.serialize(sess.msg(msg_type, content))) - if not sock.poll(timeout * 1000): raise TimeoutError(f"no reply to {msg_type}") - idents, rest = sess.feed_identities(sock.recv_multipart()) - return sess.deserialize(rest) - - -@pytest.fixture -def echo_kernel(tmp_path): - key = "test-key-123" - shell_p, iopub_p, stdin_p, control_p, hb_p = _ports(5) - conn = dict(transport="tcp", ip="127.0.0.1", shell_port=shell_p, iopub_port=iopub_p, stdin_port=stdin_p, - control_port=control_p, hb_port=hb_p, key=key, signature_scheme="hmac-sha256") - cf = tmp_path / "conn.json" - cf.write_text(json.dumps(conn)) - runner = tmp_path / "echo_runner.py" - runner.write_text(RUNNER) - proc = subprocess.Popen([sys.executable, str(runner), str(cf)], stderr=subprocess.PIPE) - ctx = zmq.Context.instance() - sess = MiniSession(key=key.encode(), username="testclient") - _drain_iopub.sess = MiniSession(key=key.encode()) - shell, control, sub = _sock(ctx, zmq.DEALER, shell_p), _sock(ctx, zmq.DEALER, control_p), _sock(ctx, zmq.SUB, iopub_p) - _await_welcome(sub) - try: yield proc, sess, shell, control, sub - finally: - for s in (shell, control, sub): s.close(0) - if proc.poll() is None: - proc.terminate() - proc.wait(timeout=5) - - -def echo_kernel_story(kernel, supported_features=None): - proc, sess, shell, control, sub = kernel - - info = _request(shell, sess, "kernel_info_request", {}, timeout=30) - assert info["msg_type"] == "kernel_info_reply" - c = info["content"] - assert c["implementation"] == "echokernel" and c["language_info"]["name"] == "echo" - assert c["supported_features"] == (supported_features or []) and c["debugger"] is False - - _drain_iopub(sub) # busy/idle for kernel_info - reply = _request(shell, sess, "execute_request", dict(code="hello world")) - msgs = _drain_iopub(sub) - assert reply["content"]["status"] == "ok" and reply["content"]["execution_count"] == 1 - kinds = [m["msg_type"] for m in msgs] - assert kinds == ["status", "execute_input", "stream", "execute_result", "status"] - assert msgs[2]["content"]["text"] == "echo: hello world\n" - assert msgs[3]["content"]["data"]["text/plain"] == "HELLO WORLD" - - reply = _request(shell, sess, "execute_request", dict(code="boom")) - msgs = _drain_iopub(sub) - assert reply["content"]["status"] == "error" and reply["content"]["ename"] == "EchoError" - assert any(m["msg_type"] == "error" for m in msgs) - - reply = _request(control, sess, "shutdown_request", dict(restart=False)) - assert reply["content"]["status"] == "ok" - assert proc.wait(timeout=10) in (0, -9) # group-leader kernels SIGKILL their own process group as the designed last act - - -def test_echo_kernel_end_to_end(echo_kernel): echo_kernel_story(echo_kernel, ["kernel subshells"]) - - -def test_python_adapter_binary_result(echo_kernel): - _,sess,shell,_,sub = echo_kernel - reply = _request(shell, sess, "execute_request", dict(code="bytes")) - msgs = _drain_iopub(sub) - result, = (m for m in msgs if m["msg_type"] == "execute_result") - assert reply["content"]["status"] == "ok" and result["content"]["data"]["image/png"] == "cmF3" - - -def _send(sock, sess, msg_type, content, metadata=None, subshell_id=None): - "Send without awaiting the reply; returns the msg_id." - m = sess.msg(msg_type, content, metadata=metadata) - if subshell_id: m["header"]["subshell_id"] = subshell_id - sock.send_multipart(sess.serialize(m)) - return m["header"]["msg_id"] - - -def _replies(sock, sess, n, timeout=10.0): - "Collect `n` shell replies in arrival order as (parent_msg_id, content) pairs." - out, deadline = [], time.monotonic() + timeout - while len(out) < n and time.monotonic() < deadline: - if not sock.poll(200): continue - _, rest = sess.feed_identities(sock.recv_multipart()) - msg = sess.deserialize(rest) - out.append((msg["parent_header"]["msg_id"], msg["content"])) - assert len(out) == n, f"expected {n} replies, got {len(out)}" - return out - - -def test_priority_and_hold(echo_kernel): - proc, sess, shell, control, sub = echo_kernel - _request(shell, sess, "kernel_info_request", {}, timeout=30) - _drain_iopub(sub) - - # priority: a queued higher-priority execute overtakes a queued normal one - mid_s = _send(shell, sess, "execute_request", dict(code="sleep:0.3")) - time.sleep(0.1) # the sleeper takes the baton; the next two queue behind it - mid_a = _send(shell, sess, "execute_request", dict(code="a")) - mid_b = _send(shell, sess, "execute_request", dict(code="b"), metadata=dict(priority=1)) - order = [mid for mid, c in _replies(shell, sess, 3)] - assert order == [mid_s, mid_b, mid_a], "priority 1 must overtake the queued normal execute" - - # hold: parks the queue; higher priority passes, normal waits, release completes - mid_h = _send(shell, sess, "execute_request", dict(code=""), metadata=dict(hold=True)) - time.sleep(0.1) - mid_x = _send(shell, sess, "execute_request", dict(code="x")) - mid_y = _send(shell, sess, "execute_request", dict(code="y"), metadata=dict(priority=1)) - (got_y, c_y), = _replies(shell, sess, 1) - assert got_y == mid_y and c_y["status"] == "ok", "priority 1 must run during the hold" - rel = _request(control, sess, "release_request", dict(msg_id=mid_h)) - assert rel["content"]["status"] == "ok" and rel["content"]["found"] is True - (got_h, c_h), (got_x, c_x) = _replies(shell, sess, 2) - assert (got_h, c_h["status"]) == (mid_h, "ok"), "release completes the hold" - assert (got_x, c_x["status"]) == (mid_x, "ok"), "the parked normal execute runs after release" - rel = _request(control, sess, "release_request", dict(msg_id=mid_h)) - assert rel["content"]["found"] is False, "a completed hold is gone; late release is a quiet no-op" - - # release with status=error: the hold errors and aborts the queued tail - mid_h2 = _send(shell, sess, "execute_request", dict(code=""), metadata=dict(hold=True)) - time.sleep(0.1) - mid_z = _send(shell, sess, "execute_request", dict(code="z")) - time.sleep(0.1) # let z reach the shell queue: control and shell are separate sockets with no cross-channel ordering - _request(control, sess, "release_request", dict(msg_id=mid_h2, status="error")) - (got_h2, c_h2), (got_z, c_z) = _replies(shell, sess, 2) - assert (got_h2, c_h2["status"], c_h2["ename"]) == (mid_h2, "error", "HoldError") - assert (got_z, c_z["status"]) == (mid_z, "aborted"), "an error hold aborts the queued tail" - - # interrupt during a hold: the hold aborts, and so does the queued tail - mid_h3 = _send(shell, sess, "execute_request", dict(code=""), metadata=dict(hold=True)) - time.sleep(0.1) - mid_w = _send(shell, sess, "execute_request", dict(code="w")) - time.sleep(0.1) # as above: w must be queued before the interrupt lands - _request(control, sess, "interrupt_request", {}) - (got_h3, c_h3), (got_w, c_w) = _replies(shell, sess, 2) - assert (got_h3, c_h3["status"], c_h3["ename"]) == (mid_h3, "error", "KeyboardInterrupt") - assert (got_w, c_w["status"]) == (mid_w, "aborted") - - -def test_hold_timeout(tmp_path): - key = "test-key-123" - shell_p, iopub_p, stdin_p, control_p, hb_p = _ports(5) - conn = dict(transport="tcp", ip="127.0.0.1", shell_port=shell_p, iopub_port=iopub_p, stdin_port=stdin_p, - control_port=control_p, hb_port=hb_p, key=key, signature_scheme="hmac-sha256") - cf = tmp_path / "conn.json" - cf.write_text(json.dumps(conn)) - runner = tmp_path / "echo_runner.py" - runner.write_text(RUNNER) - import os - env = os.environ | dict(KERNMINI_HOLD_TIMEOUT="0.2") - proc = subprocess.Popen([sys.executable, str(runner), str(cf)], stderr=subprocess.PIPE, env=env) - ctx = zmq.Context.instance() - sess = MiniSession(key=key.encode(), username="testclient") - _drain_iopub.sess = MiniSession(key=key.encode()) - shell, control, sub = _sock(ctx, zmq.DEALER, shell_p), _sock(ctx, zmq.DEALER, control_p), _sock(ctx, zmq.SUB, iopub_p) - try: - _await_welcome(sub) - mid_h = _send(shell, sess, "execute_request", dict(code=""), metadata=dict(hold=True)) - (got_h, c_h), = _replies(shell, sess, 1, timeout=5) - assert (got_h, c_h["status"], c_h["ename"]) == (mid_h, "error", "HoldTimeout") - finally: - for s in (shell, control, sub): s.close(0) - if proc.poll() is None: - proc.terminate() - proc.wait(timeout=5) + if msg['msg_type'] == 'stream' and msg['content']['text'] == text: return out + + +async def _busy(qs, msg_id): + return await qs.jmsg_for('status', pred=lambda m: parent_id(m) == msg_id and m['content']['execution_state'] == 'busy', queue='iopub', timeout=10) + + +def _watch(waiter, order): + task = asyncio.ensure_future(waiter) + def _done(task): + if not task.cancelled() and task.exception() is None: order.append(parent_id(task.result())) + task.add_done_callback(_done) + return task + + +async def echo_kernel_story(kc, supported_features=None, binary=False): + info = await kc.cmd.kernel_info(timeout=30) + content = info['content'] + assert info['msg_type'] == 'kernel_info_reply' + assert content['implementation'] == 'echokernel' and content['language_info']['name'] == 'echo' + assert content['supported_features'] == (supported_features or []) and content['debugger'] is False + + msgs = await _run(kc, 'hello world') + reply = _one(msgs, 'execute_reply') + pubs = _pubs(msgs) + assert reply['content']['status'] == 'ok' and reply['content']['execution_count'] == 1 + assert [m['msg_type'] for m in pubs] == ['status', 'execute_input', 'stream', 'execute_result', 'status'] + assert pubs[2]['content']['text'] == 'echo: hello world\n' + assert pubs[3]['content']['data']['text/plain'] == 'HELLO WORLD' + + msgs = await _run(kc, 'boom') + assert _one(msgs, 'execute_reply')['content']['status'] == 'error' + assert _one(msgs, 'error')['content']['ename'] == 'EchoError' + + if binary: + msgs = await _run(kc, 'bytes') + assert _one(msgs, 'execute_reply')['content']['status'] == 'ok' + assert _one(msgs, 'execute_result')['content']['data']['image/png'] == 'cmF3' + + +async def execution_queue_story(kc): + qs = JmsgQueues(kc) + + # Once the sleeper has emitted output it owns the execution lane; priority then overtakes normal. + sleeper_id = kc.new_msg_id() + sleeper = kc.run('sleep:0.2', msg_id=sleeper_id, timeout=5) + await _until_stream(sleeper, 'echo: sleep:0.2\n') + order = [] + normal_id, priority_id = kc.new_msg_id(), kc.new_msg_id() + normal = _watch(kc.reply('normal', msg_id=normal_id, timeout=5), order) + priority = _watch(kc.reply('priority', msg_id=priority_id, metadata=dict(priority=1), timeout=5), order) + async for _ in sleeper: pass + await asyncio.gather(normal, priority) + assert order == [priority_id, normal_id] + + # A hold parks normal work, lets priority through, and completes on release. + order = [] + held_id, normal_id, priority_id = (kc.new_msg_id() for _ in range(3)) + held = _watch(kc.reply('', msg_id=held_id, metadata=dict(hold=True), timeout=5), order) + await _busy(qs, held_id) + normal = _watch(kc.reply('normal', msg_id=normal_id, timeout=5), order) + priority = _watch(kc.reply('priority', msg_id=priority_id, metadata=dict(priority=1), timeout=5), order) + assert (await priority)['content']['status'] == 'ok' and order == [priority_id] + assert (await kc.ctl.release(msg_id=held_id, timeout=5))['content']['found'] is True + held_reply, normal_reply = await asyncio.gather(held, normal) + assert [held_reply['content']['status'], normal_reply['content']['status']] == ['ok', 'ok'] + assert order == [priority_id, held_id, normal_id] + assert (await kc.ctl.release(msg_id=held_id, timeout=5))['content']['found'] is False + + # A priority barrier proves the normal tail reached the shell queue before control releases the hold as an error. + order = [] + held_id, normal_id, barrier_id = (kc.new_msg_id() for _ in range(3)) + held = _watch(kc.reply('', msg_id=held_id, metadata=dict(hold=True), timeout=5), order) + await _busy(qs, held_id) + normal = _watch(kc.reply('normal', msg_id=normal_id, timeout=5), order) + barrier = _watch(kc.reply('barrier', msg_id=barrier_id, metadata=dict(priority=1), timeout=5), order) + await barrier + await kc.ctl.release(msg_id=held_id, status='error', timeout=5) + held_reply, normal_reply = await asyncio.gather(held, normal) + assert (held_reply['content']['status'], held_reply['content']['ename']) == ('error', 'HoldError') + assert normal_reply['content']['status'] == 'aborted' + assert order == [barrier_id, held_id, normal_id] + + # The same barrier makes interrupt ordering deterministic. + order = [] + held_id, normal_id, barrier_id = (kc.new_msg_id() for _ in range(3)) + held = _watch(kc.reply('', msg_id=held_id, metadata=dict(hold=True), timeout=5), order) + await _busy(qs, held_id) + normal = _watch(kc.reply('normal', msg_id=normal_id, timeout=5), order) + barrier = _watch(kc.reply('barrier', msg_id=barrier_id, metadata=dict(priority=1), timeout=5), order) + await barrier + await kc.interrupt(timeout=5) + held_reply, normal_reply = await asyncio.gather(held, normal) + assert (held_reply['content']['status'], held_reply['content']['ename']) == ('error', 'KeyboardInterrupt') + assert normal_reply['content']['status'] == 'aborted' + assert order == [barrier_id, held_id, normal_id] + + +@pytest.mark.asyncio +async def test_python_adapter_story(): + async with run_kernel('echo', ECHO_ARGV) as (_, kc): + await echo_kernel_story(kc, ['kernel subshells'], binary=True) + await execution_queue_story(kc) + + +@pytest.mark.asyncio +async def test_hold_timeout(): + env = os.environ | dict(KERNMINI_HOLD_TIMEOUT='0.2') + async with run_kernel('echo', ECHO_ARGV, env=env) as (_, kc): + reply = await kc.reply('', metadata=dict(hold=True), timeout=5) + assert (reply['content']['status'], reply['content']['ename']) == ('error', 'HoldTimeout') diff --git a/tests/test_rust_ipython_kernel.py b/tests/test_rust_ipython_kernel.py index d1c344c..614796e 100644 --- a/tests/test_rust_ipython_kernel.py +++ b/tests/test_rust_ipython_kernel.py @@ -1,176 +1,97 @@ -"The Rust kernel engine hosting ipymini's real MiniShell." - -import json, subprocess, sys, time - -import pytest, zmq - -from client import MiniSession -from test_kernel_echo import _await_welcome, _drain_iopub, _ports, _replies, _request, _send, _sock - - -RUNNER = ''' -import asyncio, sys -from ipymini.shell import MiniShell -from kernmini._native import run_kernel - - -async def main(): - user_ns, first = {}, True - def shell_factory(): - nonlocal first - shell = MiniShell(request_input=lambda *_: "", user_ns=user_ns, use_singleton=first) - first = False - return shell - await run_kernel(sys.argv[-1], shell_factory, asyncio.new_event_loop) - - -with asyncio.Runner() as runner: runner.run(main()) -''' - - -@pytest.fixture -def rust_ipython_kernel(tmp_path): - key = "test-key-123" - shell_p, iopub_p, stdin_p, control_p, hb_p = _ports(5) - conn = dict(transport="tcp", ip="127.0.0.1", shell_port=shell_p, iopub_port=iopub_p, stdin_port=stdin_p, - control_port=control_p, hb_port=hb_p, key=key, signature_scheme="hmac-sha256") - cf = tmp_path / "conn.json" - cf.write_text(json.dumps(conn)) - runner = tmp_path / "runner.py" - runner.write_text(RUNNER) - proc = subprocess.Popen([sys.executable, str(runner), str(cf)]) - ctx = zmq.Context.instance() - sess = MiniSession(key=key.encode(), username="testclient") - _drain_iopub.sess = MiniSession(key=key.encode()) - identity = b"testclient" - shell = _sock(ctx, zmq.DEALER, shell_p, identity) - control = _sock(ctx, zmq.DEALER, control_p, identity) - stdin = _sock(ctx, zmq.DEALER, stdin_p, identity) - sub = _sock(ctx, zmq.SUB, iopub_p) - try: - _await_welcome(sub) - yield proc, sess, shell, control, stdin, sub - finally: - for socket in (shell, control, stdin, sub): socket.close(0) - if proc.poll() is None: - proc.terminate() - proc.wait(timeout=5) - - -def _wait_stream(sub, text, timeout=10): - deadline, seen = time.monotonic() + timeout, [] - while time.monotonic() < deadline: - if not sub.poll(200): continue - _, rest = _drain_iopub.sess.feed_identities(sub.recv_multipart()) - msg = _drain_iopub.sess.deserialize(rest) - seen.append((msg["msg_type"], msg["content"])) - if msg["msg_type"] == "stream" and text in msg["content"]["text"]: return msg - raise TimeoutError(f"no stream containing {text!r}; got {seen}") - - -def test_ipython_story(rust_ipython_kernel): - proc, sess, shell, control, stdin, sub = rust_ipython_kernel - - info = _request(shell, sess, "kernel_info_request", {}, timeout=30) - assert info["content"]["implementation"] == "ipymini" - assert "kernel subshells" in info["content"]["supported_features"] - _drain_iopub(sub) - - reply = _request(shell, sess, "execute_request", dict(code="x = 41\nprint('ready')\nx + 1")) - msgs = _drain_iopub(sub) - assert reply["content"]["status"] == "ok" - execute_input, = (m for m in msgs if m["msg_type"] == "execute_input") - assert execute_input["content"]["execution_count"] == 1 - assert [(m["content"]["name"], m["content"]["text"]) for m in msgs if m["msg_type"] == "stream"] == [("stdout", "ready\n")] - result, = (m for m in msgs if m["msg_type"] == "execute_result") - assert result["content"]["data"]["text/plain"] == "42" - - child = "sidecar" - child_id = _send(shell, sess, "execute_request", dict(code="x + 1"), subshell_id=child) - (reply_id, child_reply), = _replies(shell, sess, 1) - assert reply_id == child_id and child_reply["status"] == "ok" and child_reply["execution_count"] == 1 - msgs = _drain_iopub(sub) - result, = (m for m in msgs if m["msg_type"] == "execute_result") - assert result["content"]["data"]["text/plain"] == "42" and result["parent_header"]["subshell_id"] == child - assert _request(control, sess, "list_subshell_request", {})["content"]["subshell_id"] == [child] - created = _request(control, sess, "create_subshell_request", dict(subshell_id=child), timeout=30) - again = _request(control, sess, "create_subshell_request", dict(subshell_id=child), timeout=30) - assert created["content"] == again["content"] == dict(status="ok", subshell_id=child) - assert _request(control, sess, "delete_subshell_request", dict(subshell_id=child))["content"]["status"] == "ok" - assert _request(control, sess, "list_subshell_request", {})["content"]["subshell_id"] == [] - - caller = _send(shell, sess, "execute_request", dict(code="import asyncio\nfrom ipymini import sidecar\nloop = asyncio.get_running_loop()\ngate2 = asyncio.Event()\nwith sidecar():\n print('sidecar ready', flush=True)\n await asyncio.wait_for(gate2.wait(), 5)")) - _wait_stream(sub, "sidecar ready") - routed = _send(shell, sess, "execute_request", dict(code="loop.call_soon_threadsafe(gate2.set)")) - replies = dict(_replies(shell, sess, 2)) - assert replies[caller]["status"] == replies[routed]["status"] == "ok" - assert replies[routed]["execution_count"] == 1 - _drain_iopub(sub) - _drain_iopub(sub) - assert _request(control, sess, "list_subshell_request", {})["content"]["subshell_id"] == ["sidecar"] - - input_id = _send(shell, sess, "execute_request", dict(code="print(input('Name: '))", allow_stdin=True)) - assert stdin.poll(10_000), "no input_request" - _, stdin_request = sess.recv(stdin) - assert stdin_request["content"] == {"prompt": "Name: ", "password": False} - sess.send(stdin, "input_reply", {"value": "Ada"}, parent=stdin_request) - (reply_id, input_reply), = _replies(shell, sess, 1) - assert reply_id == input_id and input_reply["status"] == "ok" - msgs = _drain_iopub(sub) - assert any(m["msg_type"] == "stream" and m["content"]["text"] == "Ada\n" for m in msgs) - - complete = _request(shell, sess, "complete_request", dict(code="x.rea", cursor_pos=5)) - _drain_iopub(sub) - assert any(match.endswith("real") for match in complete["content"]["matches"]) - - inspect = _request(shell, sess, "inspect_request", dict(code="x", cursor_pos=1, detail_level=0)) - _drain_iopub(sub) - assert inspect["content"]["found"] and "int" in inspect["content"]["data"]["text/plain"] - - complete_code = _request(shell, sess, "is_complete_request", dict(code="for i in range(2):")) - _drain_iopub(sub) - assert complete_code["content"] == {"status": "incomplete", "indent": " "} - - history = _request(shell, sess, "history_request", dict(hist_access_type="tail", output=False, raw=True, n=1)) - _drain_iopub(sub) - assert history["content"]["history"][-1][-1] == "print(input('Name: '))" - - task_id = _send(shell, sess, "execute_request", dict(code="import asyncio\nasync def later():\n await asyncio.sleep(.01)\n print('background')\n return x + 1\ntask = asyncio.create_task(later())")) - (task_reply_id, task_reply), = _replies(shell, sess, 1) - assert task_reply_id == task_id and task_reply["status"] == "ok" - _drain_iopub(sub) - reply = _request(shell, sess, "execute_request", dict(code="await task")) - msgs = _drain_iopub(sub) - assert reply["content"]["status"] == "ok" - result, = (m for m in msgs if m["msg_type"] == "execute_result") - assert result["content"]["data"]["text/plain"] == "42" - background, = (m for m in msgs if m["msg_type"] == "stream" and m["content"]["text"] == "background\n") - assert background["parent_header"]["msg_id"] == task_id - - sleeper = _send(shell, sess, "execute_request", dict(code="print('sleeping', flush=True)\nawait asyncio.sleep(.3)")) - _wait_stream(sub, "sleeping") - completer = _send(shell, sess, "complete_request", dict(code="x.rea", cursor_pos=5)) - (first_id, first), (second_id, second) = _replies(shell, sess, 2) - assert first_id == completer and first["status"] == "ok" - assert second_id == sleeper and second["status"] == "ok" - _drain_iopub(sub) - _drain_iopub(sub) - - for code in ("await asyncio.sleep(60)", "while True: pass"): - interrupted = _send(shell, sess, "execute_request", dict(code=code)) - time.sleep(.1) - assert _request(control, sess, "interrupt_request", {})["content"]["status"] == "ok" - (reply_id, reply), = _replies(shell, sess, 1) - assert reply_id == interrupted and (reply["status"], reply["ename"]) == ("error", "KeyboardInterrupt") - _drain_iopub(sub) - - missing = _request(shell, sess, "execute_request", {}) - msgs = _drain_iopub(sub) - assert missing["content"]["status"] == "error" and missing["content"]["ename"] == "MissingField" - assert [m["content"]["execution_state"] for m in msgs if m["msg_type"] == "status"] == ["busy", "idle"] - missing = _request(shell, sess, "complete_request", {}) - assert missing["content"]["status"] == "error" and missing["content"]["matches"] == [] - - reply = _request(control, sess, "shutdown_request", dict(restart=False)) - assert reply["content"]["status"] == "ok" - assert proc.wait(timeout=10) == 0 +"The real ipymini language adapter hosted directly by kernmini and driven by ConKernelClient." + +import pytest +from conkernelclient import JmsgQueues, run_kernel +from jupywire.ops import parent_id + +from test_kernel_echo import ROOT, _one, _pubs, _run, _until_stream + + +IPYTHON_ARGV = [__import__('sys').executable, str(ROOT/'tests'/'ipython_kernel.py'), '{connection_file}'] + + +@pytest.mark.asyncio +async def test_ipython_story(): + async with run_kernel('rust-ipython', IPYTHON_ARGV) as (_, kc): + qs = JmsgQueues(kc) + + info = await kc.cmd.kernel_info(timeout=30) + assert info['content']['implementation'] == 'ipymini' + assert 'kernel subshells' in info['content']['supported_features'] + + msgs = await _run(kc, "x = 41\nprint('ready')\nx + 1") + reply, pubs = _one(msgs, 'execute_reply'), _pubs(msgs) + assert reply['content']['status'] == 'ok' + assert _one(msgs, 'execute_input')['content']['execution_count'] == 1 + assert [(m['content']['name'], m['content']['text']) for m in pubs if m['msg_type'] == 'stream'] == [('stdout', 'ready\n')] + assert _one(msgs, 'execute_result')['content']['data']['text/plain'] == '42' + + child = 'sidecar' + msgs = await _run(kc, 'x + 1', subshell_id=child) + reply, result = _one(msgs, 'execute_reply'), _one(msgs, 'execute_result') + assert reply['content']['status'] == 'ok' and reply['content']['execution_count'] == 1 + assert result['content']['data']['text/plain'] == '42' and result['parent_header']['subshell_id'] == child + assert (await kc.ctl.list_subshell(timeout=5))['content']['subshell_id'] == [child] + created = await kc.ctl.create_subshell(subshell_id=child, timeout=5) + again = await kc.ctl.create_subshell(subshell_id=child, timeout=5) + assert created['content'] == again['content'] == dict(status='ok', subshell_id=child) + assert (await kc.ctl.delete_subshell(subshell_id=child, timeout=5))['content']['status'] == 'ok' + assert (await kc.ctl.list_subshell(timeout=5))['content']['subshell_id'] == [] + + caller = kc.run("import asyncio\nfrom ipymini import sidecar\nloop = asyncio.get_running_loop()\ngate2 = asyncio.Event()\nwith sidecar():\n" + " print('sidecar ready', flush=True)\n await asyncio.wait_for(gate2.wait(), 5)", timeout=10) + caller_msgs = await _until_stream(caller, 'sidecar ready\n') + routed = await kc.reply('loop.call_soon_threadsafe(gate2.set)', timeout=5) + caller_msgs += [m async for m in caller] + assert routed['content']['status'] == _one(caller_msgs, 'execute_reply')['content']['status'] == 'ok' + assert routed['content']['execution_count'] == 1 + assert (await kc.ctl.list_subshell(timeout=5))['content']['subshell_id'] == ['sidecar'] + + msgs = await _run(kc, "print(input('Name: '))", on_stdin=lambda _: 'Ada') + assert _one(msgs, 'execute_reply')['content']['status'] == 'ok' + assert any(m['msg_type'] == 'stream' and m['content']['text'] == 'Ada\n' for m in msgs) + + complete = await kc.cmd.complete(code='x.rea', cursor_pos=5, timeout=5) + assert any(match.endswith('real') for match in complete['content']['matches']) + inspect = await kc.cmd.inspect(code='x', cursor_pos=1, detail_level=0, timeout=5) + assert inspect['content']['found'] and 'int' in inspect['content']['data']['text/plain'] + complete_code = await kc.shell_request('is_complete_request', code='for i in range(2):', timeout=5) + assert complete_code['content'] == {'status': 'incomplete', 'indent': ' '} + history = await kc.cmd.history(hist_access_type='tail', output=False, raw=True, n=1, timeout=5) + assert history['content']['history'][-1][-1] == "print(input('Name: '))" + + task_id = kc.new_msg_id() + created = await _run(kc, "import asyncio\nbackground_gate = asyncio.Event()\nasync def later():\n await background_gate.wait()\n" + " print('background')\n return x + 1\ntask = asyncio.create_task(later())", msg_id=task_id) + assert not any(m['msg_type'] == 'stream' and m['content']['text'] == 'background\n' for m in created) + awaited = await _run(kc, 'background_gate.set()\nawait task') + assert _one(awaited, 'execute_reply')['content']['status'] == 'ok' + assert _one(awaited, 'execute_result')['content']['data']['text/plain'] == '42' + background = await qs.jmsg_for('stream', pred=lambda m: parent_id(m) == task_id, queue='iopub', timeout=5) + assert background['content']['text'] == 'background\n' + + sleeper = kc.run("print('sleeping', flush=True)\nawait asyncio.sleep(.3)", timeout=5) + sleeper_msgs = await _until_stream(sleeper, 'sleeping\n') + complete = await kc.cmd.complete(code='x.rea', cursor_pos=5, timeout=5) + sleeper_msgs += [m async for m in sleeper] + assert complete['content']['status'] == 'ok' + assert _one(sleeper_msgs, 'execute_reply')['content']['status'] == 'ok' + + for marker, code in [('waiting', 'await asyncio.sleep(60)'), ('spinning', 'while True: pass')]: + running = kc.run(f"print('{marker}', flush=True)\n{code}", timeout=10) + running_msgs = await _until_stream(running, f'{marker}\n') + assert (await kc.interrupt(timeout=5))['content']['status'] == 'ok' + running_msgs += [m async for m in running] + interrupted = _one(running_msgs, 'execute_reply')['content'] + assert (interrupted['status'], interrupted['ename']) == ('error', 'KeyboardInterrupt') + + missing_id = kc.new_msg_id() + missing = await kc.shell_request('execute_request', msg_id=missing_id, timeout=5) + states = [] + while states[-1:] != ['idle']: + status = await qs.jmsg_for('status', pred=lambda m: parent_id(m) == missing_id, queue='iopub', timeout=5) + states.append(status['content']['execution_state']) + assert missing['content']['status'] == 'error' and missing['content']['ename'] == 'MissingField' + assert states == ['busy', 'idle'] + missing = await kc.shell_request('complete_request', timeout=5) + assert missing['content']['status'] == 'error' and missing['content']['matches'] == [] diff --git a/tests/test_rust_kernel_echo.py b/tests/test_rust_kernel_echo.py index 56a1d54..95cd717 100644 --- a/tests/test_rust_kernel_echo.py +++ b/tests/test_rust_kernel_echo.py @@ -1,82 +1,30 @@ -"The native Rust echo language tells the complete kernel protocol story." +"The native Rust echo language tells its kernel protocol story through ConKernelClient." -import json, subprocess +import asyncio -import pytest, zmq +import pytest +from conkernelclient import run_kernel -from client import MiniSession -from test_kernel_echo import _await_welcome, _drain_iopub, _ports, _replies, _request, _send, _sock, echo_kernel_story +from test_kernel_echo import ROOT, _until_stream, _watch, echo_kernel_story -@pytest.fixture -def rust_echo_kernel(tmp_path): - key = "test-key-123" - shell_p, iopub_p, stdin_p, control_p, hb_p = _ports(5) - conn = dict(transport="tcp", ip="127.0.0.1", shell_port=shell_p, iopub_port=iopub_p, stdin_port=stdin_p, - control_port=control_p, hb_port=hb_p, key=key, signature_scheme="hmac-sha256") - cf = tmp_path / "conn.json" - cf.write_text(json.dumps(conn)) - root = __import__("pathlib").Path(__file__).parents[1] - proc = subprocess.Popen(["cargo", "run", "--quiet", "--manifest-path", str(root / "Cargo.toml"), - "-p", "kernmini", "--bin", "kernmini-echo", "--", str(cf)]) - ctx = zmq.Context.instance() - sess = MiniSession(key=key.encode(), username="testclient") - _drain_iopub.sess = MiniSession(key=key.encode()) - shell, control, sub = _sock(ctx, zmq.DEALER, shell_p), _sock(ctx, zmq.DEALER, control_p), _sock(ctx, zmq.SUB, iopub_p) - _await_welcome(sub) - try: yield proc, sess, shell, control, sub - finally: - for socket in (shell, control, sub): socket.close(0) - if proc.poll() is None: - proc.terminate() - proc.wait(timeout=5) +RUST_ECHO_ARGV = ['cargo', 'run', '--quiet', '--manifest-path', str(ROOT/'Cargo.toml'), + '-p', 'kernmini', '--bin', 'kernmini-echo', '--', '{connection_file}'] -def test_rust_echo_kernel_end_to_end(rust_echo_kernel): echo_kernel_story(rust_echo_kernel) +@pytest.mark.asyncio +async def test_rust_language_story(): + async with run_kernel('rust-echo', RUST_ECHO_ARGV) as (_, kc): + await echo_kernel_story(kc) - -def test_rust_execution_queue(rust_echo_kernel): - _, sess, shell, control, sub = rust_echo_kernel - _request(shell, sess, "kernel_info_request", {}, timeout=30) - _drain_iopub(sub) - - sleeper = _send(shell, sess, "execute_request", dict(code="sleep:0.2")) - __import__("time").sleep(.05) - normal = _send(shell, sess, "execute_request", dict(code="normal")) - priority = _send(shell, sess, "execute_request", dict(code="priority"), metadata=dict(priority=1)) - assert [mid for mid, _ in _replies(shell, sess, 3)] == [sleeper, priority, normal] - - held = _send(shell, sess, "execute_request", dict(code=""), metadata=dict(hold=True)) - __import__("time").sleep(.05) - normal = _send(shell, sess, "execute_request", dict(code="normal")) - priority = _send(shell, sess, "execute_request", dict(code="priority"), metadata=dict(priority=1)) - (priority_reply, _), = _replies(shell, sess, 1) - assert priority_reply == priority - release = _request(control, sess, "release_request", dict(msg_id=held)) - assert release["content"]["found"] is True - assert [mid for mid, _ in _replies(shell, sess, 2)] == [held, normal] - assert _request(control, sess, "release_request", dict(msg_id=held))["content"]["found"] is False - - held = _send(shell, sess, "execute_request", dict(code=""), metadata=dict(hold=True)) - __import__("time").sleep(.05) - normal = _send(shell, sess, "execute_request", dict(code="normal")) - _request(control, sess, "release_request", dict(msg_id=held, status="error")) - replies = _replies(shell, sess, 2) - assert [mid for mid, _ in replies] == [held, normal] - assert [content["status"] for _, content in replies] == ["error", "aborted"] - - held = _send(shell, sess, "execute_request", dict(code=""), metadata=dict(hold=True)) - __import__("time").sleep(.05) - normal = _send(shell, sess, "execute_request", dict(code="normal")) - _request(control, sess, "interrupt_request", {}) - replies = _replies(shell, sess, 2) - assert [mid for mid, _ in replies] == [held, normal] - assert [(content["status"], content.get("ename")) for _, content in replies] == [("error", "KeyboardInterrupt"), ("aborted", None)] - - sleeper = _send(shell, sess, "execute_request", dict(code="sleep:0.2")) - __import__("time").sleep(.05) - failed = _send(shell, sess, "execute_request", dict(code="boom")) - aborted = _send(shell, sess, "execute_request", dict(code="never")) - replies = _replies(shell, sess, 3) - assert [mid for mid, _ in replies] == [sleeper, failed, aborted] - assert [content["status"] for _, content in replies] == ["ok", "error", "aborted"] + # Synchronize on live output, then prove a failed execute aborts its queued tail. + sleeper = kc.run('sleep:0.2', timeout=5) + await _until_stream(sleeper, 'echo: sleep:0.2\n') + order = [] + failed_id, aborted_id = kc.new_msg_id(), kc.new_msg_id() + failed = _watch(kc.reply('boom', msg_id=failed_id, timeout=5), order) + aborted = _watch(kc.reply('never', msg_id=aborted_id, timeout=5), order) + async for _ in sleeper: pass + failed_reply, aborted_reply = await asyncio.gather(failed, aborted) + assert [failed_reply['content']['status'], aborted_reply['content']['status']] == ['error', 'aborted'] + assert order == [failed_id, aborted_id]