feat: define Jupyter transport message boundary
This commit is contained in:
@@ -0,0 +1,20 @@
|
|||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class KernelMessage:
|
||||||
|
"""A decoded Jupyter message kept at the transport boundary."""
|
||||||
|
|
||||||
|
msg_type: str
|
||||||
|
parent_id: str | None
|
||||||
|
content: dict[str, object]
|
||||||
|
buffers: list[bytes]
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class InputRequest:
|
||||||
|
"""A request for user input emitted by a running kernel call."""
|
||||||
|
|
||||||
|
call_id: str
|
||||||
|
prompt: str
|
||||||
|
password: bool
|
||||||
@@ -0,0 +1,21 @@
|
|||||||
|
import unittest
|
||||||
|
|
||||||
|
from ipython_shell.transport import KernelMessage
|
||||||
|
|
||||||
|
|
||||||
|
class TransportTests(unittest.TestCase):
|
||||||
|
def test_kernel_message_keeps_content_and_buffers(self):
|
||||||
|
message = KernelMessage(
|
||||||
|
msg_type="display_data",
|
||||||
|
parent_id="call-1",
|
||||||
|
content={"data": {"text/plain": "42"}},
|
||||||
|
buffers=[b"binary"],
|
||||||
|
)
|
||||||
|
|
||||||
|
self.assertEqual(message.parent_id, "call-1")
|
||||||
|
self.assertEqual(message.content["data"], {"text/plain": "42"})
|
||||||
|
self.assertEqual(message.buffers, [b"binary"])
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user