move fixture to conftest.py

Author: Mike Prosser

tests/conftest.py

@@ -0,0 +1,38 @@
+"""Fixtures for testing gRPC services."""
+
+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
+
+
+@pytest.fixture
+def fake_python_panel_service() -> Generator[tuple[grpc.Server, int], Any, None]:
+    """Fixture to create a FakePythonPanelServicer for testing."""
+    # Create an in-process gRPC server
+    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
+    servicer = FakePythonPanelServicer()
+
+    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)
+
+
+@pytest.fixture
+def fake_python_panel_service_stub(
+    fake_python_panel_service: tuple[grpc.Server, int],
+) -> Generator[PythonPanelServiceStub, Any, None]:
+    """Fixture to create a gRPC stub for the FakePythonPanelService."""
+    _, port = fake_python_panel_service
+    channel = grpc.insecure_channel(f"localhost:{port}")
+    yield PythonPanelServiceStub(channel)
+    channel.close()

tests/unit/test_fake_python_panel_servicer.py

@@ -0,0 +1,41 @@
+from google.protobuf.any_pb2 import Any
+from ni.pythonpanel.v1.python_panel_service_pb2 import (
+    ConnectRequest,
+    DisconnectRequest,
+    GetValueRequest,
+    SetValueRequest,
+)
+from ni.pythonpanel.v1.python_panel_service_pb2_grpc import PythonPanelServiceStub
+
+
+def test___connect___gets_response(fake_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)
+    assert response is not None  # Ensure response is returned
+
+
+def test___disconnect___gets_response(
+    fake_python_panel_service_stub: PythonPanelServiceStub,
+) -> None:
+    request = DisconnectRequest(panel_id="test_panel")
+    response = fake_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,
+) -> None:
+    request = GetValueRequest(panel_id="test_panel", value_id="test_value")
+    response = fake_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,
+) -> None:
+    value = Any()
+    value.value = b"test_data"
+    request = SetValueRequest(panel_id="test_panel", value_id="test_value", value=value)
+    response = fake_python_panel_service_stub.SetValue(request)
+    assert response is not None  # Ensure response is returned

tests/unit/test_fake_server.py

@@ -1,28 +1,6 @@
-from concurrent import futures
-from typing import Generator
-
 import grpc
-import pytest
-from google.protobuf.any_pb2 import Any
 
 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:
@@ -32,9 +10,9 @@ def test___panel___has_panel_id_and_panel_uri() -> None:
 
 
 def test___connected_panel___set_value___gets_same_value(
-    grpc_server: tuple[grpc.Server, int],
+    fake_python_panel_service: tuple[grpc.Server, int],
 ) -> None:
-    _, port = grpc_server
+    _, port = fake_python_panel_service
     panel = FakePanel(port, "my_panel", "path/to/script")
     panel.connect()
 
@@ -45,8 +23,10 @@ def test___connected_panel___set_value___gets_same_value(
     panel.disconnect()
 
 
-def test___with_panel___set_value___gets_same_value(grpc_server: tuple[grpc.Server, int]) -> None:
-    _, port = grpc_server
+def test___with_panel___set_value___gets_same_value(
+    fake_python_panel_service: tuple[grpc.Server, int],
+) -> None:
+    _, port = fake_python_panel_service
     with FakePanel(port, "my_panel", "path/to/script") as panel:
         panel.set_value("test_id", "test_value")
 

tests/unit/test_panel.py

@@ -20,8 +20,8 @@
 )
 
 
-class FakePythonPanelService(PythonPanelServiceServicer):
-    """Fake implementation of the PythonPanelService for testing."""
+class FakePythonPanelServicer(PythonPanelServiceServicer):
+    """Fake implementation of the PythonPanelServicer for testing."""
 
     _values = {"test_value": any_pb2.Any()}
 
@@ -49,7 +49,7 @@ def SetValue(self, request: SetValueRequest, context: Any) -> SetValueResponse:
 def serve() -> None:
     """Run the gRPC server."""
     server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
-    add_PythonPanelServiceServicer_to_server(FakePythonPanelService(), server)
+    add_PythonPanelServiceServicer_to_server(FakePythonPanelServicer(), server)
     server.add_insecure_port("[::]:50051")  # TODO: do we need to find a free port?
     server.start()
     print("Server is running on port 50051...")