Work on outputs processor tests.

Author: ellisonbg

jupyter_rtc_core/outputs/output_processor.py

@@ -11,7 +11,7 @@ class OutputProcessor(LoggingConfigurable):
 
     _cell_ids = Dict(default_value={}) # a map from msg_id -> cell_id
     _cell_indices = Dict(default_value={}) # a map from cell_id -> cell index in notebook
-    _file_id = Unicode(default_value=None, allow_None=True)
+    _file_id = Unicode(default_value=None, allow_none=True)
 
     @property
     def settings(self):
@@ -83,19 +83,22 @@ def process_incoming_message(self, channel: str, msg: list[bytes]):
         When output messages are send back to the frontend, this map is used
         to find the cell_id for a given parent msg_id.
         """
-        header = json.loads(msg[0])
-        msg_id = header["msg_id"]
-        msg_type = header["msg_type"]
-        metadata = json.loads(msg[2])
+        if channel != "shell":
+            return
+        header = json.loads(msg[0]) # TODO use session unpack
+        msg_type = header.get("msg_type")
+        if msg_type != "execute_request":
+            return
+        msg_id = header.get("msg_id")
+        metadata = json.loads(msg[2]) # TODO use session unpack
         cell_id = metadata.get("cellId")
         if cell_id is None:
             return
 
         existing_msg_id = self.get_msg_id(cell_id)
         if existing_msg_id != msg_id:  # cell is being re-run, clear output state
+            self.clear(cell_id)
             if self._file_id is not None:
-                self.log.info(f"Cell has been rerun, removing old outputs: {self._file_id} {cell_id}")
-                self.clear(cell_id)
                 self.outputs_manager.clear(file_id=self._file_id, cell_id=cell_id)
         self.log.info(f"Saving (msg_id, cell_id): ({msg_id} {cell_id})")
         self.set_cell_id(msg_id, cell_id)

jupyter_rtc_core/tests/test_output_processor.py

@@ -1,4 +1,51 @@
-from ..outputs import OutputProcessor
+import json
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from uuid import uuid4
+
+from ..outputs import OutputProcessor, OutputsManager
+
+class TestOutputProcessor(OutputProcessor):
+
+    settings = {}
+
 
 def test_instantiation():
-    op = OutputProcessor()
+    op = OutputProcessor()
+
+def create_incoming_message(cell_id):
+    msg_id = str(uuid4())
+    header = {"msg_id": msg_id, "msg_type": "shell"}
+    parent_header = {}
+    metadata = {"cellId": cell_id}
+    msg = [json.dumps(item) for item in [header, parent_header, metadata]]
+    return msg_id, msg
+
+def test_incoming_message():
+
+    with TemporaryDirectory() as td:
+        op = TestOutputProcessor()
+        om = OutputsManager()
+        op.settings["outputs_manager"] = om
+        op.outputs_path = Path(td) / "outputs"
+        # Simulate the running of a cell
+        cell_id = str(uuid4())
+        msg_id, msg = create_incoming_message(cell_id)
+        op.process_incoming_message('shell', msg)
+        assert op.get_cell_id(msg_id) == cell_id
+        assert op.get_msg_id(cell_id) == msg_id
+        # # Clear the cell_id
+        # op.clear(cell_id)
+        # assert op.get_cell_id(msg_id) is None
+        # assert op.get_msg_id(cell_id) is None
+        # # Simulate the running of a cell
+        # cell_id = str(uuid4())
+        # msg_id, msg = create_incoming_message(cell_id)
+        # op.process_incoming_message('shell', msg)
+        # assert op.get_cell_id(msg_id) == cell_id
+        # assert op.get_msg_id(cell_id) == msg_id
+        # # Run it again without clearing to ensure it self clears
+        # cell_id = str(uuid4())
+        # msg_id, msg = create_incoming_message(cell_id)
+        # assert op.get_cell_id(msg_id) == cell_id
+        # assert op.get_msg_id(cell_id) == msg_id