Add support for builtin types for get_value and set_value

src/nipanel/_converters.py

@@ -3,24 +3,25 @@
 from __future__ import annotations
 
 import logging
-import typing
+from typing import Callable, NamedTuple, Union
 
 import google.protobuf.any_pb2
-import google.protobuf.wrappers_pb2
+from google.protobuf import wrappers_pb2
+from typing_extensions import TypeAlias
 
 _logger = logging.getLogger(__name__)
 
 
-_builtin_protobuf_type = (
-    google.protobuf.wrappers_pb2.BoolValue
-    | google.protobuf.wrappers_pb2.BytesValue
-    | google.protobuf.wrappers_pb2.DoubleValue
-    | google.protobuf.wrappers_pb2.Int64Value
-    | google.protobuf.wrappers_pb2.StringValue
-)
+_builtin_protobuf_type: TypeAlias = Union[
+    wrappers_pb2.BoolValue,
+    wrappers_pb2.BytesValue,
+    wrappers_pb2.DoubleValue,
+    wrappers_pb2.Int64Value,
+    wrappers_pb2.StringValue,
+]
 
 
-class ConvertibleType(typing.NamedTuple):
+class ConvertibleType(NamedTuple):
     """A Python type that can be converted to and from a protobuf Any."""
 
     name: str
@@ -32,11 +33,11 @@ class ConvertibleType(typing.NamedTuple):
     protobuf_typename: str
     """The protobuf name for the type."""
 
-    protobuf_initializer: typing.Callable[..., _builtin_protobuf_type]
+    protobuf_initializer: Callable[..., _builtin_protobuf_type]
     """A callable that can be used to create an instance of the protobuf type."""
 
 
