tests: Add integration and acceptance tests for NI-DCPower driver specific session management APIs

pyproject.toml

@@ -124,6 +124,8 @@ module = [
   # https://github.com/briancurtin/deprecation/issues/56 - Add type information (PEP 561)
   "deprecation.*",
   "grpc.framework.foundation.*",
+  # https://github.com/ni/hightime/issues/4 - Add type annotations
+  "hightime.*",
   # https://github.com/microsoft/tracelogging/issues/57 - Python traceloggingdynamic package is missing py.typed marker file
   "traceloggingdynamic",
   # https://github.com/ni/nidaqmx-python/issues/209 - Support type annotations

tests/acceptance/conftest.py

@@ -0,0 +1,10 @@
+"""Pytest configuration file for acceptance tests."""
+import pathlib
+
+import pytest
+
+
+@pytest.fixture(scope="module")
+def pin_map_directory(test_assets_directory: pathlib.Path) -> pathlib.Path:
+    """Test fixture that returns the pin map directory."""
+    return test_assets_directory / "acceptance" / "session_management"

tests/acceptance/test_nidcpower_measurement.py

@@ -0,0 +1,110 @@
+import pathlib
+from typing import Generator, Iterable, NamedTuple
+
+import pytest
+
+from ni_measurementlink_service._internal.stubs.ni.measurementlink.measurement.v2.measurement_service_pb2 import (
+    MeasureRequest,
+)
+from ni_measurementlink_service._internal.stubs.ni.measurementlink.measurement.v2.measurement_service_pb2_grpc import (
+    MeasurementServiceStub,
+)
+from ni_measurementlink_service._internal.stubs.ni.measurementlink.pin_map_context_pb2 import (
+    PinMapContext,
+)
+from ni_measurementlink_service.measurement.service import MeasurementService
+from tests.assets.stubs.nidcpower_measurement.types_pb2 import (
+    Configurations,
+    Outputs,
+)
+from tests.utilities import nidcpower_measurement
+from tests.utilities.pin_map_client import PinMapClient
+
+_SITE = 0
+
+
+def test___single_session___measure___returns_measured_values(
+    pin_map_context: PinMapContext,
+    stub_v2: MeasurementServiceStub,
+) -> None:
+    configurations = Configurations(pin_names=["Pin1"], multi_session=False)
+
+    outputs = _measure(stub_v2, pin_map_context, configurations)
+
+    assert outputs.voltage_measurements == [5]
+    assert outputs.current_measurements == [0.0001]
+
+
+def test___single_session___measure___creates_single_session(
+    pin_map_context: PinMapContext,
+    stub_v2: MeasurementServiceStub,
+) -> None:
+    configurations = Configurations(pin_names=["Pin1"], multi_session=False)
+
+    outputs = _measure(stub_v2, pin_map_context, configurations)
+
+    assert _get_output(outputs) == [
+        _MeasurementOutput("DCPower1/0", "DCPower1/0", "DCPower1/0", "DCPower1/0")
+    ]
+
+
+def test___multiple_sessions___measure___creates_multiple_sessions(
+    pin_map_context: PinMapContext,
+    stub_v2: MeasurementServiceStub,
+) -> None:
+    configurations = Configurations(pin_names=["Pin1", "Pin2"], multi_session=True)
+
+    outputs = _measure(stub_v2, pin_map_context, configurations)
+
+    assert _get_output(outputs) == [
+        _MeasurementOutput("DCPower1/0", "DCPower1/0", "DCPower1/0", "DCPower1/0"),
+        _MeasurementOutput("DCPower1/2", "DCPower1/2", "DCPower1/2", "DCPower1/2"),
+    ]
+
+
+def _measure(
+    stub_v2: MeasurementServiceStub,
+    pin_map_context: PinMapContext,
+    configurations: Configurations,
+) -> Outputs:
+    request = MeasureRequest(pin_map_context=pin_map_context)
+    request.configuration_parameters.Pack(configurations)
+    response_iterator = stub_v2.Measure(request)
+    responses = list(response_iterator)
+    assert len(responses) == 1
+    outputs = Outputs.FromString(responses[0].outputs.value)
+    return outputs
+
+
+@pytest.fixture(scope="module")
+def measurement_service() -> Generator[MeasurementService, None, None]:
+    """Test fixture that creates and hosts a measurement service."""
+    with nidcpower_measurement.measurement_service.host_service() as service:
+        yield service
+
+
+@pytest.fixture
+def pin_map_context(pin_map_client: PinMapClient, pin_map_directory: pathlib.Path) -> PinMapContext:
+    pin_map_name = "1Smu2ChannelGroup2Pin1Site.pinmap"
+    pin_map_id = pin_map_client.update_pin_map(pin_map_directory / pin_map_name)
+
+    return PinMapContext(pin_map_id=pin_map_id, sites=[_SITE])
+
+
+class _MeasurementOutput(NamedTuple):
+    session_name: str
+    resource_name: str
+    channel_list: str
+    connected_channels: str
+
+
+def _get_output(outputs: Outputs) -> Iterable[_MeasurementOutput]:
+    return [
+        _MeasurementOutput(session_name, resource_name, channel_list, connected_channels)
+        for session_name, resource_name, channel_list, connected_channels in zip(
+            outputs.session_names,
+            outputs.resource_names,
+            outputs.channel_lists,
+            outputs.connected_channels,
+        )
+    ]

tests/acceptance/test_session_management.py

@@ -297,9 +297,3 @@ def measurement_service(
     """Test fixture that creates and hosts a measurement service."""
     with pin_aware_measurement.measurement_service.host_service() as service:
         yield service
-
-
-@pytest.fixture
-def pin_map_directory(test_assets_directory: pathlib.Path) -> pathlib.Path:
-    """Test fixture that returns the pin map directory."""
-    return test_assets_directory / "acceptance" / "session_management"

tests/assets/acceptance/session_management/1Smu2ChannelGroup2Pin1Site.pinmap

@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="utf-8"?>
+<PinMap xmlns="http://www.ni.com/TestStand/SemiconductorModule/PinMap.xsd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" schemaVersion="1.6">
+	<Instruments>
+		<NIDCPowerInstrument name="DCPower1" numberOfChannels="4">
+			<ChannelGroup name="DCPowerChannelGroup0" channels="0,1" />
+			<ChannelGroup name="DCPowerChannelGroup1" channels="2,3" />
+		</NIDCPowerInstrument>
+	</Instruments>
+	<Pins>
+		<DUTPin name="Pin1" />
+		<DUTPin name="Pin2" />
+	</Pins>
+	<PinGroups></PinGroups>
+	<Sites>
+		<Site siteNumber="0" />
+	</Sites>
+	<Connections>
+		<Connection pin="Pin1" siteNumber="0" instrument="DCPower1" channel="0" />
+		<Connection pin="Pin2" siteNumber="0" instrument="DCPower1" channel="2" />
+	</Connections>
+</PinMap>

tests/assets/integration/session_management/1Smu2ChannelGroup2Pin1Site.pinmap

@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="utf-8"?>
+<PinMap xmlns="http://www.ni.com/TestStand/SemiconductorModule/PinMap.xsd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" schemaVersion="1.6">
+	<Instruments>
+		<NIDCPowerInstrument name="DCPower1" numberOfChannels="4">
+			<ChannelGroup name="DCPowerChannelGroup0" channels="0,1" />
+			<ChannelGroup name="DCPowerChannelGroup1" channels="2,3" />
+		</NIDCPowerInstrument>
+	</Instruments>
+	<Pins>
+		<DUTPin name="Pin1" />
+		<DUTPin name="Pin2" />
+	</Pins>
+	<PinGroups></PinGroups>
+	<Sites>
+		<Site siteNumber="0" />
+	</Sites>
+	<Connections>
+		<Connection pin="Pin1" siteNumber="0" instrument="DCPower1" channel="0" />
+		<Connection pin="Pin2" siteNumber="0" instrument="DCPower1" channel="2" />
+	</Connections>
+</PinMap>

tests/assets/stubs/__init__.py

@@ -0,0 +1 @@
+"""Auto generated gRPC files."""

tests/assets/stubs/nidcpower_measurement/__init__.py

@@ -0,0 +1 @@
+"""Auto generated gRPC files for nidcpower test measurement."""

tests/assets/stubs/nidcpower_measurement/types.proto

@@ -0,0 +1,16 @@
+syntax = "proto3";
+package ni.measurementlink.measurement.tests.nidcpower_measurement;
+
+message Configurations {
+    repeated string pin_names = 1;
+    bool multi_session = 2;
+}
+
+message Outputs {
+    repeated string session_names = 1;
+    repeated string resource_names = 2;
+    repeated string channel_lists = 3;
+    repeated string connected_channels = 4;
+    repeated double voltage_measurements = 5;
+    repeated double current_measurements = 6;
+}

tests/assets/stubs/nidcpower_measurement/types_pb2.pyi

@@ -0,0 +1,72 @@
+"""
+@generated by mypy-protobuf.  Do not edit manually!
+isort:skip_file
+"""
+import builtins
+import collections.abc
+import google.protobuf.descriptor
+import google.protobuf.internal.containers
+import google.protobuf.message
+import sys
+
+if sys.version_info >= (3, 8):
+    import typing as typing_extensions
+else:
+    import typing_extensions
+
+DESCRIPTOR: google.protobuf.descriptor.FileDescriptor
+
+@typing_extensions.final
+class Configurations(google.protobuf.message.Message):
+    DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+    PIN_NAMES_FIELD_NUMBER: builtins.int
+    MULTI_SESSION_FIELD_NUMBER: builtins.int
+    @property
+    def pin_names(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: ...
+    multi_session: builtins.bool
+    def __init__(
+        self,
+        *,
+        pin_names: collections.abc.Iterable[builtins.str] | None = ...,
+        multi_session: builtins.bool = ...,
+    ) -> None: ...
+    def ClearField(self, field_name: typing_extensions.Literal["multi_session", b"multi_session", "pin_names", b"pin_names"]) -> None: ...
+
+global___Configurations = Configurations
+
+@typing_extensions.final
+class Outputs(google.protobuf.message.Message):
+    DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+    SESSION_NAMES_FIELD_NUMBER: builtins.int
+    RESOURCE_NAMES_FIELD_NUMBER: builtins.int
+    CHANNEL_LISTS_FIELD_NUMBER: builtins.int
+    CONNECTED_CHANNELS_FIELD_NUMBER: builtins.int
+    VOLTAGE_MEASUREMENTS_FIELD_NUMBER: builtins.int
+    CURRENT_MEASUREMENTS_FIELD_NUMBER: builtins.int
+    @property
+    def session_names(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: ...
+    @property
+    def resource_names(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: ...
+    @property
+    def channel_lists(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: ...
+    @property
+    def connected_channels(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: ...
+    @property
+    def voltage_measurements(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.float]: ...
+    @property
+    def current_measurements(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.float]: ...
+    def __init__(
+        self,
+        *,
+        session_names: collections.abc.Iterable[builtins.str] | None = ...,
+        resource_names: collections.abc.Iterable[builtins.str] | None = ...,
+        channel_lists: collections.abc.Iterable[builtins.str] | None = ...,
+        connected_channels: collections.abc.Iterable[builtins.str] | None = ...,
+        voltage_measurements: collections.abc.Iterable[builtins.float] | None = ...,
+        current_measurements: collections.abc.Iterable[builtins.float] | None = ...,
+    ) -> None: ...
+    def ClearField(self, field_name: typing_extensions.Literal["channel_lists", b"channel_lists", "connected_channels", b"connected_channels", "current_measurements", b"current_measurements", "resource_names", b"resource_names", "session_names", b"session_names", "voltage_measurements", b"voltage_measurements"]) -> None: ...
+
+global___Outputs = Outputs

tests/assets/stubs/nidcpower_measurement/types_pb2_grpc.py

@@ -0,0 +1,4 @@
+# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
+"""Client and server classes corresponding to protobuf-defined services."""
+import grpc
+

tests/assets/stubs/nidcpower_measurement/types_pb2_grpc.pyi

@@ -0,0 +1,17 @@
+"""
+@generated by mypy-protobuf.  Do not edit manually!
+isort:skip_file
+"""
+import abc
+import collections.abc
+import grpc
+import grpc.aio
+import typing
+
+_T = typing.TypeVar('_T')
+
+class _MaybeAsyncIterator(collections.abc.AsyncIterator[_T], collections.abc.Iterator[_T], metaclass=abc.ABCMeta):
+    ...
+
+class _ServicerContext(grpc.ServicerContext, grpc.aio.ServicerContext):  # type: ignore
+    ...

tests/integration/session_management/__init__.py

@@ -0,0 +1 @@
+"""Integration tests for driver-specific session management APIs."""

tests/integration/session_management/conftest.py

@@ -0,0 +1,10 @@
+"""Pytest configuration file for integration tests."""
+import pathlib
+
+import pytest
+
+
+@pytest.fixture(scope="module")
+def pin_map_directory(test_assets_directory: pathlib.Path) -> pathlib.Path:
+    """Test fixture that returns the pin map directory."""
+    return test_assets_directory / "integration" / "session_management"