80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
import json
|
|
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(
|
|
transport=transport,
|
|
base_url="http://testserver",
|
|
) as client:
|
|
try:
|
|
response = await client.post(
|
|
"/shells/new/run",
|
|
content="{'answer': 42}",
|
|
headers={"content-type": "text/plain"},
|
|
)
|
|
finally:
|
|
await shell_app.shutdown()
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(
|
|
response.headers["content-type"].startswith("application/x-ndjson")
|
|
)
|
|
events = [json.loads(line) for line in response.text.splitlines()]
|
|
result = next(
|
|
event for event in events if event["msg_type"] == "execute_result"
|
|
)
|
|
self.assertEqual(result["content"]["data"]["text/plain"], "{'answer': 42}")
|
|
|
|
async def test_input_request_can_be_replied_to_while_run_streams(self):
|
|
try:
|
|
response = await run_code("new", "answer = input('name? '); answer")
|
|
self.assertEqual(response.media_type, "application/x-ndjson")
|
|
events = []
|
|
async for line in response.body_iterator:
|
|
event = json.loads(line)
|
|
events.append(event)
|
|
if event["type"] == "input_request":
|
|
reply = await reply_to_input(
|
|
event["call_id"],
|
|
InputReply(value="Ada"),
|
|
)
|
|
self.assertEqual(reply.status_code, 204)
|
|
finally:
|
|
await shell_app.shutdown()
|
|
|
|
result = next(
|
|
event for event in events if event.get("msg_type") == "execute_result"
|
|
)
|
|
self.assertEqual(result["content"]["data"]["text/plain"], "'Ada'")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|