-_CONVERTIBLE_TYPES: dict[str, ConvertibleType] = {
+_CONVERTIBLE_TYPES = {
     "bool": ConvertibleType(
         name="Boolean",
         python_typename=bool.__name__,

tests/conftest.py

@@ -1,10 +1,10 @@
 """Fixtures for testing."""
 
 from collections.abc import Generator
-from concurrent import futures
 
 import grpc
 import pytest
+from grpc.framework.foundation import logging_pool  # type: ignore # types-grpcio does not cover this yet
 from ni.pythonpanel.v1.python_panel_service_pb2_grpc import (
     PythonPanelServiceStub,
 )
@@ -15,15 +15,15 @@
 @pytest.fixture
 def fake_python_panel_service() -> Generator[FakePythonPanelService]:
     """Fixture to create a FakePythonPanelServicer for testing."""
-    with futures.ThreadPoolExecutor(max_workers=10) as thread_pool:
+    with logging_pool.pool(max_workers=10) as thread_pool:
         service = FakePythonPanelService()
         service.start(thread_pool)
         yield service
         service.stop()
 
 
 @pytest.fixture
-def grpc_channel_for_fake_panel_service(
+def fake_panel_channel(
     fake_python_panel_service: FakePythonPanelService,
 ) -> Generator[grpc.Channel]:
     """Fixture to get a channel to the FakePythonPanelService."""
@@ -35,8 +35,7 @@ def grpc_channel_for_fake_panel_service(
 
 @pytest.fixture
 def python_panel_service_stub(
-    grpc_channel_for_fake_panel_service: grpc.Channel,
+    fake_panel_channel: grpc.Channel,
 ) -> Generator[PythonPanelServiceStub]:
     """Fixture to get a PythonPanelServiceStub, attached to a FakePythonPanelService."""
-    channel = grpc_channel_for_fake_panel_service
-    yield PythonPanelServiceStub(channel)
+    yield PythonPanelServiceStub(fake_panel_channel)

tests/types.py

@@ -0,0 +1,59 @@
+"""Types that support conversion to and from protobuf."""
+
+import enum
+
+
+class MyIntFlags(enum.IntFlag):
+    """Example of an IntFlag enum."""
+
+    VALUE1 = 1
+    VALUE2 = 2
+    VALUE4 = 4
+
+
+class MyIntEnum(enum.IntEnum):
+    """Example of an IntEnum enum."""
+
+    VALUE10 = 10
+    VALUE20 = 20
+    VALUE30 = 30
+
+
+class MixinIntEnum(int, enum.Enum):
+    """Example of an IntEnum using a mixin."""
+
+    VALUE11 = 11
+    VALUE22 = 22
+    VALUE33 = 33
+
+
+class MyStrEnum(enum.StrEnum):
+    """Example of a StrEnum enum."""
+
+    VALUE1 = "value1"
+    VALUE2 = "value2"
+    VALUE3 = "value3"
+
+
+class MixinStrEnum(str, enum.Enum):
+    """Example of a StrEnum using a mixin."""
+
+    VALUE11 = "value11"
+    VALUE22 = "value22"
+    VALUE33 = "value33"
+
+
+class MyEnum(enum.Enum):
+    """Example of a simple enum."""
+
+    VALUE100 = 100
+    VALUE200 = 200
+    VALUE300 = 300
+
+
+class MyFlags(enum.Flag):
+    """Example of a simple flag."""
+
+    VALUE8 = 8
+    VALUE16 = 16
+    VALUE32 = 32

tests/unit/test_streamlit_panel.py

@@ -1,6 +1,7 @@
 import grpc
 import pytest
 
+import tests.types as test_types
 from nipanel._streamlit_panel import StreamlitPanel
 from tests.utils._fake_python_panel_service import FakePythonPanelService
 
@@ -12,10 +13,9 @@ def test___panel___has_panel_id_and_panel_uri() -> None:
 
 
 def test___opened_panel___set_value___gets_same_value(
-    grpc_channel_for_fake_panel_service: grpc.Channel,
+    fake_panel_channel: grpc.Channel,
 ) -> None:
-    channel = grpc_channel_for_fake_panel_service
-    panel = StreamlitPanel("my_panel", "path/to/script", grpc_channel=channel)
+    panel = StreamlitPanel("my_panel", "path/to/script", grpc_channel=fake_panel_channel)
     panel.open_panel()
 
     value_id = "test_id"
@@ -27,14 +27,13 @@ def test___opened_panel___set_value___gets_same_value(
 
 def test___first_open_panel_fails___open_panel___gets_value(
     fake_python_panel_service: FakePythonPanelService,
-    grpc_channel_for_fake_panel_service: grpc.Channel,
+    fake_panel_channel: grpc.Channel,
 ) -> None:
     """Test that panel.open_panel() will automatically retry once."""
-    channel = grpc_channel_for_fake_panel_service
     service = fake_python_panel_service
     # Simulate a failure on the first attempt
     service.servicer.fail_next_open_panel()
-    panel = StreamlitPanel("my_panel", "path/to/script", grpc_channel=channel)
+    panel = StreamlitPanel("my_panel", "path/to/script", grpc_channel=fake_panel_channel)
 
     panel.open_panel()
 
@@ -45,11 +44,10 @@ def test___first_open_panel_fails___open_panel___gets_value(
 
 
 def test___unopened_panel___set_value___sets_value(
-    grpc_channel_for_fake_panel_service: grpc.Channel,
+    fake_panel_channel: grpc.Channel,
 ) -> None:
     """Test that set_value() succeeds before the user opens the panel."""
-    channel = grpc_channel_for_fake_panel_service
-    panel = StreamlitPanel("my_panel", "path/to/script", grpc_channel=channel)
+    panel = StreamlitPanel("my_panel", "path/to/script", grpc_channel=fake_panel_channel)
 
     value_id = "test_id"
     string_value = "test_value"
@@ -59,23 +57,21 @@ def test___unopened_panel___set_value___sets_value(
 
 
 def test___unopened_panel___get_unset_value___raises_exception(
-    grpc_channel_for_fake_panel_service: grpc.Channel,
+    fake_panel_channel: grpc.Channel,
 ) -> None:
     """Test that get_value() raises an exception for an unset value."""
-    channel = grpc_channel_for_fake_panel_service
-    panel = StreamlitPanel("my_panel", "path/to/script", grpc_channel=channel)
+    panel = StreamlitPanel("my_panel", "path/to/script", grpc_channel=fake_panel_channel)
 
     value_id = "test_id"
     with pytest.raises(grpc.RpcError):
         panel.get_value(value_id)
 
 
 def test___unopened_panel___get_set_value___gets_value(
-    grpc_channel_for_fake_panel_service: grpc.Channel,
+    fake_panel_channel: grpc.Channel,
 ) -> None:
     """Test that get_value() succeeds for a set value before the user opens the panel."""
-    channel = grpc_channel_for_fake_panel_service
-    panel = StreamlitPanel("my_panel", "path/to/script", grpc_channel=channel)
+    panel = StreamlitPanel("my_panel", "path/to/script", grpc_channel=fake_panel_channel)
 
     value_id = "test_id"
     string_value = "test_value"
@@ -84,31 +80,48 @@ def test___unopened_panel___get_set_value___gets_value(
     assert panel.get_value(value_id) == string_value
 
 
-def test___builtin_scalar_types___set_value___gets_same_value(
-    grpc_channel_for_fake_panel_service: grpc.Channel,
+@pytest.mark.parametrize(
+    "value_payload",
+    [
+        "firstname bunchanumbers",
+        42,
+        3.14,
+        True,
+        b"robotext",
+        test_types.MyIntFlags.VALUE1 | test_types.MyIntFlags.VALUE4,
+        test_types.MyIntEnum.VALUE20,
+        test_types.MyStrEnum.VALUE3,
+        test_types.MixinIntEnum.VALUE33,
+        test_types.MixinStrEnum.VALUE11,
+    ],
+)
+def test___builtin_scalar_type___set_value___gets_same_value(
+    fake_panel_channel: grpc.Channel,
+    value_payload: object,
 ) -> None:
     """Test that set_value() and get_value() work for builtin scalar types."""
-    channel = grpc_channel_for_fake_panel_service
-    panel = StreamlitPanel("my_panel", "path/to/script", grpc_channel=channel)
+    panel = StreamlitPanel("my_panel", "path/to/script", grpc_channel=fake_panel_channel)
 
     value_id = "test_id"
-    string_value = "test_value"
-    int_value = 42
-    float_value = 3.14
-    bool_value = True
-    bytes_value = b"robotext"
-
-    panel.set_value(value_id, string_value)
-    assert panel.get_value(value_id) == string_value
+    panel.set_value(value_id, value_payload)
 
-    panel.set_value(value_id, int_value)
-    assert panel.get_value(value_id) == int_value
+    assert panel.get_value(value_id) == value_payload
 
-    panel.set_value(value_id, float_value)
-    assert panel.get_value(value_id) == float_value
 
-    panel.set_value(value_id, bool_value)
-    assert panel.get_value(value_id) == bool_value
+@pytest.mark.parametrize(
+    "value_payload",
+    [
+        test_types.MyEnum.VALUE300,
+        test_types.MyFlags.VALUE8 | test_types.MyFlags.VALUE16,
+    ],
+)
+def test___unsupported_type___set_value___raises(
+    fake_panel_channel: grpc.Channel,
+    value_payload: object,
+) -> None:
+    """Test that set_value() raises for unsupported types."""
+    panel = StreamlitPanel("my_panel", "path/to/script", grpc_channel=fake_panel_channel)
 
-    panel.set_value(value_id, bytes_value)
-    assert panel.get_value(value_id) == bytes_value
+    value_id = "test_id"
+    with pytest.raises(TypeError):
+        panel.set_value(value_id, value_payload)

tests/utils/_fake_python_panel_servicer.py

@@ -16,8 +16,10 @@
 class FakePythonPanelServicer(PythonPanelServiceServicer):
     """Fake implementation of the PythonPanelServicer for testing."""
 
-    _values = {"test_value": any_pb2.Any()}
-    _fail_next_open_panel = False
+    def __init__(self) -> None:
+        """Initialize the fake PythonPanelServicer."""
+        self._values = {"test_value": any_pb2.Any()}
+        self._fail_next_open_panel = False
 
     def OpenPanel(self, request: OpenPanelRequest, context: Any) -> OpenPanelResponse:  # noqa: N802
         """Trivial implementation for testing."""