use FakePanel to fix failing tests

Author: Mike Prosser

tests/unit/test_fake_server.py

@@ -1,25 +1,29 @@
 from concurrent import futures
+from typing import Generator
+
 import grpc
 import pytest
 from google.protobuf.any_pb2 import Any
-from ni.pythonpanel.v1.python_panel_service_pb2_grpc import PythonPanelServiceStub
-from tests.utils._fake_python_panel_service import FakePythonPanelService
 from ni.pythonpanel.v1.python_panel_service_pb2 import (
     ConnectRequest,
     DisconnectRequest,
     GetValueRequest,
     SetValueRequest,
 )
+from ni.pythonpanel.v1.python_panel_service_pb2_grpc import PythonPanelServiceStub
+
+from tests.utils._fake_python_panel_service import FakePythonPanelService
 
 
 @pytest.fixture
-def grpc_server():
+def grpc_server() -> Generator[tuple[grpc.Server, int], Any, None]:
     # Create an in-process gRPC server
     server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
     servicer = FakePythonPanelService()
     from ni.pythonpanel.v1.python_panel_service_pb2_grpc import (
         add_PythonPanelServiceServicer_to_server,
     )
+
     add_PythonPanelServiceServicer_to_server(servicer, server)
     port = server.add_insecure_port("[::]:0")  # Bind to an available port
     server.start()
@@ -28,35 +32,37 @@ def grpc_server():
 
 
 @pytest.fixture
-def grpc_client(grpc_server):
+def grpc_client(
+    grpc_server: tuple[grpc.Server, int],
+) -> Generator[PythonPanelServiceStub, Any, None]:
     _, port = grpc_server
     channel = grpc.insecure_channel(f"localhost:{port}")
     yield PythonPanelServiceStub(channel)
     channel.close()
 
 
-def test_connect(grpc_client):
+def test___connect___gets_response(grpc_client: PythonPanelServiceStub) -> None:
     request = ConnectRequest(panel_id="test_panel", panel_uri="path/to/panel")
     response = grpc_client.Connect(request)
     assert response is not None  # Ensure response is returned
 
 
-def test_disconnect(grpc_client):
+def test___disconnect___gets_response(grpc_client: PythonPanelServiceStub) -> None:
     request = DisconnectRequest(panel_id="test_panel")
     response = grpc_client.Disconnect(request)
     assert response is not None  # Ensure response is returned
 
 
-def test_get_value(grpc_client):
+def test___get_value___gets_response(grpc_client: PythonPanelServiceStub) -> None:
     request = GetValueRequest(panel_id="test_panel", value_id="test_value")
     response = grpc_client.GetValue(request)
     assert response is not None  # Ensure response is returned
     assert isinstance(response.value, Any)  # Ensure the value is of type `Any`
 
 
-def test_set_value(grpc_client):
+def test___set_value___gets_response(grpc_client: PythonPanelServiceStub) -> None:
     value = Any()
     value.value = b"test_data"
     request = SetValueRequest(panel_id="test_panel", value_id="test_value", value=value)
     response = grpc_client.SetValue(request)
-    assert response is not None  # Ensure response is returned
+    assert response is not None  # Ensure response is returned

tests/unit/test_panel.py

@@ -1,14 +1,41 @@
-import nipanel
+from concurrent import futures
+from typing import Generator
 
+import grpc
+import pytest
+from google.protobuf.any_pb2 import Any
 
-def test___streamlit_panel___has_panel_id_and_panel_uri() -> None:
-    panel = nipanel.StreamlitPanel("my_panel", "path/to/script")
+from tests.utils._fake_panel import FakePanel
+from tests.utils._fake_python_panel_service import FakePythonPanelService
+
+
+@pytest.fixture
+def grpc_server() -> Generator[tuple[grpc.Server, int], Any, None]:
+    # Create an in-process gRPC server
+    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
+    servicer = FakePythonPanelService()
+    from ni.pythonpanel.v1.python_panel_service_pb2_grpc import (
+        add_PythonPanelServiceServicer_to_server,
+    )
+
+    add_PythonPanelServiceServicer_to_server(servicer, server)
+    port = server.add_insecure_port("[::]:0")  # Bind to an available port
+    server.start()
+    yield server, port
+    server.stop(None)
+
+
+def test___panel___has_panel_id_and_panel_uri() -> None:
+    panel = FakePanel(0, "my_panel", "path/to/script")
     assert panel.panel_id == "my_panel"
     assert panel.panel_uri == "path/to/script"
 
 
-def test___connected_panel___set_value___gets_same_value() -> None:
-    panel = nipanel.StreamlitPanel("my_panel", "path/to/script")
+def test___connected_panel___set_value___gets_same_value(
+    grpc_server: tuple[grpc.Server, int],
+) -> None:
+    _, port = grpc_server
+    panel = FakePanel(port, "my_panel", "path/to/script")
     panel.connect()
 
     panel.set_value("test_id", "test_value")
