diff --git a/tests/acceptance/test_nidigital_measurement.py b/tests/acceptance/test_nidigital_measurement.py new file mode 100644 index 000000000..6af946418 --- /dev/null +++ b/tests/acceptance/test_nidigital_measurement.py @@ -0,0 +1,121 @@ +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.nidigital_measurement.types_pb2 import Configurations, Outputs +from tests.utilities import nidigital_measurement +from tests.utilities.pin_map_client import PinMapClient + + +def test___single_session___measure___returns_measured_values( + pin_map_id: str, + stub_v2: MeasurementServiceStub, +) -> None: + pin_map_context = PinMapContext(pin_map_id=pin_map_id, sites=[0]) + configurations = Configurations(pin_names=["CS", "SCLK", "MOSI", "MISO"], multi_session=False) + + outputs = _measure(stub_v2, pin_map_context, configurations) + + assert outputs.passing_sites == [0] + assert outputs.failing_sites == [] + + +def test___single_session___measure___creates_single_session( + pin_map_id: str, + stub_v2: MeasurementServiceStub, +) -> None: + pin_map_context = PinMapContext(pin_map_id=pin_map_id, sites=[0]) + configurations = Configurations(pin_names=["CS", "SCLK", "MOSI", "MISO"], multi_session=False) + + outputs = _measure(stub_v2, pin_map_context, configurations) + + assert _get_output(outputs) == [ + _MeasurementOutput( + "DigitalPattern1", + "DigitalPattern1", + "site0/CS, site0/SCLK, site0/MOSI, site0/MISO", + "site0/CS", + ) + ] + + +def test___multiple_sessions___measure___creates_multiple_sessions( + pin_map_id: str, + stub_v2: MeasurementServiceStub, +) -> None: + pin_map_context = PinMapContext(pin_map_id=pin_map_id, sites=[0, 1]) + configurations = Configurations(pin_names=["CS", "SCLK", "MOSI", "MISO"], multi_session=True) + + outputs = _measure(stub_v2, pin_map_context, configurations) + + assert _get_output(outputs) == [ + _MeasurementOutput( + "DigitalPattern1", + "DigitalPattern1", + "site0/CS, site0/SCLK, site0/MOSI, site0/MISO", + "site0/CS, site0/SCLK, site0/MOSI, site0/MISO", + ), + _MeasurementOutput( + "DigitalPattern2", + "DigitalPattern2", + "site1/CS, site1/SCLK, site1/MOSI, site1/MISO", + "site1/CS, site1/SCLK, site1/MOSI, site1/MISO", + ), + ] + + +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 nidigital_measurement.measurement_service.host_service() as service: + yield service + + +@pytest.fixture +def pin_map_id(pin_map_client: PinMapClient, pin_map_directory: pathlib.Path) -> str: + pin_map_name = "2Digital2Group4Pin1Site.pinmap" + return pin_map_client.update_pin_map(pin_map_directory / pin_map_name) + + +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, + ) + ] diff --git a/tests/acceptance/test_niswitch_measurement.py b/tests/acceptance/test_niswitch_measurement.py new file mode 100644 index 000000000..2b156ddcf --- /dev/null +++ b/tests/acceptance/test_niswitch_measurement.py @@ -0,0 +1,96 @@ +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.niswitch_measurement.types_pb2 import ( + Configurations, + Outputs, +) +from tests.utilities import niswitch_measurement +from tests.utilities.pin_map_client import PinMapClient + +_SITE = 0 + + +def test___single_session___measure___creates_single_session( + pin_map_context: PinMapContext, + stub_v2: MeasurementServiceStub, +) -> None: + configurations = Configurations(relay_names=["SiteRelay1"], multi_session=False) + + outputs = _measure(stub_v2, pin_map_context, configurations) + + assert _get_output(outputs) == [_MeasurementOutput("RelayDriver1", "RelayDriver1", "K0", "K0")] + + +def test___multiple_sessions___measure___creates_multiple_sessions( + pin_map_context: PinMapContext, + stub_v2: MeasurementServiceStub, +) -> None: + configurations = Configurations(relay_names=["SiteRelay1", "SiteRelay2"], multi_session=True) + + outputs = _measure(stub_v2, pin_map_context, configurations) + + assert _get_output(outputs) == [ + _MeasurementOutput("RelayDriver1", "RelayDriver1", "K0", "K0"), + _MeasurementOutput("RelayDriver2", "RelayDriver2", "K1", "K1"), + ] + + +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 niswitch_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 = "2Switch2Relay1Site.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, + ) + ] diff --git a/tests/assets/acceptance/session_management/2Digital2Group4Pin1Site.pinmap b/tests/assets/acceptance/session_management/2Digital2Group4Pin1Site.pinmap new file mode 100644 index 000000000..1054c1965 --- /dev/null +++ b/tests/assets/acceptance/session_management/2Digital2Group4Pin1Site.pinmap @@ -0,0 +1,28 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/tests/assets/acceptance/session_management/2Switch2Relay1Site.pinmap b/tests/assets/acceptance/session_management/2Switch2Relay1Site.pinmap new file mode 100644 index 000000000..cf9e89756 --- /dev/null +++ b/tests/assets/acceptance/session_management/2Switch2Relay1Site.pinmap @@ -0,0 +1,21 @@ + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/tests/assets/integration/session_management/2Digital2Group4Pin1Site.pinmap b/tests/assets/integration/session_management/2Digital2Group4Pin1Site.pinmap new file mode 100644 index 000000000..1054c1965 --- /dev/null +++ b/tests/assets/integration/session_management/2Digital2Group4Pin1Site.pinmap @@ -0,0 +1,28 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/tests/assets/integration/session_management/2Switch2Relay1Site.pinmap b/tests/assets/integration/session_management/2Switch2Relay1Site.pinmap new file mode 100644 index 000000000..cf9e89756 --- /dev/null +++ b/tests/assets/integration/session_management/2Switch2Relay1Site.pinmap @@ -0,0 +1,21 @@ + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/tests/assets/stubs/nidigital_measurement/__init__.py b/tests/assets/stubs/nidigital_measurement/__init__.py new file mode 100644 index 000000000..10eddb732 --- /dev/null +++ b/tests/assets/stubs/nidigital_measurement/__init__.py @@ -0,0 +1 @@ +"""Auto generated gRPC files for nidigital test measurement.""" diff --git a/tests/assets/stubs/nidigital_measurement/types.proto b/tests/assets/stubs/nidigital_measurement/types.proto new file mode 100644 index 000000000..1ce30e29a --- /dev/null +++ b/tests/assets/stubs/nidigital_measurement/types.proto @@ -0,0 +1,16 @@ +syntax = "proto3"; +package ni.measurementlink.measurement.tests.nidigital_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 int32 passing_sites = 5; + repeated int32 failing_sites = 6; +} \ No newline at end of file diff --git a/tests/assets/stubs/nidigital_measurement/types_pb2.py b/tests/assets/stubs/nidigital_measurement/types_pb2.py new file mode 100644 index 000000000..177fcdc73 --- /dev/null +++ b/tests/assets/stubs/nidigital_measurement/types_pb2.py @@ -0,0 +1,27 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: stubs/nidigital_measurement/types.proto +"""Generated protocol buffer code.""" +from google.protobuf.internal import builder as _builder +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import symbol_database as _symbol_database +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\'stubs/nidigital_measurement/types.proto\x12:ni.measurementlink.measurement.tests.nidigital_measurement\":\n\x0e\x43onfigurations\x12\x11\n\tpin_names\x18\x01 \x03(\t\x12\x15\n\rmulti_session\x18\x02 \x01(\x08\"\x99\x01\n\x07Outputs\x12\x15\n\rsession_names\x18\x01 \x03(\t\x12\x16\n\x0eresource_names\x18\x02 \x03(\t\x12\x15\n\rchannel_lists\x18\x03 \x03(\t\x12\x1a\n\x12\x63onnected_channels\x18\x04 \x03(\t\x12\x15\n\rpassing_sites\x18\x05 \x03(\x05\x12\x15\n\rfailing_sites\x18\x06 \x03(\x05\x62\x06proto3') + +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals()) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'stubs.nidigital_measurement.types_pb2', globals()) +if _descriptor._USE_C_DESCRIPTORS == False: + + DESCRIPTOR._options = None + _CONFIGURATIONS._serialized_start=103 + _CONFIGURATIONS._serialized_end=161 + _OUTPUTS._serialized_start=164 + _OUTPUTS._serialized_end=317 +# @@protoc_insertion_point(module_scope) diff --git a/tests/assets/stubs/nidigital_measurement/types_pb2.pyi b/tests/assets/stubs/nidigital_measurement/types_pb2.pyi new file mode 100644 index 000000000..c448f2b36 --- /dev/null +++ b/tests/assets/stubs/nidigital_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 + PASSING_SITES_FIELD_NUMBER: builtins.int + FAILING_SITES_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 passing_sites(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.int]: ... + @property + def failing_sites(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.int]: ... + 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 = ..., + passing_sites: collections.abc.Iterable[builtins.int] | None = ..., + failing_sites: collections.abc.Iterable[builtins.int] | None = ..., + ) -> None: ... + def ClearField(self, field_name: typing_extensions.Literal["channel_lists", b"channel_lists", "connected_channels", b"connected_channels", "failing_sites", b"failing_sites", "passing_sites", b"passing_sites", "resource_names", b"resource_names", "session_names", b"session_names"]) -> None: ... + +global___Outputs = Outputs diff --git a/tests/assets/stubs/nidigital_measurement/types_pb2_grpc.py b/tests/assets/stubs/nidigital_measurement/types_pb2_grpc.py new file mode 100644 index 000000000..2daafffeb --- /dev/null +++ b/tests/assets/stubs/nidigital_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 + diff --git a/tests/assets/stubs/nidigital_measurement/types_pb2_grpc.pyi b/tests/assets/stubs/nidigital_measurement/types_pb2_grpc.pyi new file mode 100644 index 000000000..b13382f6a --- /dev/null +++ b/tests/assets/stubs/nidigital_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 + ... diff --git a/tests/assets/stubs/niswitch_measurement/__init__.py b/tests/assets/stubs/niswitch_measurement/__init__.py new file mode 100644 index 000000000..aa66b3a5d --- /dev/null +++ b/tests/assets/stubs/niswitch_measurement/__init__.py @@ -0,0 +1 @@ +"""Auto generated gRPC files for niswitch test measurement.""" diff --git a/tests/assets/stubs/niswitch_measurement/types.proto b/tests/assets/stubs/niswitch_measurement/types.proto new file mode 100644 index 000000000..7ff340cb0 --- /dev/null +++ b/tests/assets/stubs/niswitch_measurement/types.proto @@ -0,0 +1,14 @@ +syntax = "proto3"; +package ni.measurementlink.measurement.tests.niswitch_measurement; + +message Configurations { + repeated string relay_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; +} \ No newline at end of file diff --git a/tests/assets/stubs/niswitch_measurement/types_pb2.py b/tests/assets/stubs/niswitch_measurement/types_pb2.py new file mode 100644 index 000000000..dbe3635b2 --- /dev/null +++ b/tests/assets/stubs/niswitch_measurement/types_pb2.py @@ -0,0 +1,27 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: stubs/niswitch_measurement/types.proto +"""Generated protocol buffer code.""" +from google.protobuf.internal import builder as _builder +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import symbol_database as _symbol_database +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n&stubs/niswitch_measurement/types.proto\x12\x39ni.measurementlink.measurement.tests.niswitch_measurement\"<\n\x0e\x43onfigurations\x12\x13\n\x0brelay_names\x18\x01 \x03(\t\x12\x15\n\rmulti_session\x18\x02 \x01(\x08\"k\n\x07Outputs\x12\x15\n\rsession_names\x18\x01 \x03(\t\x12\x16\n\x0eresource_names\x18\x02 \x03(\t\x12\x15\n\rchannel_lists\x18\x03 \x03(\t\x12\x1a\n\x12\x63onnected_channels\x18\x04 \x03(\tb\x06proto3') + +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals()) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'stubs.niswitch_measurement.types_pb2', globals()) +if _descriptor._USE_C_DESCRIPTORS == False: + + DESCRIPTOR._options = None + _CONFIGURATIONS._serialized_start=101 + _CONFIGURATIONS._serialized_end=161 + _OUTPUTS._serialized_start=163 + _OUTPUTS._serialized_end=270 +# @@protoc_insertion_point(module_scope) diff --git a/tests/assets/stubs/niswitch_measurement/types_pb2.pyi b/tests/assets/stubs/niswitch_measurement/types_pb2.pyi new file mode 100644 index 000000000..6bdcb40cc --- /dev/null +++ b/tests/assets/stubs/niswitch_measurement/types_pb2.pyi @@ -0,0 +1,64 @@ +""" +@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 + + RELAY_NAMES_FIELD_NUMBER: builtins.int + MULTI_SESSION_FIELD_NUMBER: builtins.int + @property + def relay_names(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: ... + multi_session: builtins.bool + def __init__( + self, + *, + relay_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", "relay_names", b"relay_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 + @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]: ... + 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 = ..., + ) -> None: ... + def ClearField(self, field_name: typing_extensions.Literal["channel_lists", b"channel_lists", "connected_channels", b"connected_channels", "resource_names", b"resource_names", "session_names", b"session_names"]) -> None: ... + +global___Outputs = Outputs diff --git a/tests/assets/stubs/niswitch_measurement/types_pb2_grpc.py b/tests/assets/stubs/niswitch_measurement/types_pb2_grpc.py new file mode 100644 index 000000000..2daafffeb --- /dev/null +++ b/tests/assets/stubs/niswitch_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 + diff --git a/tests/assets/stubs/niswitch_measurement/types_pb2_grpc.pyi b/tests/assets/stubs/niswitch_measurement/types_pb2_grpc.pyi new file mode 100644 index 000000000..b13382f6a --- /dev/null +++ b/tests/assets/stubs/niswitch_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 + ... diff --git a/tests/integration/session_management/test_nidigital_reservation.py b/tests/integration/session_management/test_nidigital_reservation.py new file mode 100644 index 000000000..fd708bad9 --- /dev/null +++ b/tests/integration/session_management/test_nidigital_reservation.py @@ -0,0 +1,94 @@ +import pathlib +from contextlib import ExitStack + +import pytest + +from ni_measurementlink_service.session_management import ( + PinMapContext, + SessionManagementClient, +) +from tests.utilities.connection_subset import ConnectionSubset, get_connection_subset +from tests.utilities.pin_map_client import PinMapClient + + +def test___single_session_reserved___initialize_nidigital_session___creates_single_session( + pin_map_id: str, + session_management_client: SessionManagementClient, +) -> None: + pin_names = ["CS"] + pin_map_context = PinMapContext(pin_map_id=pin_map_id, sites=[0]) + with ExitStack() as stack: + reservation = stack.enter_context( + session_management_client.reserve_session(pin_map_context, pin_names) + ) + + session_info = stack.enter_context(reservation.initialize_nidigital_session()) + + assert session_info.session is not None + assert session_info.session_name == "DigitalPattern1" + + +def test___multiple_sessions_reserved___initialize_nidigital_sessions___creates_multiple_sessions( + pin_map_id: str, + session_management_client: SessionManagementClient, +) -> None: + pin_names = ["CS", "SCLK"] + nidigital_resource = ["DigitalPattern1", "DigitalPattern2"] + pin_map_context = PinMapContext(pin_map_id=pin_map_id, sites=[0, 1]) + with ExitStack() as stack: + reservation = stack.enter_context( + session_management_client.reserve_sessions(pin_map_context, pin_names) + ) + + session_infos = stack.enter_context(reservation.initialize_nidigital_sessions()) + + assert all([session_info.session is not None for session_info in session_infos]) + assert [ + session_info.session_name == expected_resouce + for session_info, expected_resouce in zip(session_infos, nidigital_resource) + ] + + +def test___session_created___get_nidigital_connection___returns_connection( + pin_map_id: str, + session_management_client: SessionManagementClient, +) -> None: + pin_names = ["CS"] + pin_map_context = PinMapContext(pin_map_id=pin_map_id, sites=[0]) + with ExitStack() as stack: + reservation = stack.enter_context( + session_management_client.reserve_session(pin_map_context, pin_names) + ) + stack.enter_context(reservation.initialize_nidigital_session()) + + connection = reservation.get_nidigital_connection(pin_names[0]) + + assert get_connection_subset(connection) == ConnectionSubset( + pin_names[0], 0, "DigitalPattern1", "site0/CS" + ) + + +def test___session_created___get_nidigital_connections___returns_connections( + pin_map_id: str, + session_management_client: SessionManagementClient, +) -> None: + pin_names = ["CS", "SCLK"] + pin_map_context = PinMapContext(pin_map_id=pin_map_id, sites=[0]) + with ExitStack() as stack: + reservation = stack.enter_context( + session_management_client.reserve_session(pin_map_context, pin_names) + ) + stack.enter_context(reservation.initialize_nidigital_session()) + + connections = reservation.get_nidigital_connections(pin_names) + + assert [get_connection_subset(connection) for connection in connections] == [ + ConnectionSubset(pin_names[0], 0, "DigitalPattern1", "site0/CS"), + ConnectionSubset(pin_names[1], 0, "DigitalPattern1", "site0/SCLK"), + ] + + +@pytest.fixture +def pin_map_id(pin_map_client: PinMapClient, pin_map_directory: pathlib.Path) -> str: + pin_map_name = "2Digital2Group4Pin1Site.pinmap" + return pin_map_client.update_pin_map(pin_map_directory / pin_map_name) diff --git a/tests/integration/session_management/test_niswitch_reservation.py b/tests/integration/session_management/test_niswitch_reservation.py new file mode 100644 index 000000000..922d1a848 --- /dev/null +++ b/tests/integration/session_management/test_niswitch_reservation.py @@ -0,0 +1,94 @@ +import pathlib +from contextlib import ExitStack + +import pytest + +from ni_measurementlink_service.session_management import ( + PinMapContext, + SessionManagementClient, +) +from tests.utilities.connection_subset import ConnectionSubset, get_connection_subset +from tests.utilities.pin_map_client import PinMapClient + +_SITE = 0 + + +def test___single_session_reserved___initialize_niswitch_session___creates_single_session( + pin_map_context: PinMapContext, + session_management_client: SessionManagementClient, +) -> None: + relay_names = ["SiteRelay1"] + with ExitStack() as stack: + reservation = stack.enter_context( + session_management_client.reserve_session(pin_map_context, relay_names) + ) + + session_info = stack.enter_context(reservation.initialize_niswitch_session()) + + assert session_info.session is not None + assert session_info.session_name == "RelayDriver1" + + +def test___multiple_sessions_reserved___initialize_niswitch_sessions___creates_multiple_sessions( + pin_map_context: PinMapContext, + session_management_client: SessionManagementClient, +) -> None: + relay_names = ["SiteRelay1", "SiteRelay2"] + niswitch_resource = ["RelayDriver1", "RelayDriver2"] + with ExitStack() as stack: + reservation = stack.enter_context( + session_management_client.reserve_sessions(pin_map_context, relay_names) + ) + + session_infos = stack.enter_context(reservation.initialize_niswitch_sessions()) + + assert all([session_info.session is not None for session_info in session_infos]) + assert [ + session_info.session_name == expected_resource + for session_info, expected_resource in zip(session_infos, niswitch_resource) + ] + + +def test___session_created___get_niswitch_connection___returns_connection( + pin_map_context: PinMapContext, + session_management_client: SessionManagementClient, +) -> None: + relay_names = ["SiteRelay1"] + with ExitStack() as stack: + reservation = stack.enter_context( + session_management_client.reserve_session(pin_map_context, relay_names) + ) + stack.enter_context(reservation.initialize_niswitch_session()) + + connection = reservation.get_niswitch_connection(relay_names[0]) + + assert get_connection_subset(connection) == ConnectionSubset( + relay_names[0], _SITE, "RelayDriver1", "K0" + ) + + +def test___sessions_created___get_niswitch_connections___returns_connections( + pin_map_context: PinMapContext, + session_management_client: SessionManagementClient, +) -> None: + relay_names = ["SiteRelay1", "SiteRelay2"] + with ExitStack() as stack: + reservation = stack.enter_context( + session_management_client.reserve_sessions(pin_map_context, relay_names) + ) + stack.enter_context(reservation.initialize_niswitch_sessions()) + + connections = reservation.get_niswitch_connections(relay_names) + + assert [get_connection_subset(connection) for connection in connections] == [ + ConnectionSubset(relay_names[0], _SITE, "RelayDriver1", "K0"), + ConnectionSubset(relay_names[1], _SITE, "RelayDriver2", "K1"), + ] + + +@pytest.fixture +def pin_map_context(pin_map_client: PinMapClient, pin_map_directory: pathlib.Path) -> PinMapContext: + pin_map_name = "2Switch2Relay1Site.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]) diff --git a/tests/utilities/nidigital_measurement/NIDigitalMeasurement.serviceconfig b/tests/utilities/nidigital_measurement/NIDigitalMeasurement.serviceconfig new file mode 100644 index 000000000..80b796820 --- /dev/null +++ b/tests/utilities/nidigital_measurement/NIDigitalMeasurement.serviceconfig @@ -0,0 +1,19 @@ +{ + "services": [ + { + "displayName": "NI-Digital Measurement (Py)", + "serviceClass": "ni.tests.NIDigitalMeasurement_Python", + "descriptionUrl": "", + "providedInterfaces": [ + "ni.measurementlink.measurement.v1.MeasurementService", + "ni.measurementlink.measurement.v2.MeasurementService" + ], + "path": "start.bat", + "annotations": { + "ni/service.description": "NI-Digital MeasurementLink test service.", + "ni/service.collection": "NI.Tests", + "ni/service.tags": [] + } + } + ] +} \ No newline at end of file diff --git a/tests/utilities/nidigital_measurement/Pattern.digipat b/tests/utilities/nidigital_measurement/Pattern.digipat new file mode 100644 index 000000000..a3566238d Binary files /dev/null and b/tests/utilities/nidigital_measurement/Pattern.digipat differ diff --git a/tests/utilities/nidigital_measurement/PinLevels.digilevels b/tests/utilities/nidigital_measurement/PinLevels.digilevels new file mode 100644 index 000000000..ba3551968 --- /dev/null +++ b/tests/utilities/nidigital_measurement/PinLevels.digilevels @@ -0,0 +1,51 @@ + + + + + + dc.vcc * 0.7 + 0 + 2.5 + 0.5 + 1.5 m + -1.5 m + 0 + 0 + HighZ + + + dc.vcc * 0.7 + 0 + 2.5 + 0.5 + 1.5 m + -1.5 m + 0 + 0 + HighZ + + + dc.vcc * 0.7 + 0 + 2.5 + 0.5 + 1.5 m + -1.5 m + 0 + 0 + HighZ + + + dc.vcc * 0.7 + 0 + 2.5 + 0.5 + 1.5 m + -1.5 m + 0 + 0 + HighZ + + + + \ No newline at end of file diff --git a/tests/utilities/nidigital_measurement/Specifications.specs b/tests/utilities/nidigital_measurement/Specifications.specs new file mode 100644 index 000000000..8792bc803 --- /dev/null +++ b/tests/utilities/nidigital_measurement/Specifications.specs @@ -0,0 +1,16 @@ + + +
+ + 5 + V + +
+
+ + 1 / 1000000 + 1 MHz + +
+ + \ No newline at end of file diff --git a/tests/utilities/nidigital_measurement/Timing.digitiming b/tests/utilities/nidigital_measurement/Timing.digitiming new file mode 100644 index 000000000..df69bee5d --- /dev/null +++ b/tests/utilities/nidigital_measurement/Timing.digitiming @@ -0,0 +1,57 @@ + + + + + + ac.period + + + + 0 + 0 + ac.period + + + (3 * ac.period) / 4 + + Pattern + + + + ac.period / 2 + ac.period / 2 + ac.period + ac.period + + + (3 * ac.period) / 4 + + Pattern + + + + 0 + 0 + ac.period + + + (3 * ac.period) / 4 + + Pattern + + + + 0 + 0 + ac.period + + + (3 * ac.period) / 4 + + Pattern + + + + + + \ No newline at end of file diff --git a/tests/utilities/nidigital_measurement/__init__.py b/tests/utilities/nidigital_measurement/__init__.py new file mode 100644 index 000000000..0cb3c709b --- /dev/null +++ b/tests/utilities/nidigital_measurement/__init__.py @@ -0,0 +1,136 @@ +"""NI-Digital MeasurementLink test service.""" +import pathlib +from itertools import groupby +from typing import Iterable, Sequence, Tuple, Union + +import nidigital + +import ni_measurementlink_service as nims +from ni_measurementlink_service.session_management import TypedConnection, TypedSessionInformation + +service_directory = pathlib.Path(__file__).resolve().parent +measurement_service = nims.MeasurementService( + service_config_path=service_directory / "NIDigitalMeasurement.serviceconfig", + version="0.1.0.0", + ui_file_paths=[ + service_directory, + ], +) + + +@measurement_service.register_measurement +@measurement_service.configuration("pin_names", nims.DataType.StringArray1D, ["CS"]) +@measurement_service.configuration("multi_session", nims.DataType.Boolean, False) +@measurement_service.output("session_names", nims.DataType.StringArray1D) +@measurement_service.output("resource_names", nims.DataType.StringArray1D) +@measurement_service.output("channel_lists", nims.DataType.StringArray1D) +@measurement_service.output("connected_channels", nims.DataType.StringArray1D) +@measurement_service.output("passing_sites", nims.DataType.Int32Array1D) +@measurement_service.output("failing_sites", nims.DataType.Int32Array1D) +def measure( + pin_names: Iterable[str], + multi_session: bool, +) -> Tuple[ + Iterable[str], Iterable[str], Iterable[str], Iterable[str], Iterable[str], Iterable[str] +]: + """NI-Digital MeasurementLink test service.""" + if multi_session: + with measurement_service.context.reserve_sessions(pin_names) as reservation: + with reservation.initialize_nidigital_sessions() as session_infos: + connections = reservation.get_nidigital_connections(pin_names) + assert all([session_info.session is not None for session_info in session_infos]) + passing_sites, failing_sites = _burst_spi_pattern(session_infos) + connections_by_session = [ + list(g) for _, g in groupby(sorted(connections, key=_key_func), key=_key_func) + ] + + return ( + [session_info.session_name for session_info in session_infos], + [session_info.resource_name for session_info in session_infos], + [session_info.channel_list for session_info in session_infos], + [ + ", ".join(conn.channel_name for conn in conns) + for conns in connections_by_session + ], + list(passing_sites), + list(failing_sites), + ) + else: + with measurement_service.context.reserve_session(pin_names) as reservation: + with reservation.initialize_nidigital_session() as session_info: + connection = reservation.get_nidigital_connection(list(pin_names)[0]) + assert session_info.session is not None + passing_sites, failing_sites = _burst_spi_pattern([session_info]) + + return ( + [session_info.session_name], + [session_info.resource_name], + [session_info.channel_list], + [connection.channel_name], + passing_sites, + failing_sites, + ) + + +def _burst_spi_pattern( + session_infos: Sequence[TypedSessionInformation[nidigital.Session]], +) -> Tuple: + specifications_file_path = "Specifications.specs" + levels_file_path = "PinLevels.digilevels" + timing_file_path = "Timing.digitiming" + pattern_file_path = "Pattern.digipat" + pin_map_context = measurement_service.context.pin_map_context + selected_sites_string = ",".join(f"site{i}" for i in pin_map_context.sites or []) + + passing_sites_list, failing_sites_list = [], [] + for session_info in session_infos: + session = session_info.session + selected_sites = session.sites[selected_sites_string] + + if not session_info.session_exists: + session.load_pin_map(pin_map_context.pin_map_id) + session.load_specifications_levels_and_timing( + str(_resolve_relative_path(service_directory, specifications_file_path)), + str(_resolve_relative_path(service_directory, levels_file_path)), + str(_resolve_relative_path(service_directory, timing_file_path)), + ) + session.load_pattern( + str(_resolve_relative_path(service_directory, pattern_file_path)), + ) + + levels_file_name = pathlib.Path(levels_file_path).stem + timing_file_name = pathlib.Path(timing_file_path).stem + + for session_info in session_infos: + selected_sites = session_info.session.sites[selected_sites_string] + selected_sites.apply_levels_and_timing(levels_file_name, timing_file_name) + + for session_info in session_infos: + selected_sites = session_info.session.sites[selected_sites_string] + selected_sites.burst_pattern(start_label="SPI_Pattern", wait_until_done=False) + + for session_info in session_infos: + selected_sites = session_info.session.sites[selected_sites_string] + session_info.session.wait_until_done() + site_pass_fail = selected_sites.get_site_pass_fail() + passing_sites = [site for site, pass_fail in site_pass_fail.items() if pass_fail] + failing_sites = [site for site, pass_fail in site_pass_fail.items() if not pass_fail] + passing_sites_list.extend(passing_sites) + failing_sites_list.extend(failing_sites) + session.selected_function = nidigital.SelectedFunction.DISCONNECT + + return (passing_sites_list, failing_sites_list) + + +def _resolve_relative_path( + directory_path: pathlib.Path, file_path: Union[str, pathlib.Path] +) -> pathlib.Path: + file_path = pathlib.Path(file_path) + if file_path.is_absolute(): + return file_path + else: + return (directory_path / file_path).resolve() + + +def _key_func(conn: TypedConnection[nidigital.Session]) -> str: + return conn.session_info.session_name diff --git a/tests/utilities/niswitch_measurement/NISwitchMeasurement.serviceconfig b/tests/utilities/niswitch_measurement/NISwitchMeasurement.serviceconfig new file mode 100644 index 000000000..3e3cd5afd --- /dev/null +++ b/tests/utilities/niswitch_measurement/NISwitchMeasurement.serviceconfig @@ -0,0 +1,19 @@ +{ + "services": [ + { + "displayName": "NI-SWITCH Measurement (Py)", + "serviceClass": "ni.tests.NISwitchMeasurement_Python", + "descriptionUrl": "", + "providedInterfaces": [ + "ni.measurementlink.measurement.v1.MeasurementService", + "ni.measurementlink.measurement.v2.MeasurementService" + ], + "path": "start.bat", + "annotations": { + "ni/service.description": "NI-SWITCH MeasurementLink test service.", + "ni/service.collection": "NI.Tests", + "ni/service.tags": [] + } + } + ] +} \ No newline at end of file diff --git a/tests/utilities/niswitch_measurement/__init__.py b/tests/utilities/niswitch_measurement/__init__.py new file mode 100644 index 000000000..f31e8d68e --- /dev/null +++ b/tests/utilities/niswitch_measurement/__init__.py @@ -0,0 +1,69 @@ +"""NI-SWITCH MeasurementLink test service.""" +import pathlib +from typing import Iterable, Sequence, Tuple + +import niswitch + +import ni_measurementlink_service as nims +from ni_measurementlink_service.session_management import TypedSessionInformation + +service_directory = pathlib.Path(__file__).resolve().parent +measurement_service = nims.MeasurementService( + service_config_path=service_directory / "NISwitchMeasurement.serviceconfig", + version="0.1.0.0", + ui_file_paths=[ + service_directory, + ], +) + + +@measurement_service.register_measurement +@measurement_service.configuration("relay_names", nims.DataType.StringArray1D, ["SiteRelay1"]) +@measurement_service.configuration("multi_session", nims.DataType.Boolean, False) +@measurement_service.output("session_names", nims.DataType.StringArray1D) +@measurement_service.output("resource_names", nims.DataType.StringArray1D) +@measurement_service.output("channel_lists", nims.DataType.StringArray1D) +@measurement_service.output("connected_channels", nims.DataType.StringArray1D) +def measure( + relay_names: Iterable[str], + multi_session: bool, +) -> Tuple[Iterable[str], Iterable[str], Iterable[str], Iterable[str]]: + """NI-SWITCH MeasurementLink test service.""" + if multi_session: + with measurement_service.context.reserve_sessions(relay_names) as reservation: + with reservation.initialize_niswitch_sessions() as session_infos: + connections = reservation.get_niswitch_connections(relay_names) + assert all([session_info.session is not None for session_info in session_infos]) + _control_relays(session_infos) + + return ( + [session_info.session_name for session_info in session_infos], + [session_info.resource_name for session_info in session_infos], + [session_info.channel_list for session_info in session_infos], + [connection.channel_name for connection in connections], + ) + else: + with measurement_service.context.reserve_session(relay_names) as reservation: + with reservation.initialize_niswitch_session() as session_info: + connection = reservation.get_niswitch_connection(list(relay_names)[0]) + assert session_info.session is not None + _control_relays([session_info]) + + return ( + [session_info.session_name], + [session_info.resource_name], + [session_info.channel_list], + [connection.channel_name], + ) + + +def _control_relays( + session_infos: Sequence[TypedSessionInformation[niswitch.Session]], +) -> None: + for session_info in session_infos: + session_info.session.relay_control( + session_info.channel_list, niswitch.enums.RelayAction.CLOSE + ) + + for session_info in session_infos: + session_info.session.wait_for_debounce()