test: cover Jupyter MIME output paths
This commit is contained in:
@@ -29,8 +29,8 @@ def serialize_event(event: KernelMessage | InputRequest) -> dict[str, object]:
|
||||
"password": event.password,
|
||||
}
|
||||
|
||||
# Jupyter keeps binary MIME payloads in message buffers. Base64 makes
|
||||
# those buffers safe to carry in the same NDJSON stream as text events.
|
||||
# Some Jupyter messages keep binary payloads in message buffers. Base64
|
||||
# makes those buffers safe to carry beside JSON MIME data in the stream.
|
||||
return {
|
||||
"type": "kernel_message",
|
||||
"msg_type": event.msg_type,
|
||||
|
||||
@@ -3,16 +3,30 @@ import unittest
|
||||
|
||||
import httpx
|
||||
|
||||
from ipython_shell.jupyter.transport import KernelMessage
|
||||
from ipython_webapp.jupyter.app import (
|
||||
InputReply,
|
||||
app,
|
||||
reply_to_input,
|
||||
run_code,
|
||||
serialize_event,
|
||||
shell_app,
|
||||
)
|
||||
|
||||
|
||||
class JupyterWebAppTests(unittest.IsolatedAsyncioTestCase):
|
||||
def test_serialize_event_encodes_binary_buffers(self):
|
||||
event = KernelMessage(
|
||||
msg_type="display_data",
|
||||
parent_id="call-1",
|
||||
content={"data": {"application/octet-stream": "present"}},
|
||||
buffers=[b"\x00\xff"],
|
||||
)
|
||||
|
||||
serialized = serialize_event(event)
|
||||
|
||||
self.assertEqual(serialized["buffers"], ["AP8="])
|
||||
|
||||
async def test_run_endpoint_streams_json_events_with_mime_data(self):
|
||||
transport = httpx.ASGITransport(app=app)
|
||||
async with httpx.AsyncClient(
|
||||
|
||||
@@ -67,6 +67,27 @@ class AsyncTransportTests(unittest.IsolatedAsyncioTestCase):
|
||||
)
|
||||
self.assertEqual(error.content["ename"], "ValueError")
|
||||
|
||||
async def test_transport_preserves_matplotlib_mime_output(self):
|
||||
transport = JupyterTransport()
|
||||
await transport.start()
|
||||
try:
|
||||
call_id = await transport.execute(
|
||||
"import matplotlib.pyplot as plt\n"
|
||||
"plt.plot([1, 2, 3], [4, 5, 6])"
|
||||
)
|
||||
messages = [
|
||||
message async for message in transport.messages_for(call_id)
|
||||
]
|
||||
finally:
|
||||
await transport.shutdown()
|
||||
|
||||
display = next(
|
||||
message
|
||||
for message in messages
|
||||
if message.msg_type == "display_data"
|
||||
)
|
||||
self.assertIn("image/png", display.content["data"])
|
||||
|
||||
async def test_transport_routes_input_reply_without_parent_stdin(self):
|
||||
transport = JupyterTransport()
|
||||
await transport.start()
|
||||
|
||||
Reference in New Issue
Block a user