cleanup

Author: Mike Prosser

src/nipanel/_streamlit_panel.py

@@ -7,6 +7,8 @@
 class StreamlitPanel(Panel):
     """This class allows you to connect to a Streamlit panel and specify values for its controls."""
 
+    PYTHON_PANEL_SERVICE = "ni.pythonpanel.v1.PythonPanelService"
+
     __slots__ = ()
 
     def __init__(self, panel_id: str, streamlit_script_uri: str) -> None:
@@ -24,7 +26,5 @@ def __init__(self, panel_id: str, streamlit_script_uri: str) -> None:
     def _resolve_service_address(self) -> str:
         with GrpcChannelPool() as grpc_channel_pool:
             discovery_client = DiscoveryClient(grpc_channel_pool=grpc_channel_pool)
-            service_location = discovery_client.resolve_service(
-                "ni.pythonpanel.v1.PythonPanelService"
-            )
-        return service_location.insecure_address
+            service_location = discovery_client.resolve_service(self.PYTHON_PANEL_SERVICE)
+            return service_location.insecure_address

tests/conftest.py

@@ -1,4 +1,4 @@
-"""Fixtures for testing gRPC services."""
+"""Fixtures for testing."""
 
 from concurrent import futures
 from typing import Any, Generator
@@ -14,24 +14,23 @@
 
 
 @pytest.fixture
-def fake_python_panel_service() -> Generator[tuple[grpc.Server, int], Any, None]:
+def fake_python_panel_service() -> Generator[tuple[FakePythonPanelServicer, int], Any, None]:
     """Fixture to create a FakePythonPanelServicer for testing."""
-    # Create an in-process gRPC server
-    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
+    thread_pool = futures.ThreadPoolExecutor(max_workers=10)
+    server = grpc.server(thread_pool)
     servicer = FakePythonPanelServicer()
-
     add_PythonPanelServiceServicer_to_server(servicer, server)
-    port = server.add_insecure_port("[::]:0")  # Bind to an available port
+    port = server.add_insecure_port("[::]:0")
     server.start()
-    yield server, port
+    yield servicer, port
     server.stop(None)
 
 
 @pytest.fixture
 def fake_python_panel_service_stub(
-    fake_python_panel_service: tuple[grpc.Server, int],
+    fake_python_panel_service: tuple[FakePythonPanelServicer, int],
 ) -> Generator[PythonPanelServiceStub, Any, None]:
-    """Fixture to create a gRPC stub for the FakePythonPanelService."""
+    """Fixture to attach a PythonPanelSericeStub to a FakePythonPanelService."""
     _, port = fake_python_panel_service
     channel = grpc.insecure_channel(f"localhost:{port}")
     yield PythonPanelServiceStub(channel)

tests/unit/test_panel.py

@@ -1,4 +1,3 @@
-import grpc
 import pytest
 
 from tests.utils._fake_python_panel_servicer import FakePythonPanelServicer
@@ -12,7 +11,7 @@ def test___panel___has_panel_id_and_panel_uri() -> None:
 
 
 def test___connected_panel___set_value___gets_same_value(
-    fake_python_panel_service: tuple[grpc.Server, int],
+    fake_python_panel_service: tuple[FakePythonPanelServicer, int],
 ) -> None:
     _, port = fake_python_panel_service
     panel = PortPanel(port, "my_panel", "path/to/script")
@@ -26,7 +25,7 @@ def test___connected_panel___set_value___gets_same_value(
 
 
 def test___with_panel___set_value___gets_same_value(
-    fake_python_panel_service: tuple[grpc.Server, int],
+    fake_python_panel_service: tuple[FakePythonPanelServicer, int],
 ) -> None:
     _, port = fake_python_panel_service
     with PortPanel(port, "my_panel", "path/to/script") as panel:
@@ -37,7 +36,7 @@ def test___with_panel___set_value___gets_same_value(
 
 
 def test___new_panel___disconnect___raises_runtime_error(
-    fake_python_panel_service: tuple[grpc.Server, int],
+    fake_python_panel_service: tuple[FakePythonPanelServicer, int],
 ) -> None:
     _, port = fake_python_panel_service
     panel = PortPanel(port, "my_panel", "path/to/script")
@@ -47,12 +46,12 @@ def test___new_panel___disconnect___raises_runtime_error(
 
 
 def test___first_connect_fails___connect___gets_value(
-    fake_python_panel_service: tuple[grpc.Server, int],
+    fake_python_panel_service: tuple[FakePythonPanelServicer, int],
 ) -> None:
     """Test that panel.connect() will automatically retry once."""
+    servicer, port = fake_python_panel_service
     # Simulate a failure on the first connect attempt
-    FakePythonPanelServicer.fail_next_connect()
-    _, port = fake_python_panel_service
+    servicer.fail_next_connect()
     panel = PortPanel(port, "my_panel", "path/to/script")
 
     panel.connect()

tests/utils/_fake_python_panel_servicer.py

@@ -21,7 +21,7 @@ class FakePythonPanelServicer(PythonPanelServiceServicer):
     _fail_next_connect = False
 
     def Connect(self, request: ConnectRequest, context: Any) -> ConnectResponse:  # noqa: N802
-        """Just a trivial implementation for testing."""
+        """Trivial implementation for testing."""
         if self._fail_next_connect:
             self._fail_next_connect = False
             raise ValueError("Simulate a failure to Connect.")
@@ -30,20 +30,19 @@ def Connect(self, request: ConnectRequest, context: Any) -> ConnectResponse:  #
     def Disconnect(  # noqa: N802
         self, request: DisconnectRequest, context: Any
     ) -> DisconnectResponse:
-        """Just a trivial implementation for testing."""
+        """Trivial implementation for testing."""
         return DisconnectResponse()
 
     def GetValue(self, request: GetValueRequest, context: Any) -> GetValueResponse:  # noqa: N802
-        """Just a trivial implementation for testing."""
+        """Trivial implementation for testing."""
         value = self._values[request.value_id]
         return GetValueResponse(value=value)
 
     def SetValue(self, request: SetValueRequest, context: Any) -> SetValueResponse:  # noqa: N802
-        """Just a trivial implementation for testing."""
+        """Trivial implementation for testing."""
         self._values[request.value_id] = request.value
         return SetValueResponse()
 
-    @classmethod
-    def fail_next_connect(cls) -> None:
+    def fail_next_connect(self) -> None:
         """Set whether the Connect method should fail the next time it is called."""
-        cls._fail_next_connect = True
+        self._fail_next_connect = True