Fixed failing unit tests

jupyter_rtc_core/tests/test_output_processor.py

@@ -1,13 +1,25 @@
 import json
+import pytest
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from uuid import uuid4
 
 from ..outputs import OutputProcessor, OutputsManager
 
-class TestOutputProcessor(OutputProcessor):
 
-    settings = {}
+class OutputProcessorForTest(OutputProcessor):
+    """Test subclass of OutputProcessor that overrides the settings property."""
+    _test_settings = {}
+
+    @property
+    def settings(self):
+        """Override the settings property to return a test dictionary."""
+        return self._test_settings
+
+@pytest.fixture
+def output_processor():
+    """Fixture that returns an instance of TestOutputProcessor."""
+    return OutputProcessorForTest()
 
 def create_incoming_message(cell_id):
     msg_id = str(uuid4())
@@ -22,10 +34,10 @@ def test_instantiation():
     op = OutputProcessor()
     assert isinstance(op, OutputProcessor)
 
-def test_incoming_message():
+def test_incoming_message(output_processor):
     """Test incoming message processing."""
     with TemporaryDirectory() as td:
-        op = TestOutputProcessor()
+        op = output_processor
         om = OutputsManager()
         op.settings["outputs_manager"] = om
         op.outputs_path = Path(td) / "outputs"

jupyter_rtc_core/tests/test_outputs_manager.py

@@ -47,7 +47,7 @@ def test_stream():
         file_id = str(uuid4())
         cell_id = str(uuid4())
         for s in streams:
-            op.write_stream(file_id, cell_id, s)
+            op.write(file_id, cell_id, s)
         assert op.get_stream(file_id, cell_id) == text
 
 def test_display_data():
@@ -90,4 +90,4 @@ def file_not_found():
         with pytest.raises(FileNotFoundError):
             op.get_output('a','b',0)
         with pytest.raises(FileNotFoundError):
-            op.get_stream('a','b')       
+            op.get_stream('a','b')

jupyter_rtc_core/websockets/clients.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 from datetime import timedelta, timezone, datetime
 from logging import Logger
-from typing import TYPE_CHECKING
+from typing import Optional, TYPE_CHECKING
 import uuid
 import asyncio
 
@@ -91,7 +91,7 @@ def __init__(self, *, room_id: str, log: Logger, loop: asyncio.AbstractEventLoop
         self.desynced_timeout_seconds = desynced_timeout_seconds
         self._stopped = False
 
-    def add(self, websocket: WebSocketHandler) -> str | None:
+    def add(self, websocket: WebSocketHandler) -> Optional[str]:
         """
         Adds a new Websocket as a client to the group.
 

jupyter_rtc_core/websockets/yroom_ws.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 from tornado.httpclient import HTTPError
 from tornado.websocket import WebSocketHandler
-from typing import TYPE_CHECKING
+from typing import Optional, TYPE_CHECKING
 import asyncio
 from ..rooms import YRoomManager
 import logging
@@ -14,7 +14,7 @@
 class YRoomWebsocket(WebSocketHandler):
     yroom: YRoom
     room_id: str
-    client_id: str | None
+    client_id: Optional[str]
     # TODO: change this. we should pass `self.log` from our
     # `ExtensionApp` to log messages w/ "RtcCoreExtension" prefix
     log = logging.Logger("TEMP")