@@ -18,8 +45,9 @@ def test___connected_panel___set_value___gets_same_value() -> None:
     panel.disconnect()
 
 
-def test___with_panel___set_value___gets_same_value() -> None:
-    with nipanel.StreamlitPanel("my_panel", "path/to/script") as panel:
+def test___with_panel___set_value___gets_same_value(grpc_server: tuple[grpc.Server, int]) -> None:
+    _, port = grpc_server
+    with FakePanel(port, "my_panel", "path/to/script") as panel:
         panel.set_value("test_id", "test_value")
 
         # TODO: AB#3095681 - change asserted value to test_value

tests/utils/_fake_panel.py

@@ -0,0 +1,22 @@
+from nipanel._panel import Panel
+
+
+class FakePanel(Panel):
+    """This class allows you to connect to the FakePythonPanelService, for testing."""
+
+    def __init__(self, port: int, panel_id: str, panel_uri: str) -> None:
+        """Create a fake panel, for testing.
+
+        Args:
+            port: The port number for the gRPC server.
+            panel_id: A unique identifier for the panel.
+            panel_uri: The file path of the panel script.
+
+        Returns:
+            A new FakePanel instance.
+        """
+        super().__init__(panel_id, panel_uri)
+        self.port = port
+
+    def _get_channel_url(self) -> str:
+        return f"localhost:{self.port}"

tests/utils/_fake_python_panel_service.py

@@ -1,36 +1,56 @@
-from concurrent import futures
-import grpc
 import time
+from concurrent import futures
+from typing import Any
+
 import google.protobuf.any_pb2 as any_pb2
-from ni.pythonpanel.v1.python_panel_service_pb2 import ConnectRequest, ConnectResponse, DisconnectRequest, DisconnectResponse, GetValueRequest, GetValueResponse, SetValueRequest, SetValueResponse
-from ni.pythonpanel.v1.python_panel_service_pb2_grpc import PythonPanelServiceServicer, add_PythonPanelServiceServicer_to_server
+import grpc
+from ni.pythonpanel.v1.python_panel_service_pb2 import (
+    ConnectRequest,
+    ConnectResponse,
+    DisconnectRequest,
+    DisconnectResponse,
+    GetValueRequest,
+    GetValueResponse,
+    SetValueRequest,
+    SetValueResponse,
+)
+from ni.pythonpanel.v1.python_panel_service_pb2_grpc import (
+    PythonPanelServiceServicer,
+    add_PythonPanelServiceServicer_to_server,
+)
+
 
 class FakePythonPanelService(PythonPanelServiceServicer):
-    def Connect(self, request, context):
-        # Basic implementation for testing
-        print(f"Connecting to panel: {request.panel_id} at {request.panel_uri}")
+    """Fake implementation of the PythonPanelService for testing."""
+
+    _values = {"test_value": any_pb2.Any()}
+
+    def Connect(self, request: ConnectRequest, context: Any) -> ConnectResponse:  # noqa: N802
+        """Just a trivial implementation for testing."""
         return ConnectResponse()
 
-    def Disconnect(self, request, context):
-        # Basic implementation for testing
-        print(f"Disconnecting from panel: {request.panel_id}")
+    def Disconnect(  # noqa: N802
+        self, request: DisconnectRequest, context: Any
+    ) -> DisconnectResponse:
+        """Just a trivial implementation for testing."""
         return DisconnectResponse()
 
-    def GetValue(self, request, context):
-        # Basic implementation for testing
-        print(f"Getting value for panel: {request.panel_id}, value_id: {request.value_id}")
-        value = any_pb2.Any()  # Placeholder for actual value
+    def GetValue(self, request: GetValueRequest, context: Any) -> GetValueResponse:  # noqa: N802
+        """Just a trivial implementation for testing."""
+        value = self._values[request.value_id]
         return GetValueResponse(value=value)
 
-    def SetValue(self, request, context):
-        # Basic implementation for testing
-        print(f"Setting value for panel: {request.panel_id}, value_id: {request.value_id}, value: {request.value}")
+    def SetValue(self, request: SetValueRequest, context: Any) -> SetValueResponse:  # noqa: N802
+        """Just a trivial implementation for testing."""
+        self._values[request.value_id] = request.value
         return SetValueResponse()
 
-def serve():
+
+def serve() -> None:
+    """Run the gRPC server."""
     server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
     add_PythonPanelServiceServicer_to_server(FakePythonPanelService(), server)
-    server.add_insecure_port('[::]:50051') # TODO: do we need to find a free port?
+    server.add_insecure_port("[::]:50051")  # TODO: do we need to find a free port?
     server.start()
     print("Server is running on port 50051...")
     try:
@@ -39,5 +59,6 @@ def serve():
     except KeyboardInterrupt:
         server.stop(0)
 
-if __name__ == '__main__':
-    serve()
+
+if __name__ == "__main__":
+    serve()