misc feedback and FakePythonPanelService

Author: Mike Prosser

CONTRIBUTING.md

@@ -22,7 +22,7 @@ See [GitHub's official documentation](https://help.github.com/articles/using-pul
 # Getting Started
 
 This is the command to generate the files in /src/ni/pythonpanel/v1/:
-`poetry run python -m grpc_tools.protoc --proto_path=protos --python_out=src/ --grpc_python_out=src/ --plugin=protoc-gen-mypy=.venv\Scripts\protoc-gen-mypy.exe --mypy_out=src/ --mypy_grpc_out=src/ ni/pythonpanel/v1/python_panel_service.proto`
+`poetry run python -m grpc_tools.protoc --proto_path=protos --python_out=src/ --grpc_python_out=src/ --mypy_out=src/ --mypy_grpc_out=src/ ni/pythonpanel/v1/python_panel_service.proto`
 
 # Testing
 

poetry.lock

@@ -15,6 +15,7 @@ ni-measurement-plugin-sdk = {version=">=2.3"}
 [tool.poetry.group.dev.dependencies]
 grpc-stubs = "^1.53"
 types-protobuf = ">=4.21"
+pyupgrade = "^3.19.1"
 
 [tool.poetry.group.lint.dependencies]
 bandit = { version = ">=1.7", extras = ["toml"] }
@@ -41,6 +42,9 @@ build-backend = "poetry.core.masonry.api"
 [tool.ni-python-styleguide]
 extend_exclude = ".tox,docs,src/ni/pythonpanel/v1"
 
+[tool.black]
+extend-exclude = '\.tox/|docs/|src/ni/pythonpanel/v1/'
+
 [tool.mypy]
 files = "examples/,src/nipanel/,tests/"
 namespace_packages = true
@@ -60,9 +64,4 @@ skips = [
 
 [tool.pytest.ini_options]
 addopts = "--doctest-modules --strict-markers"
-testpaths = ["src/nipanel", "tests"]
-
-[tool.black]
-extend-exclude = '''
-/src/ni/pythonpanel/v1/
-'''
+testpaths = ["src/nipanel", "tests"]

pyproject.toml

@@ -3,7 +3,7 @@
 import sys
 from abc import ABC, abstractmethod
 from types import TracebackType
-from typing import TYPE_CHECKING, Optional, Type
+from typing import TYPE_CHECKING
 
 from ni_measurement_plugin_sdk_service.discovery import DiscoveryClient
 
@@ -48,10 +48,10 @@ def __enter__(self) -> Self:
 
     def __exit__(
         self,
-        exctype: Optional[Type[BaseException]],
-        excinst: Optional[BaseException],
-        exctb: Optional[TracebackType],
-    ) -> Optional[bool]:
+        exctype: type[BaseException] | None,
+        excinst: BaseException | None,
+        exctb: TracebackType | None,
+    ) -> bool | None:
         """Exit the runtime context related to this object."""
         self.disconnect()
         return None

src/nipanel/_panel.py

@@ -1,37 +1,42 @@
 """Fixtures for testing."""
 
+from collections.abc import Generator
 from concurrent import futures
-from typing import Any, Generator
 
 import grpc
 import pytest
 from ni.pythonpanel.v1.python_panel_service_pb2_grpc import (
     PythonPanelServiceStub,
-    add_PythonPanelServiceServicer_to_server,
 )
 
-from tests.utils._fake_python_panel_servicer import FakePythonPanelServicer
+from tests.utils._fake_python_panel_service import FakePythonPanelService
 
 
 @pytest.fixture
-def fake_python_panel_service() -> Generator[tuple[FakePythonPanelServicer, int], Any, None]:
+def fake_python_panel_service() -> Generator[FakePythonPanelService]:
     """Fixture to create a FakePythonPanelServicer for testing."""
-    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")
-    server.start()
-    yield servicer, port
-    server.stop(None)
+    with futures.ThreadPoolExecutor(max_workers=10) as thread_pool:
+        service = FakePythonPanelService()
+        service.start(thread_pool)
+        yield service
+        service.stop()
 
 
 @pytest.fixture
-def fake_python_panel_service_stub(
-    fake_python_panel_service: tuple[FakePythonPanelServicer, int],
-) -> Generator[PythonPanelServiceStub, Any, None]:
-    """Fixture to attach a PythonPanelSericeStub to a FakePythonPanelService."""
-    _, port = fake_python_panel_service
-    channel = grpc.insecure_channel(f"localhost:{port}")
-    yield PythonPanelServiceStub(channel)
+def grpc_channel_to_fake_panel_service(
+    fake_python_panel_service: FakePythonPanelService,
+) -> Generator[grpc.Channel]:
+    """Fixture to get a channel to the FakePythonPanelService."""
+    service = fake_python_panel_service
+    channel = grpc.insecure_channel(f"localhost:{service.port}")
+    yield channel
     channel.close()
+
+
+@pytest.fixture
+def python_panel_service_stub(
+    grpc_channel_to_fake_panel_service: grpc.Channel,
+) -> Generator[PythonPanelServiceStub]:
+    """Fixture to get a PythonPanelServiceStub, attached to a FakePythonPanelService."""
+    channel = grpc_channel_to_fake_panel_service
+    yield PythonPanelServiceStub(channel)

tests/conftest.py

@@ -1,4 +1,5 @@
 from google.protobuf.any_pb2 import Any
+from google.protobuf.wrappers_pb2 import StringValue
 from ni.pythonpanel.v1.python_panel_service_pb2 import (
     ConnectRequest,
     DisconnectRequest,
@@ -8,37 +9,38 @@
 from ni.pythonpanel.v1.python_panel_service_pb2_grpc import PythonPanelServiceStub
 
 
-def test___connect___gets_response(fake_python_panel_service_stub: PythonPanelServiceStub) -> None:
+def test___connect___gets_response(python_panel_service_stub: PythonPanelServiceStub) -> None:
     request = ConnectRequest(panel_id="test_panel", panel_uri="path/to/panel")
-    response = fake_python_panel_service_stub.Connect(request)
+    response = python_panel_service_stub.Connect(request)
 
     assert response is not None  # Ensure response is returned
 
 
 def test___disconnect___gets_response(
-    fake_python_panel_service_stub: PythonPanelServiceStub,
+    python_panel_service_stub: PythonPanelServiceStub,
 ) -> None:
     request = DisconnectRequest(panel_id="test_panel")
-    response = fake_python_panel_service_stub.Disconnect(request)
+    response = python_panel_service_stub.Disconnect(request)
 
     assert response is not None  # Ensure response is returned
 
 
 def test___get_value___gets_response(
-    fake_python_panel_service_stub: PythonPanelServiceStub,
+    python_panel_service_stub: PythonPanelServiceStub,
 ) -> None:
     request = GetValueRequest(panel_id="test_panel", value_id="test_value")
-    response = fake_python_panel_service_stub.GetValue(request)
+    response = python_panel_service_stub.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___gets_response(
-    fake_python_panel_service_stub: PythonPanelServiceStub,
+    python_panel_service_stub: PythonPanelServiceStub,
 ) -> None:
-    test_value = Any(value=b"test_data")
+    test_value = Any()
+    test_value.Pack(StringValue(value="test_value"))
     request = SetValueRequest(panel_id="test_panel", value_id="test_value", value=test_value)
-    response = fake_python_panel_service_stub.SetValue(request)
+    response = python_panel_service_stub.SetValue(request)
 
     assert response is not None  # Ensure response is returned

tests/unit/test_fake_python_panel_servicer.py

@@ -1,4 +1,4 @@
-from tests.utils._fake_python_panel_servicer import FakePythonPanelServicer
+from tests.utils._fake_python_panel_service import FakePythonPanelService
 from tests.utils._port_panel import PortPanel
 
 
@@ -9,10 +9,10 @@ def test___panel___has_panel_id_and_panel_uri() -> None:
 
 
 def test___connected_panel___set_value___gets_same_value(
-    fake_python_panel_service: tuple[FakePythonPanelServicer, int],
+    fake_python_panel_service: FakePythonPanelService,
 ) -> None:
-    _, port = fake_python_panel_service
-    panel = PortPanel(port, "my_panel", "path/to/script")
+    service = fake_python_panel_service
+    panel = PortPanel(service.port, "my_panel", "path/to/script")
     panel.connect()
 
     panel.set_value("test_id", "test_value")
@@ -23,24 +23,24 @@ def test___connected_panel___set_value___gets_same_value(
 
 
 def test___with_panel___set_value___gets_same_value(
-    fake_python_panel_service: tuple[FakePythonPanelServicer, int],
+    fake_python_panel_service: FakePythonPanelService,
 ) -> None:
-    _, port = fake_python_panel_service
-    with PortPanel(port, "my_panel", "path/to/script") as panel:
+    service = fake_python_panel_service
+    with PortPanel(service.port, "my_panel", "path/to/script") as panel:
         panel.set_value("test_id", "test_value")
 
         # TODO: AB#3095681 - change asserted value to test_value
         assert panel.get_value("test_id") == "placeholder value"
 
 
 def test___first_connect_fails___connect___gets_value(
-    fake_python_panel_service: tuple[FakePythonPanelServicer, int],
+    fake_python_panel_service: FakePythonPanelService,
 ) -> None:
     """Test that panel.connect() will automatically retry once."""
-    servicer, port = fake_python_panel_service
+    service = fake_python_panel_service
     # Simulate a failure on the first connect attempt
-    servicer.fail_next_connect()
-    panel = PortPanel(port, "my_panel", "path/to/script")
+    service.servicer.fail_next_connect()
+    panel = PortPanel(service.port, "my_panel", "path/to/script")
 
     panel.connect()
 

tests/unit/test_panel.py

@@ -0,0 +1 @@
+"""Test Utilities for the Panel."""

tests/utils/__init__.py

@@ -0,0 +1,42 @@
+from concurrent import futures
+
+import grpc
+from ni.pythonpanel.v1.python_panel_service_pb2_grpc import (
+    add_PythonPanelServiceServicer_to_server,
+)
+
+from tests.utils._fake_python_panel_servicer import FakePythonPanelServicer
+
+
+class FakePythonPanelService:
+    """Encapsulates a fake PythonPanelService with a gRPC server for testing."""
+
+    _server: grpc.Server
+    _port: int
+    _servicer: FakePythonPanelServicer
+
+    def __init__(self) -> None:
+        """Initialize the fake PythonPanelService."""
+        self._servicer = FakePythonPanelServicer()
+
+    def start(self, thread_pool: futures.ThreadPoolExecutor) -> None:
+        """Start the gRPC server and return the port it is bound to."""
+        self._server = grpc.server(thread_pool)
+        add_PythonPanelServiceServicer_to_server(self._servicer, self._server)
+        self._port = self._server.add_insecure_port("[::1]:0")
+        self._server.start()
+
+    def stop(self) -> None:
+        """Stop the gRPC server."""
+        if self._server:
+            self._server.stop(None)
+
+    @property
+    def servicer(self) -> FakePythonPanelServicer:
+        """Get the servicer instance."""
+        return self._servicer
+
+    @property
+    def port(self) -> int:
+        """Get the port the server is bound to."""
+        return self._port