diff --git a/.github/workflows/Publish_NIMS.yml b/.github/workflows/Publish_NIMS.yml index 01b67e7c6..f1b27cd83 100644 --- a/.github/workflows/Publish_NIMS.yml +++ b/.github/workflows/Publish_NIMS.yml @@ -12,6 +12,14 @@ env: PYTHON_VERSION: 3.9 jobs: + check_nims: + name: Check NIMS + uses: ./.github/workflows/check_nims.yml + check_nimg: + name: Check NIMG + uses: ./.github/workflows/check_nimg.yml + # Don't run check_examples.yml because the examples may reference a + # package version that doesn't exist yet. build: name: Update API reference docs and Publish NIMS Package to PyPI runs-on : ubuntu-latest @@ -32,16 +40,6 @@ jobs: path: ~/.cache/pypoetry/virtualenvs key: ${{ runner.os }}-poetry-${{ hashFiles('poetry.lock') }} - - name: Setup NIMS - run: poetry install - - - name: Setup NIMG - run: poetry install - working-directory: ./ni_measurementlink_generator - - - name: Lint NIMS, NIMG, and examples - run: poetry run ni-python-styleguide lint - # If the tag is 0.1.0, this will set the version of NIMS package to 0.1.0 - name: Store version from Tag id: vars @@ -56,16 +54,16 @@ jobs: poetry version ${{ steps.vars.outputs.tag }} working-directory: ./ni_measurementlink_generator - - name: Setup NIMS with docs - run: poetry install --with docs + - name: Install ni-measurementlink-service (all extras, docs) + run: poetry install -v --all-extras --with docs - - name: Update API Reference Documents(Generated using Sphinx) + - name: Build docs run: | rm -rf docs mkdir -p docs poetry run sphinx-build _docs_source docs -b html - - name: Commit file changes. + - name: Commit file changes if: ${{ startsWith(github.event.release.target_commitish, 'main') || startsWith(github.event.release.target_commitish, 'releases/') }} run: | git config --local user.email "action@github.com" diff --git a/.github/workflows/check_nims.yml b/.github/workflows/check_nims.yml index 87b0340ba..b0601929c 100644 --- a/.github/workflows/check_nims.yml +++ b/.github/workflows/check_nims.yml @@ -30,8 +30,8 @@ jobs: with: path: ~/.cache/pypoetry/virtualenvs key: ${{ runner.os }}-poetry-${{ hashFiles('poetry.lock') }} - - name: Install ni-measurementlink-service - run: poetry install -v + - name: Install ni-measurementlink-service (all extras) + run: poetry install -v --all-extras - name: Lint ni-measurementlink-service run: poetry run ni-python-styleguide lint - name: Mypy static analysis (ni-measurementlink-service, Linux) @@ -42,8 +42,8 @@ jobs: run: poetry run mypy tests - name: Mypy static analysis (tests, Windows) run: poetry run mypy tests --platform win32 - - name: Install ni-measurementlink-service with docs - run: poetry install --with docs + - name: Install ni-measurementlink-service (all extras, docs) + run: poetry install -v --all-extras --with docs - name: Build docs and check for errors/warnings run: | rm -rf docs diff --git a/.github/workflows/run_tests.yml b/.github/workflows/run_tests.yml index 0e1c6490b..bfbd8ae3c 100644 --- a/.github/workflows/run_tests.yml +++ b/.github/workflows/run_tests.yml @@ -34,10 +34,14 @@ jobs: with: path: ~/.cache/pypoetry/virtualenvs key: ${{ runner.os }}-poetry-${{ hashFiles('poetry.lock') }} - - name: Install ni-measurementlink-service + - name: Install ni-measurementlink-service (no extras) run: poetry install -v - - name: Run tests and code coverage (ni-measurementlink-service) - run: poetry run pytest ./tests -v --cov=ni_measurementlink_service --junitxml=test_results/nims-${{ matrix.os }}-py${{ matrix.python-version}}.xml + - name: Run tests and code coverage (ni-measurementlink-service, no extras) + run: poetry run pytest ./tests -v --cov=ni_measurementlink_service --junitxml=test_results/nims-${{ matrix.os }}-py${{ matrix.python-version}}-no-extras.xml + - name: Install ni-measurementlink-service (all extras) + run: poetry install -v --all-extras + - name: Run tests and code coverage (ni-measurementlink-service, all extras) + run: poetry run pytest ./tests -v --cov=ni_measurementlink_service --junitxml=test_results/nims-${{ matrix.os }}-py${{ matrix.python-version}}-all-extras.xml - name: Install ni-measurementlink-generator run: poetry install -v working-directory: ./ni_measurementlink_generator diff --git a/ni_measurementlink_service/_drivers/_nidcpower.py b/ni_measurementlink_service/_drivers/_nidcpower.py new file mode 100644 index 000000000..d7244501c --- /dev/null +++ b/ni_measurementlink_service/_drivers/_nidcpower.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +from typing import Any, Dict, Optional + +import nidcpower + +from ni_measurementlink_service import session_management # circular import +from ni_measurementlink_service._channelpool import GrpcChannelPool +from ni_measurementlink_service._configuration import NIDCPOWER_OPTIONS +from ni_measurementlink_service._drivers._grpcdevice import ( + get_insecure_grpc_device_channel, +) +from ni_measurementlink_service._internal.discovery_client import DiscoveryClient +from ni_measurementlink_service._sessiontypes import SessionInitializationBehavior + +_INITIALIZATION_BEHAVIOR = { + SessionInitializationBehavior.AUTO: nidcpower.SessionInitializationBehavior.AUTO, + SessionInitializationBehavior.INITIALIZE_SERVER_SESSION: nidcpower.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION, + SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION: nidcpower.SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION, +} + + +class SessionConstructor: + """Constructs sessions based on SessionInformation.""" + + def __init__( + self, + discovery_client: DiscoveryClient, + grpc_channel_pool: GrpcChannelPool, + reset: bool, + options: Optional[Dict[str, Any]], + initialization_behavior: SessionInitializationBehavior, + ) -> None: + """Initialize the session constructor.""" + self._grpc_channel = get_insecure_grpc_device_channel( + discovery_client, grpc_channel_pool, nidcpower.GRPC_SERVICE_INTERFACE_NAME + ) + self._reset = reset + self._options = NIDCPOWER_OPTIONS.to_dict() if options is None else options + self._initialization_behavior = _INITIALIZATION_BEHAVIOR[initialization_behavior] + + def __call__(self, session_info: session_management.SessionInformation) -> nidcpower.Session: + """Construct a session object.""" + kwargs: Dict[str, Any] = {} + if self._grpc_channel: + kwargs["grpc_options"] = nidcpower.GrpcSessionOptions( + grpc_channel=self._grpc_channel, + session_name=session_info.session_name, + initialization_behavior=self._initialization_behavior, + ) + + return nidcpower.Session( + resource_name=session_info.resource_name, + reset=self._reset, + options=self._options, + **kwargs, + ) diff --git a/ni_measurementlink_service/_drivers/_nidigital.py b/ni_measurementlink_service/_drivers/_nidigital.py new file mode 100644 index 000000000..1d3540cbf --- /dev/null +++ b/ni_measurementlink_service/_drivers/_nidigital.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +from typing import Any, Dict, Optional + +import nidigital + +from ni_measurementlink_service import session_management # circular import +from ni_measurementlink_service._channelpool import GrpcChannelPool +from ni_measurementlink_service._configuration import NIDIGITAL_OPTIONS +from ni_measurementlink_service._drivers._grpcdevice import ( + get_insecure_grpc_device_channel, +) +from ni_measurementlink_service._internal.discovery_client import DiscoveryClient +from ni_measurementlink_service._sessiontypes import SessionInitializationBehavior + +_INITIALIZATION_BEHAVIOR = { + SessionInitializationBehavior.AUTO: nidigital.SessionInitializationBehavior.AUTO, + SessionInitializationBehavior.INITIALIZE_SERVER_SESSION: nidigital.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION, + SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION: nidigital.SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION, +} + + +class SessionConstructor: + """Constructs sessions based on SessionInformation.""" + + def __init__( + self, + discovery_client: DiscoveryClient, + grpc_channel_pool: GrpcChannelPool, + reset_device: bool, + options: Optional[Dict[str, Any]], + initialization_behavior: SessionInitializationBehavior, + ) -> None: + """Initialize the session constructor.""" + self._grpc_channel = get_insecure_grpc_device_channel( + discovery_client, grpc_channel_pool, nidigital.GRPC_SERVICE_INTERFACE_NAME + ) + self._reset_device = reset_device + self._options = NIDIGITAL_OPTIONS.to_dict() if options is None else options + self._initialization_behavior = _INITIALIZATION_BEHAVIOR[initialization_behavior] + + def __call__(self, session_info: session_management.SessionInformation) -> nidigital.Session: + """Construct a session object.""" + kwargs: Dict[str, Any] = {} + if self._grpc_channel: + kwargs["grpc_options"] = nidigital.GrpcSessionOptions( + grpc_channel=self._grpc_channel, + session_name=session_info.session_name, + initialization_behavior=self._initialization_behavior, + ) + + # Omit id_query because it has no effect. + return nidigital.Session( + resource_name=session_info.resource_name, + reset_device=self._reset_device, + options=self._options, + **kwargs, + ) diff --git a/ni_measurementlink_service/_drivers/_nidmm.py b/ni_measurementlink_service/_drivers/_nidmm.py new file mode 100644 index 000000000..50e4ed398 --- /dev/null +++ b/ni_measurementlink_service/_drivers/_nidmm.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +from typing import Any, Dict, Optional + +import nidmm + +from ni_measurementlink_service import session_management # circular import +from ni_measurementlink_service._channelpool import GrpcChannelPool +from ni_measurementlink_service._configuration import NIDMM_OPTIONS +from ni_measurementlink_service._drivers._grpcdevice import ( + get_insecure_grpc_device_channel, +) +from ni_measurementlink_service._internal.discovery_client import DiscoveryClient +from ni_measurementlink_service._sessiontypes import SessionInitializationBehavior + +_INITIALIZATION_BEHAVIOR = { + SessionInitializationBehavior.AUTO: nidmm.SessionInitializationBehavior.AUTO, + SessionInitializationBehavior.INITIALIZE_SERVER_SESSION: nidmm.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION, + SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION: nidmm.SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION, +} + + +class SessionConstructor: + """Constructs sessions based on SessionInformation.""" + + def __init__( + self, + discovery_client: DiscoveryClient, + grpc_channel_pool: GrpcChannelPool, + reset_device: bool, + options: Optional[Dict[str, Any]], + initialization_behavior: SessionInitializationBehavior, + ) -> None: + """Initialize the session constructor.""" + self._grpc_channel = get_insecure_grpc_device_channel( + discovery_client, grpc_channel_pool, nidmm.GRPC_SERVICE_INTERFACE_NAME + ) + self._reset_device = reset_device + self._options = NIDMM_OPTIONS.to_dict() if options is None else options + self._initialization_behavior = _INITIALIZATION_BEHAVIOR[initialization_behavior] + + def __call__(self, session_info: session_management.SessionInformation) -> nidmm.Session: + """Construct a session object.""" + kwargs: Dict[str, Any] = {} + if self._grpc_channel: + kwargs["grpc_options"] = nidmm.GrpcSessionOptions( + grpc_channel=self._grpc_channel, + session_name=session_info.session_name, + initialization_behavior=self._initialization_behavior, + ) + + # Omit id_query because it has no effect. + return nidmm.Session( + resource_name=session_info.resource_name, + reset_device=self._reset_device, + options=self._options, + **kwargs, + ) diff --git a/ni_measurementlink_service/_drivers/_nifgen.py b/ni_measurementlink_service/_drivers/_nifgen.py new file mode 100644 index 000000000..7ee77ff99 --- /dev/null +++ b/ni_measurementlink_service/_drivers/_nifgen.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +from typing import Any, Dict, Optional + +import nifgen + +from ni_measurementlink_service import session_management # circular import +from ni_measurementlink_service._channelpool import GrpcChannelPool +from ni_measurementlink_service._configuration import NIFGEN_OPTIONS +from ni_measurementlink_service._drivers._grpcdevice import ( + get_insecure_grpc_device_channel, +) +from ni_measurementlink_service._internal.discovery_client import DiscoveryClient +from ni_measurementlink_service._sessiontypes import SessionInitializationBehavior + +_INITIALIZATION_BEHAVIOR = { + SessionInitializationBehavior.AUTO: nifgen.SessionInitializationBehavior.AUTO, + SessionInitializationBehavior.INITIALIZE_SERVER_SESSION: nifgen.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION, + SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION: nifgen.SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION, +} + + +class SessionConstructor: + """Constructs sessions based on SessionInformation.""" + + def __init__( + self, + discovery_client: DiscoveryClient, + grpc_channel_pool: GrpcChannelPool, + reset_device: bool, + options: Optional[Dict[str, Any]], + initialization_behavior: SessionInitializationBehavior, + ) -> None: + """Initialize the session constructor.""" + self._grpc_channel = get_insecure_grpc_device_channel( + discovery_client, grpc_channel_pool, nifgen.GRPC_SERVICE_INTERFACE_NAME + ) + self._reset_device = reset_device + self._options = NIFGEN_OPTIONS.to_dict() if options is None else options + self._initialization_behavior = _INITIALIZATION_BEHAVIOR[initialization_behavior] + + def __call__(self, session_info: session_management.SessionInformation) -> nifgen.Session: + """Construct a session object.""" + kwargs: Dict[str, Any] = {} + if self._grpc_channel: + kwargs["grpc_options"] = nifgen.GrpcSessionOptions( + grpc_channel=self._grpc_channel, + session_name=session_info.session_name, + initialization_behavior=self._initialization_behavior, + ) + + return nifgen.Session( + resource_name=session_info.resource_name, + reset_device=self._reset_device, + options=self._options, + **kwargs, + ) diff --git a/ni_measurementlink_service/_drivers/_niscope.py b/ni_measurementlink_service/_drivers/_niscope.py new file mode 100644 index 000000000..e2752895d --- /dev/null +++ b/ni_measurementlink_service/_drivers/_niscope.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +from typing import Any, Dict, Optional + +import niscope + +from ni_measurementlink_service import session_management # circular import +from ni_measurementlink_service._channelpool import GrpcChannelPool +from ni_measurementlink_service._configuration import NISCOPE_OPTIONS +from ni_measurementlink_service._drivers._grpcdevice import ( + get_insecure_grpc_device_channel, +) +from ni_measurementlink_service._internal.discovery_client import DiscoveryClient +from ni_measurementlink_service._sessiontypes import SessionInitializationBehavior + +_INITIALIZATION_BEHAVIOR = { + SessionInitializationBehavior.AUTO: niscope.SessionInitializationBehavior.AUTO, + SessionInitializationBehavior.INITIALIZE_SERVER_SESSION: niscope.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION, + SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION: niscope.SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION, +} + + +class SessionConstructor: + """Constructs sessions based on SessionInformation.""" + + def __init__( + self, + discovery_client: DiscoveryClient, + grpc_channel_pool: GrpcChannelPool, + reset_device: bool, + options: Optional[Dict[str, Any]], + initialization_behavior: SessionInitializationBehavior, + ) -> None: + """Initialize the session constructor.""" + self._grpc_channel = get_insecure_grpc_device_channel( + discovery_client, grpc_channel_pool, niscope.GRPC_SERVICE_INTERFACE_NAME + ) + self._reset_device = reset_device + self._options = NISCOPE_OPTIONS.to_dict() if options is None else options + self._initialization_behavior = _INITIALIZATION_BEHAVIOR[initialization_behavior] + + def __call__(self, session_info: session_management.SessionInformation) -> niscope.Session: + """Construct a session object.""" + kwargs: Dict[str, Any] = {} + if self._grpc_channel: + kwargs["grpc_options"] = niscope.GrpcSessionOptions( + grpc_channel=self._grpc_channel, + session_name=session_info.session_name, + initialization_behavior=self._initialization_behavior, + ) + + # Omit id_query because it has no effect. + return niscope.Session( + resource_name=session_info.resource_name, + reset_device=self._reset_device, + options=self._options, + **kwargs, + ) diff --git a/ni_measurementlink_service/_drivers/_niswitch.py b/ni_measurementlink_service/_drivers/_niswitch.py new file mode 100644 index 000000000..74eedabca --- /dev/null +++ b/ni_measurementlink_service/_drivers/_niswitch.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from typing import Any, Dict, Optional + +import niswitch + +from ni_measurementlink_service import session_management # circular import +from ni_measurementlink_service._channelpool import GrpcChannelPool +from ni_measurementlink_service._configuration import NISWITCH_OPTIONS +from ni_measurementlink_service._drivers._grpcdevice import ( + get_insecure_grpc_device_channel, +) +from ni_measurementlink_service._internal.discovery_client import DiscoveryClient +from ni_measurementlink_service._sessiontypes import SessionInitializationBehavior + +_INITIALIZATION_BEHAVIOR = { + SessionInitializationBehavior.AUTO: niswitch.SessionInitializationBehavior.AUTO, + SessionInitializationBehavior.INITIALIZE_SERVER_SESSION: niswitch.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION, + SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION: niswitch.SessionInitializationBehavior.ATTACH_TO_SERVER_SESSION, +} + + +class SessionConstructor: + """Constructs sessions based on SessionInformation.""" + + def __init__( + self, + discovery_client: DiscoveryClient, + grpc_channel_pool: GrpcChannelPool, + topology: Optional[str], + simulate: Optional[bool], + reset_device: bool, + initialization_behavior: SessionInitializationBehavior, + ) -> None: + """Initialize the session constructor.""" + self._grpc_channel = get_insecure_grpc_device_channel( + discovery_client, grpc_channel_pool, niswitch.GRPC_SERVICE_INTERFACE_NAME + ) + self._topology = NISWITCH_OPTIONS.topology if topology is None else topology + self._simulate = NISWITCH_OPTIONS.simulate if simulate is None else simulate + self._reset_device = reset_device + self._initialization_behavior = _INITIALIZATION_BEHAVIOR[initialization_behavior] + + def __call__(self, session_info: session_management.SessionInformation) -> niswitch.Session: + """Construct a session object.""" + kwargs: Dict[str, Any] = {} + if self._grpc_channel: + kwargs["grpc_options"] = niswitch.GrpcSessionOptions( + grpc_channel=self._grpc_channel, + session_name=session_info.session_name, + initialization_behavior=self._initialization_behavior, + ) + + return niswitch.Session( + resource_name=session_info.resource_name, + topology=self._topology, + simulate=self._simulate, + reset_device=self._reset_device, + **kwargs, + ) diff --git a/ni_measurementlink_service/_sessiontypes.py b/ni_measurementlink_service/_sessiontypes.py new file mode 100644 index 000000000..6e4694edb --- /dev/null +++ b/ni_measurementlink_service/_sessiontypes.py @@ -0,0 +1,43 @@ +"""Session management data types. + +This submodule is intended to prevent circular imports between +session_management.py and driver-specific code such as +_drivers/_nidcpower.py. +""" +from enum import IntEnum + + +class SessionInitializationBehavior(IntEnum): + """Specifies whether to initialize a new session or attach to an existing session.""" + + AUTO = 0 + """ + The NI gRPC Device Server will attach to an existing session with the + specified name if it exists, otherwise the server will initialize a new + session. + + Note: When using the Session as a context manager and the context exits, the + behavior depends on what happened when the constructor was called. If it + resulted in a new session being initialized on the NI gRPC Device Server, + then it will automatically close the server session. If it instead attached + to an existing session, then it will detach from the server session and + leave it open. + """ + + INITIALIZE_SERVER_SESSION = 1 + """ + Require the NI gRPC Device Server to initialize a new session with the + specified name. + + Note: When using the Session as a context manager and the context exits, it + will automatically close the server session. + """ + + ATTACH_TO_SERVER_SESSION = 2 + """ + Require the NI gRPC Device Server to attach to an existing session with the + specified name. + + Note: When using the Session as a context manager and the context exits, it + will detach from the server session and leave it open. + """ diff --git a/ni_measurementlink_service/session_management.py b/ni_measurementlink_service/session_management.py index 90f5260e9..75b22c43b 100644 --- a/ni_measurementlink_service/session_management.py +++ b/ni_measurementlink_service/session_management.py @@ -49,8 +49,20 @@ session_management_service_pb2, session_management_service_pb2_grpc, ) +from ni_measurementlink_service._sessiontypes import ( + SessionInitializationBehavior as SessionInitializationBehavior, # re-export +) if TYPE_CHECKING: + # Driver API packages are optional dependencies, so only import them lazily + # at run time or when type-checking. + import nidcpower + import nidigital + import nidmm + import nifgen + import niscope + import niswitch + if sys.version_info >= (3, 11): from typing import Self else: @@ -284,6 +296,18 @@ def __init__( self._session_info = session_info self._session_cache: Dict[str, object] = {} + @property + def _discovery_client(self) -> DiscoveryClient: + if not self._session_manager._discovery_client: + raise ValueError("This method requires a discovery client.") + return self._session_manager._discovery_client + + @property + def _grpc_channel_pool(self) -> GrpcChannelPool: + if not self._session_manager._grpc_channel_pool: + raise ValueError("This method requires a gRPC channel pool.") + return self._session_manager._grpc_channel_pool + def __enter__(self: Self) -> Self: """Context management protocol. Returns self.""" return self @@ -332,12 +356,12 @@ def _create_session_core( session_infos = self._get_matching_session_infos(instrument_type_id) if len(session_infos) == 0: raise ValueError( - f"No sessions matched instrument type ID '{instrument_type_id}'. " + f"No reserved sessions matched instrument type ID '{instrument_type_id}'. " "Expected single session, got 0 sessions." ) elif len(session_infos) > 1: raise ValueError( - f"Too many sessions matched instrument type ID '{instrument_type_id}'. " + f"Too many reserved sessions matched instrument type ID '{instrument_type_id}'. " f"Expected single session, got {len(session_infos)} sessions." ) @@ -357,7 +381,7 @@ def _create_sessions_core( session_infos = self._get_matching_session_infos(instrument_type_id) if len(session_infos) == 0: raise ValueError( - f"No sessions matched instrument type ID '{instrument_type_id}'. " + f"No reserved sessions matched instrument type ID '{instrument_type_id}'. " "Expected single or multiple sessions, got 0 sessions." ) @@ -380,16 +404,22 @@ def create_session( This is a generic method that supports any instrument driver. Args: - session_constructor: A function that constructs sessions based on session - information. + session_constructor: A function that constructs sessions based on + session information. + instrument_type_id: Instrument type ID for the session. - For NI instruments, use instrument type id constants, such as - :py:const:`INSTRUMENT_TYPE_NI_DCPOWER` or :py:const:`INSTRUMENT_TYPE_NI_DMM`. - For custom instruments, use the instrument type id defined in the pin map file. + + For custom instruments, use the instrument type id defined in + the pin map file. Returns: A context manager that yields a session information object. The created session is available via the ``session`` field. + + Raises: + ValueError: If the instrument type ID is empty, no reserved sessions + match the instrument type ID, or too many reserved sessions + match the instrument type ID. """ return self._create_session_core(session_constructor, instrument_type_id) @@ -404,20 +434,557 @@ def create_sessions( This is a generic method that supports any instrument driver. Args: - session_constructor: A function that constructs sessions based on session - information. + session_constructor: A function that constructs sessions based on + session information. + instrument_type_id: Instrument type ID for the session. - For NI instruments, use instrument type id constants, such as - :py:const:`INSTRUMENT_TYPE_NI_DCPOWER` or :py:const:`INSTRUMENT_TYPE_NI_DMM`. - For custom instruments, use the instrument type id defined in the pin map file. + + For custom instruments, use the instrument type id defined in + the pin map file. Returns: A context manager that yields a sequence of session information objects. The created sessions are available via the ``session`` field. + + Raises: + ValueError: If the instrument type ID is empty or no reserved + sessions match the instrument type ID. """ return self._create_sessions_core(session_constructor, instrument_type_id) + @requires_feature(SESSION_MANAGEMENT_2024Q1) + def create_nidcpower_session( + self, + reset: bool = False, + options: Optional[Dict[str, Any]] = None, + initialization_behavior: SessionInitializationBehavior = SessionInitializationBehavior.AUTO, + ) -> ContextManager[TypedSessionInformation[nidcpower.Session]]: + """Create a single NI-DCPower instrument session. + + Args: + reset: Specifies whether to reset channel(s) during the + initialization procedure. + + options: Specifies the initial value of certain properties for the + session. If this argument is not specified, the default value is + an empty dict, which you may override by specifying + ``NIDCPOWER_SIMULATE``, ``NIDCPOWER_BOARD_TYPE``, and + ``NIDCPOWER_MODEL`` in the configuration file (``.env``). + + initialization_behavior: Specifies whether to initialize a new + session or attach to an existing session. + + Returns: + A context manager that yields a session information object. The + created session is available via the ``session`` field. + + Raises: + ValueError: If no NI-DCPower sessions are reserved or too many + NI-DCPower sessions are reserved. + + See Also: + For more details, see :py:class:`nidcpower.Session`. + """ + from ni_measurementlink_service._drivers._nidcpower import SessionConstructor + + session_constructor = SessionConstructor( + self._discovery_client, self._grpc_channel_pool, reset, options, initialization_behavior + ) + return self._create_session_core(session_constructor, INSTRUMENT_TYPE_NI_DCPOWER) + + @requires_feature(SESSION_MANAGEMENT_2024Q1) + def create_nidcpower_sessions( + self, + reset: bool = False, + options: Optional[Dict[str, Any]] = None, + initialization_behavior: SessionInitializationBehavior = SessionInitializationBehavior.AUTO, + ) -> ContextManager[Sequence[TypedSessionInformation[nidcpower.Session]]]: + """Create multiple NI-DCPower instrument sessions. + + Args: + reset: Specifies whether to reset channel(s) during the + initialization procedure. + + options: Specifies the initial value of certain properties for the + session. If this argument is not specified, the default value is + an empty dict, which you may override by specifying + ``NIDCPOWER_SIMULATE``, ``NIDCPOWER_BOARD_TYPE``, and + ``NIDCPOWER_MODEL`` in the configuration file (``.env``). + + initialization_behavior: Specifies whether to initialize a new + session or attach to an existing session. + + Returns: + A context manager that yields a sequence of session information + objects. The created sessions are available via the ``session`` + field. + + Raises: + ValueError: If no NI-DCPower sessions are reserved. + + See Also: + For more details, see :py:class:`nidcpower.Session`. + """ + from ni_measurementlink_service._drivers._nidcpower import SessionConstructor + + session_constructor = SessionConstructor( + self._discovery_client, self._grpc_channel_pool, reset, options, initialization_behavior + ) + return self._create_sessions_core(session_constructor, INSTRUMENT_TYPE_NI_DCPOWER) + + @requires_feature(SESSION_MANAGEMENT_2024Q1) + def create_nidigital_session( + self, + reset_device: bool = False, + options: Optional[Dict[str, Any]] = None, + initialization_behavior: SessionInitializationBehavior = SessionInitializationBehavior.AUTO, + ) -> ContextManager[TypedSessionInformation[nidigital.Session]]: + """Create a single NI-Digital Pattern instrument session. + + Args: + reset_device: Specifies whether to reset the instrument during the + initialization procedure. + + options: Specifies the initial value of certain properties for the + session. If this argument is not specified, the default value is + an empty dict, which you may override by specifying + ``NIDIGITAL_SIMULATE``, ``NIDIGITAL_BOARD_TYPE``, and + ``NIDIGITAL_MODEL`` in the configuration file (``.env``). + + initialization_behavior: Specifies whether to initialize a new + session or attach to an existing session. + + Returns: + A context manager that yields a session information object. The + created session is available via the ``session`` field. + + Raises: + ValueError: If no NI-Digital sessions are reserved or too many + NI-Digital sessions are reserved. + + See Also: + For more details, see :py:class:`nidigital.Session`. + """ + from ni_measurementlink_service._drivers._nidigital import SessionConstructor + + session_constructor = SessionConstructor( + self._discovery_client, + self._grpc_channel_pool, + reset_device, + options, + initialization_behavior, + ) + return self._create_session_core(session_constructor, INSTRUMENT_TYPE_NI_DIGITAL_PATTERN) + + @requires_feature(SESSION_MANAGEMENT_2024Q1) + def create_nidigital_sessions( + self, + reset_device: bool = False, + options: Optional[Dict[str, Any]] = None, + initialization_behavior: SessionInitializationBehavior = SessionInitializationBehavior.AUTO, + ) -> ContextManager[Sequence[TypedSessionInformation[nidigital.Session]]]: + """Create multiple NI-Digital Pattern instrument sessions. + + Args: + reset_device: Specifies whether to reset the instrument during the + initialization procedure. + + options: Specifies the initial value of certain properties for the + session. If this argument is not specified, the default value is + an empty dict, which you may override by specifying + ``NIDIGITAL_SIMULATE``, ``NIDIGITAL_BOARD_TYPE``, and + ``NIDIGITAL_MODEL`` in the configuration file (``.env``). + + initialization_behavior: Specifies whether to initialize a new + session or attach to an existing session. + + Returns: + A context manager that yields a sequence of session information + objects. The created sessions are available via the ``session`` + field. + + Raises: + ValueError: If no NI-Digital sessions are reserved. + + See Also: + For more details, see :py:class:`nidigital.Session`. + """ + from ni_measurementlink_service._drivers._nidigital import SessionConstructor + + session_constructor = SessionConstructor( + self._discovery_client, + self._grpc_channel_pool, + reset_device, + options, + initialization_behavior, + ) + return self._create_sessions_core(session_constructor, INSTRUMENT_TYPE_NI_DIGITAL_PATTERN) + + @requires_feature(SESSION_MANAGEMENT_2024Q1) + def create_nidmm_session( + self, + reset_device: bool = False, + options: Optional[Dict[str, Any]] = None, + initialization_behavior: SessionInitializationBehavior = SessionInitializationBehavior.AUTO, + ) -> ContextManager[TypedSessionInformation[nidmm.Session]]: + """Create a single NI-DMM instrument session. + + Args: + reset_device: Specifies whether to reset the instrument during the + initialization procedure. + + options: Specifies the initial value of certain properties for the + session. If this argument is not specified, the default value is + an empty dict, which you may override by specifying + ``NIDMM_SIMULATE``, ``NIDMM_BOARD_TYPE``, and + ``NIDMM_MODEL`` in the configuration file (``.env``). + + initialization_behavior: Specifies whether to initialize a new + session or attach to an existing session. + + Returns: + A context manager that yields a session information object. The + created session is available via the ``session`` field. + + Raises: + ValueError: If no NI-DMM sessions are reserved or too many + NI-DMM sessions are reserved. + + See Also: + For more details, see :py:class:`nidmm.Session`. + """ + from ni_measurementlink_service._drivers._nidmm import SessionConstructor + + session_constructor = SessionConstructor( + self._discovery_client, + self._grpc_channel_pool, + reset_device, + options, + initialization_behavior, + ) + return self._create_session_core(session_constructor, INSTRUMENT_TYPE_NI_DMM) + + @requires_feature(SESSION_MANAGEMENT_2024Q1) + def create_nidmm_sessions( + self, + reset_device: bool = False, + options: Optional[Dict[str, Any]] = None, + initialization_behavior: SessionInitializationBehavior = SessionInitializationBehavior.AUTO, + ) -> ContextManager[Sequence[TypedSessionInformation[nidmm.Session]]]: + """Create multiple NI-DMM instrument sessions. + + Args: + reset_device: Specifies whether to reset the instrument during the + initialization procedure. + + options: Specifies the initial value of certain properties for the + session. If this argument is not specified, the default value is + an empty dict, which you may override by specifying + ``NIDMM_SIMULATE``, ``NIDMM_BOARD_TYPE``, and + ``NIDMM_MODEL`` in the configuration file (``.env``). + + initialization_behavior: Specifies whether to initialize a new + session or attach to an existing session. + + Returns: + A context manager that yields a sequence of session information + objects. The created sessions are available via the ``session`` + field. + + Raises: + ValueError: If no NI-DMM sessions are reserved. + + See Also: + For more details, see :py:class:`nidmm.Session`. + """ + from ni_measurementlink_service._drivers._nidmm import SessionConstructor + + session_constructor = SessionConstructor( + self._discovery_client, + self._grpc_channel_pool, + reset_device, + options, + initialization_behavior, + ) + return self._create_sessions_core(session_constructor, INSTRUMENT_TYPE_NI_DMM) + + @requires_feature(SESSION_MANAGEMENT_2024Q1) + def create_nifgen_session( + self, + reset_device: bool = False, + options: Optional[Dict[str, Any]] = None, + initialization_behavior: SessionInitializationBehavior = SessionInitializationBehavior.AUTO, + ) -> ContextManager[TypedSessionInformation[nifgen.Session]]: + """Create a single NI-FGEN instrument session. + + Args: + reset_device: Specifies whether to reset the instrument during the + initialization procedure. + + options: Specifies the initial value of certain properties for the + session. If this argument is not specified, the default value is + an empty dict, which you may override by specifying + ``NIFGEN_SIMULATE``, ``NIFGEN_BOARD_TYPE``, and ``NIFGEN_MODEL`` + in the configuration file (``.env``). + + initialization_behavior: Specifies whether to initialize a new + session or attach to an existing session. + + Returns: + A context manager that yields a session information object. The + created session is available via the ``session`` field. + + Raises: + ValueError: If no NI-FGEN sessions are reserved or too many NI-FGEN + sessions are reserved. + + See Also: + For more details, see :py:class:`nifgen.Session`. + """ + from ni_measurementlink_service._drivers._nifgen import SessionConstructor + + session_constructor = SessionConstructor( + self._discovery_client, + self._grpc_channel_pool, + reset_device, + options, + initialization_behavior, + ) + return self._create_session_core(session_constructor, INSTRUMENT_TYPE_NI_FGEN) + + @requires_feature(SESSION_MANAGEMENT_2024Q1) + def create_nifgen_sessions( + self, + reset_device: bool = False, + options: Optional[Dict[str, Any]] = None, + initialization_behavior: SessionInitializationBehavior = SessionInitializationBehavior.AUTO, + ) -> ContextManager[Sequence[TypedSessionInformation[nifgen.Session]]]: + """Create multiple NI-FGEN instrument sessions. + + Args: + reset_device: Specifies whether to reset the instrument during the + initialization procedure. + + options: Specifies the initial value of certain properties for the + session. If this argument is not specified, the default value is + an empty dict, which you may override by specifying + ``NIFGEN_SIMULATE``, ``NIFGEN_BOARD_TYPE``, and ``NIFGEN_MODEL`` + in the configuration file (``.env``). + + initialization_behavior: Specifies whether to initialize a new + session or attach to an existing session. + + Returns: + A context manager that yields a sequence of session information + objects. The created sessions are available via the ``session`` + field. + + Raises: + ValueError: If no NI-FGEN sessions are reserved. + + See Also: + For more details, see :py:class:`nifgen.Session`. + """ + from ni_measurementlink_service._drivers._nifgen import SessionConstructor + + session_constructor = SessionConstructor( + self._discovery_client, + self._grpc_channel_pool, + reset_device, + options, + initialization_behavior, + ) + return self._create_sessions_core(session_constructor, INSTRUMENT_TYPE_NI_FGEN) + + @requires_feature(SESSION_MANAGEMENT_2024Q1) + def create_niscope_session( + self, + reset_device: bool = False, + options: Optional[Dict[str, Any]] = None, + initialization_behavior: SessionInitializationBehavior = SessionInitializationBehavior.AUTO, + ) -> ContextManager[TypedSessionInformation[niscope.Session]]: + """Create a single NI-SCOPE instrument session. + + Args: + reset_device: Specifies whether to reset the instrument during the + initialization procedure. + + options: Specifies the initial value of certain properties for the + session. If this argument is not specified, the default value is + an empty dict, which you may override by specifying + ``NISCOPE_SIMULATE``, ``NISCOPE_BOARD_TYPE``, and + ``NISCOPE_MODEL`` in the configuration file (``.env``). + + initialization_behavior: Specifies whether to initialize a new + session or attach to an existing session. + + Returns: + A context manager that yields a session information object. The + created session is available via the ``session`` field. + + Raises: + ValueError: If no NI-SCOPE sessions are reserved or too many + NI-SCOPE sessions are reserved. + + See Also: + For more details, see :py:class:`niscope.Session`. + """ + from ni_measurementlink_service._drivers._niscope import SessionConstructor + + session_constructor = SessionConstructor( + self._discovery_client, + self._grpc_channel_pool, + reset_device, + options, + initialization_behavior, + ) + return self._create_session_core(session_constructor, INSTRUMENT_TYPE_NI_SCOPE) + + @requires_feature(SESSION_MANAGEMENT_2024Q1) + def create_niscope_sessions( + self, + reset_device: bool = False, + options: Optional[Dict[str, Any]] = None, + initialization_behavior: SessionInitializationBehavior = SessionInitializationBehavior.AUTO, + ) -> ContextManager[Sequence[TypedSessionInformation[niscope.Session]]]: + """Create multiple NI-SCOPE instrument sessions. + + Args: + reset_device: Specifies whether to reset the instrument during the + initialization procedure. + + options: Specifies the initial value of certain properties for the + session. If this argument is not specified, the default value is + an empty dict, which you may override by specifying + ``NISCOPE_SIMULATE``, ``NISCOPE_BOARD_TYPE``, and + ``NISCOPE_MODEL`` in the configuration file (``.env``). + + initialization_behavior: Specifies whether to initialize a new + session or attach to an existing session. + + Returns: + A context manager that yields a sequence of session information + objects. The created sessions are available via the ``session`` + field. + + Raises: + ValueError: If no NI-SCOPE sessions are reserved. + + See Also: + For more details, see :py:class:`niscope.Session`. + """ + from ni_measurementlink_service._drivers._niscope import SessionConstructor + + session_constructor = SessionConstructor( + self._discovery_client, + self._grpc_channel_pool, + reset_device, + options, + initialization_behavior, + ) + return self._create_sessions_core(session_constructor, INSTRUMENT_TYPE_NI_SCOPE) + + @requires_feature(SESSION_MANAGEMENT_2024Q1) + def create_niswitch_session( + self, + topology: Optional[str] = None, + simulate: Optional[bool] = None, + reset_device: bool = False, + initialization_behavior: SessionInitializationBehavior = SessionInitializationBehavior.AUTO, + ) -> ContextManager[TypedSessionInformation[niswitch.Session]]: + """Create a single NI-SWITCH relay driver instrument session. + + Args: + topology: Specifies the switch topology. If this argument is not + specified, the default value is "Configured Topology", which you + may override by setting ``NISWITCH_TOPOLOGY`` in the + configuration file (``.env``). + + simulate: Enables or disables simulation of the switch module. If + this argument is not specified, the default value is ``False``, + which you may override by setting ``NISWITCH_SIMULATE`` in the + configuration file (``.env``). + + reset_device: Specifies whether to reset the switch module during + the initialization procedure. + + initialization_behavior: Specifies whether to initialize a new + session or attach to an existing session. + + Returns: + A context manager that yields a session information object. The + created session is available via the ``session`` field. + + Raises: + ValueError: If no relay driver sessions are reserved or + too many relay driver sessions are reserved. + + See Also: + For more details, see :py:class:`niswitch.Session`. + """ + from ni_measurementlink_service._drivers._niswitch import SessionConstructor + + session_constructor = SessionConstructor( + self._discovery_client, + self._grpc_channel_pool, + topology, + simulate, + reset_device, + initialization_behavior, + ) + return self._create_session_core(session_constructor, INSTRUMENT_TYPE_NI_RELAY_DRIVER) + + @requires_feature(SESSION_MANAGEMENT_2024Q1) + def create_niswitch_sessions( + self, + topology: Optional[str] = None, + simulate: Optional[bool] = None, + reset_device: bool = False, + initialization_behavior: SessionInitializationBehavior = SessionInitializationBehavior.AUTO, + ) -> ContextManager[Sequence[TypedSessionInformation[niswitch.Session]]]: + """Create multiple NI-SWITCH relay driver instrument sessions. + + Args: + topology: Specifies the switch topology. If this argument is not + specified, the default value is "Configured Topology", which you + may override by setting ``NISWITCH_TOPOLOGY`` in the + configuration file (``.env``). + + simulate: Enables or disables simulation of the switch module. If + this argument is not specified, the default value is ``False``, + which you may override by setting ``NISWITCH_SIMULATE`` in the + configuration file (``.env``). + + reset_device: Specifies whether to reset the switch module during + the initialization procedure. + + initialization_behavior: Specifies whether to initialize a new + session or attach to an existing session. + + Returns: + A context manager that yields a sequence of session information + objects. The created sessions are available via the ``session`` + field. + + Raises: + ValueError: If no relay driver sessions are reserved. + + See Also: + For more details, see :py:class:`niswitch.Session`. + """ + from ni_measurementlink_service._drivers._niswitch import SessionConstructor + + session_constructor = SessionConstructor( + self._discovery_client, + self._grpc_channel_pool, + topology, + simulate, + reset_device, + initialization_behavior, + ) + return self._create_sessions_core(session_constructor, INSTRUMENT_TYPE_NI_RELAY_DRIVER) + class SingleSessionReservation(BaseReservation): """Manages reservation for a single session.""" diff --git a/poetry.lock b/poetry.lock index 8a0a1d11d..3a2d7c13c 100644 --- a/poetry.lock +++ b/poetry.lock @@ -242,6 +242,14 @@ grpcio = ">=1.59.0" protobuf = ">=4.21.6,<5.0dev" setuptools = "*" +[[package]] +name = "hightime" +version = "0.2.1" +description = "Hightime Python API" +category = "main" +optional = true +python-versions = "*" + [[package]] name = "idna" version = "3.4" @@ -403,6 +411,150 @@ isort = ">=5.10" pep8-naming = ">=0.11.1" toml = ">=0.10.1" +[[package]] +name = "nidaqmx" +version = "0.8.0" +description = "NI-DAQmx Python API" +category = "main" +optional = true +python-versions = ">=3.7,<4.0" + +[package.dependencies] +deprecation = ">=2.1" +grpcio = {version = ">=1.49.0,<2.0", optional = true, markers = "python_version >= \"3.8\" and python_version < \"4.0\" and extra == \"grpc\""} +numpy = {version = ">=1.22", markers = "python_version >= \"3.8\" and python_version < \"4.0\""} +protobuf = {version = ">=4.21,<5.0", optional = true, markers = "extra == \"grpc\""} + +[package.extras] +docs = ["Sphinx (>=4.3,<5.0)", "sphinx_rtd_theme (>=1.0,<2.0)"] +grpc = ["grpcio (>=1.49.0,<1.53)", "grpcio (>=1.49.0,<2.0)", "protobuf (>=4.21,<5.0)"] + +[[package]] +name = "nidcpower" +version = "1.4.6" +description = "NI-DCPower Python API" +category = "main" +optional = true +python-versions = "*" + +[package.dependencies] +grpcio = {version = ">=1.49.1,<2.0", optional = true, markers = "extra == \"grpc\""} +hightime = ">=0.2.0" +protobuf = {version = ">=4.21,<5.0", optional = true, markers = "extra == \"grpc\""} + +[package.extras] +grpc = ["grpcio (>=1.49.1,<2.0)", "protobuf (>=4.21,<5.0)"] + +[[package]] +name = "nidigital" +version = "1.4.6" +description = "NI-Digital Pattern Driver Python API" +category = "main" +optional = true +python-versions = "*" + +[package.dependencies] +grpcio = {version = ">=1.49.1,<2.0", optional = true, markers = "extra == \"grpc\""} +hightime = ">=0.2.0" +nitclk = "*" +protobuf = {version = ">=4.21,<5.0", optional = true, markers = "extra == \"grpc\""} + +[package.extras] +grpc = ["grpcio (>=1.49.1,<2.0)", "protobuf (>=4.21,<5.0)"] + +[[package]] +name = "nidmm" +version = "1.4.6" +description = "NI-DMM Python API" +category = "main" +optional = true +python-versions = "*" + +[package.dependencies] +grpcio = {version = ">=1.49.1,<2.0", optional = true, markers = "extra == \"grpc\""} +hightime = ">=0.2.0" +protobuf = {version = ">=4.21,<5.0", optional = true, markers = "extra == \"grpc\""} + +[package.extras] +grpc = ["grpcio (>=1.49.1,<2.0)", "protobuf (>=4.21,<5.0)"] + +[[package]] +name = "nifgen" +version = "1.4.6" +description = "NI-FGEN Python API" +category = "main" +optional = true +python-versions = "*" + +[package.dependencies] +grpcio = {version = ">=1.49.1,<2.0", optional = true, markers = "extra == \"grpc\""} +hightime = ">=0.2.0" +nitclk = "*" +protobuf = {version = ">=4.21,<5.0", optional = true, markers = "extra == \"grpc\""} + +[package.extras] +grpc = ["grpcio (>=1.49.1,<2.0)", "protobuf (>=4.21,<5.0)"] + +[[package]] +name = "niscope" +version = "1.4.6" +description = "NI-SCOPE Python API" +category = "main" +optional = true +python-versions = "*" + +[package.dependencies] +grpcio = {version = ">=1.49.1,<2.0", optional = true, markers = "extra == \"grpc\""} +hightime = ">=0.2.0" +nitclk = "*" +protobuf = {version = ">=4.21,<5.0", optional = true, markers = "extra == \"grpc\""} + +[package.extras] +grpc = ["grpcio (>=1.49.1,<2.0)", "protobuf (>=4.21,<5.0)"] + +[[package]] +name = "niswitch" +version = "1.4.6" +description = "NI-SWITCH Python API" +category = "main" +optional = true +python-versions = "*" + +[package.dependencies] +grpcio = {version = ">=1.49.1,<2.0", optional = true, markers = "extra == \"grpc\""} +hightime = ">=0.2.0" +protobuf = {version = ">=4.21,<5.0", optional = true, markers = "extra == \"grpc\""} + +[package.extras] +grpc = ["grpcio (>=1.49.1,<2.0)", "protobuf (>=4.21,<5.0)"] + +[[package]] +name = "nitclk" +version = "1.4.6" +description = "NI-TClk Python API" +category = "main" +optional = true +python-versions = "*" + +[package.dependencies] +hightime = ">=0.2.0" + +[[package]] +name = "numpy" +version = "1.24.4" +description = "Fundamental package for array computing in Python" +category = "main" +optional = false +python-versions = ">=3.8" + +[[package]] +name = "numpy" +version = "1.26.0" +description = "Fundamental package for array computing in Python" +category = "main" +optional = false +python-versions = "<3.13,>=3.9" + [[package]] name = "packaging" version = "23.2" @@ -895,10 +1047,20 @@ python-versions = ">=3.8" docs = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "rst.linker (>=1.9)", "sphinx (<7.2.5)", "sphinx (>=3.5)", "sphinx-lint"] testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "pytest (>=6)", "pytest-black (>=0.3.7)", "pytest-checkdocs (>=2.4)", "pytest-cov", "pytest-enabler (>=2.2)", "pytest-ignore-flaky", "pytest-mypy (>=0.9.1)", "pytest-ruff"] +[extras] +drivers = ["nidaqmx", "nidcpower", "nidigital", "nidmm", "nifgen", "niscope", "niswitch"] +nidaqmx = ["nidaqmx"] +nidcpower = ["nidcpower"] +nidigital = ["nidigital"] +nidmm = ["nidmm"] +nifgen = ["nifgen"] +niscope = ["niscope"] +niswitch = ["niswitch"] + [metadata] lock-version = "1.1" python-versions = "^3.8" -content-hash = "b3cc50c61ff18c4648205e059b9a6b8d7dc0de0013df74a5f3f70c1cd423f783" +content-hash = "ba553ec832ef1e6a50475e591366029cc134a6ce67aec4ab95fc5254c6b2cc8a" [metadata.files] alabaster = [ @@ -1288,6 +1450,9 @@ grpcio-tools = [ {file = "grpcio_tools-1.59.0-cp39-cp39-win32.whl", hash = "sha256:a4f6cae381f21fee1ef0a5cbbbb146680164311157ae618edf3061742d844383"}, {file = "grpcio_tools-1.59.0-cp39-cp39-win_amd64.whl", hash = "sha256:4a10e59cca462208b489478340b52a96d64e8b8b6f1ac097f3e8cb211d3f66c0"}, ] +hightime = [ + {file = "hightime-0.2.1-py3-none-any.whl", hash = "sha256:a300434692de16273ce46f9c1f2de01e90091b151ccfb4e0973a6a6e1d51f032"}, +] idna = [ {file = "idna-3.4-py3-none-any.whl", hash = "sha256:90b77e79eaa3eba6de819a0c442c0b4ceefc341a7a2ab77d7562bf49f425c5c2"}, {file = "idna-3.4.tar.gz", hash = "sha256:814f528e8dead7d329833b91c5faa87d60bf71824cd12a7530b5526063d02cb4"}, @@ -1427,6 +1592,100 @@ ni-python-styleguide = [ {file = "ni_python_styleguide-0.4.1-py3-none-any.whl", hash = "sha256:aee987fe56ae5c7c05063a0c8acff5fa5dabf22d14436260e203f72616d5be0e"}, {file = "ni_python_styleguide-0.4.1.tar.gz", hash = "sha256:12d19b7a53b94ef2803b7008bb14a552e81b299d807b070408c26759e15f14f2"}, ] +nidaqmx = [ + {file = "nidaqmx-0.8.0-py3-none-any.whl", hash = "sha256:7a5d5b3f78c125d31a8d00349f216f63bb8f6b887d0eca882001b729904d68bc"}, + {file = "nidaqmx-0.8.0.tar.gz", hash = "sha256:6c72ca4ec5da95e8603c1e6d5085d59bb3c1c2ea1f45bdfaa4acac40fe82ae0c"}, +] +nidcpower = [ + {file = "nidcpower-1.4.6-py3-none-any.whl", hash = "sha256:4a4c2ccf2e04aabec9ac9b57ea08a9e748dd888ab3503d61aa93336924946906"}, + {file = "nidcpower-1.4.6.tar.gz", hash = "sha256:47e2131a2a18a64661533fb43cd7039ec18c7cf4b5882df3aebe91279416744c"}, +] +nidigital = [ + {file = "nidigital-1.4.6-py3-none-any.whl", hash = "sha256:d79016c45fb41526acd43b92221af6a11f4b8ab5a71b833867cb9934768fcd30"}, + {file = "nidigital-1.4.6.tar.gz", hash = "sha256:d059aef452a150be842033cd910a5e5ba344acaa4cba109d009c8dde73059550"}, +] +nidmm = [ + {file = "nidmm-1.4.6-py3-none-any.whl", hash = "sha256:8ce6724e7c3eff35d83dd0d3bfbe2184b8468027bc440da2c49d24a53d927aed"}, + {file = "nidmm-1.4.6.tar.gz", hash = "sha256:95ad39272a454d6f5826fed90475de5f670536a41ba767303e6994f0eb36d2a0"}, +] +nifgen = [ + {file = "nifgen-1.4.6-py3-none-any.whl", hash = "sha256:d484360ddf499ac99fab8fcb54b36a526b9708d022ea8ea9ee50d1591cf11f90"}, + {file = "nifgen-1.4.6.tar.gz", hash = "sha256:40f81d70046ced9cc3239f71cd7d26332ab68d36ddbe9e2c446874b699ba12e0"}, +] +niscope = [ + {file = "niscope-1.4.6-py3-none-any.whl", hash = "sha256:b993cf0f1b36605398c5cc879c875eabd95ca44b2738a0265453431a6bdc9901"}, + {file = "niscope-1.4.6.tar.gz", hash = "sha256:cadb120820c4e531e8dda4751e6bfe6e81838130559a1be94ad18e0361b64246"}, +] +niswitch = [ + {file = "niswitch-1.4.6-py3-none-any.whl", hash = "sha256:e04b275de01fec29d9a7c602385bd3c56e8160eee9c72b39cc9cdb5756b99426"}, + {file = "niswitch-1.4.6.tar.gz", hash = "sha256:42091a8c1b49186831c0c710c631f8ca0bcaf1b85cbc46f35fd971d0e1fb8d90"}, +] +nitclk = [ + {file = "nitclk-1.4.6-py3-none-any.whl", hash = "sha256:68d5e1761d534c7474d7c55382a624090fac1cfa95e101e58b8a98e30cf659b1"}, + {file = "nitclk-1.4.6.tar.gz", hash = "sha256:6aac6e8646b28411058040df8b7ce825c2895d5a05b1026319b7c34c7a2e3781"}, +] +numpy = [ + {file = "numpy-1.24.4-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:c0bfb52d2169d58c1cdb8cc1f16989101639b34c7d3ce60ed70b19c63eba0b64"}, + {file = "numpy-1.24.4-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:ed094d4f0c177b1b8e7aa9cba7d6ceed51c0e569a5318ac0ca9a090680a6a1b1"}, + {file = "numpy-1.24.4-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:79fc682a374c4a8ed08b331bef9c5f582585d1048fa6d80bc6c35bc384eee9b4"}, + {file = "numpy-1.24.4-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:7ffe43c74893dbf38c2b0a1f5428760a1a9c98285553c89e12d70a96a7f3a4d6"}, + {file = "numpy-1.24.4-cp310-cp310-win32.whl", hash = "sha256:4c21decb6ea94057331e111a5bed9a79d335658c27ce2adb580fb4d54f2ad9bc"}, + {file = "numpy-1.24.4-cp310-cp310-win_amd64.whl", hash = "sha256:b4bea75e47d9586d31e892a7401f76e909712a0fd510f58f5337bea9572c571e"}, + {file = "numpy-1.24.4-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:f136bab9c2cfd8da131132c2cf6cc27331dd6fae65f95f69dcd4ae3c3639c810"}, + {file = "numpy-1.24.4-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:e2926dac25b313635e4d6cf4dc4e51c8c0ebfed60b801c799ffc4c32bf3d1254"}, + {file = "numpy-1.24.4-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:222e40d0e2548690405b0b3c7b21d1169117391c2e82c378467ef9ab4c8f0da7"}, + {file = "numpy-1.24.4-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:7215847ce88a85ce39baf9e89070cb860c98fdddacbaa6c0da3ffb31b3350bd5"}, + {file = "numpy-1.24.4-cp311-cp311-win32.whl", hash = "sha256:4979217d7de511a8d57f4b4b5b2b965f707768440c17cb70fbf254c4b225238d"}, + {file = "numpy-1.24.4-cp311-cp311-win_amd64.whl", hash = "sha256:b7b1fc9864d7d39e28f41d089bfd6353cb5f27ecd9905348c24187a768c79694"}, + {file = "numpy-1.24.4-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:1452241c290f3e2a312c137a9999cdbf63f78864d63c79039bda65ee86943f61"}, + {file = "numpy-1.24.4-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:04640dab83f7c6c85abf9cd729c5b65f1ebd0ccf9de90b270cd61935eef0197f"}, + {file = "numpy-1.24.4-cp38-cp38-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a5425b114831d1e77e4b5d812b69d11d962e104095a5b9c3b641a218abcc050e"}, + {file = "numpy-1.24.4-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:dd80e219fd4c71fc3699fc1dadac5dcf4fd882bfc6f7ec53d30fa197b8ee22dc"}, + {file = "numpy-1.24.4-cp38-cp38-win32.whl", hash = "sha256:4602244f345453db537be5314d3983dbf5834a9701b7723ec28923e2889e0bb2"}, + {file = "numpy-1.24.4-cp38-cp38-win_amd64.whl", hash = "sha256:692f2e0f55794943c5bfff12b3f56f99af76f902fc47487bdfe97856de51a706"}, + {file = "numpy-1.24.4-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:2541312fbf09977f3b3ad449c4e5f4bb55d0dbf79226d7724211acc905049400"}, + {file = "numpy-1.24.4-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:9667575fb6d13c95f1b36aca12c5ee3356bf001b714fc354eb5465ce1609e62f"}, + {file = "numpy-1.24.4-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:f3a86ed21e4f87050382c7bc96571755193c4c1392490744ac73d660e8f564a9"}, + {file = "numpy-1.24.4-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d11efb4dbecbdf22508d55e48d9c8384db795e1b7b51ea735289ff96613ff74d"}, + {file = "numpy-1.24.4-cp39-cp39-win32.whl", hash = "sha256:6620c0acd41dbcb368610bb2f4d83145674040025e5536954782467100aa8835"}, + {file = "numpy-1.24.4-cp39-cp39-win_amd64.whl", hash = "sha256:befe2bf740fd8373cf56149a5c23a0f601e82869598d41f8e188a0e9869926f8"}, + {file = "numpy-1.24.4-pp38-pypy38_pp73-macosx_10_9_x86_64.whl", hash = "sha256:31f13e25b4e304632a4619d0e0777662c2ffea99fcae2029556b17d8ff958aef"}, + {file = "numpy-1.24.4-pp38-pypy38_pp73-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:95f7ac6540e95bc440ad77f56e520da5bf877f87dca58bd095288dce8940532a"}, + {file = "numpy-1.24.4-pp38-pypy38_pp73-win_amd64.whl", hash = "sha256:e98f220aa76ca2a977fe435f5b04d7b3470c0a2e6312907b37ba6068f26787f2"}, + {file = "numpy-1.24.4.tar.gz", hash = "sha256:80f5e3a4e498641401868df4208b74581206afbee7cf7b8329daae82676d9463"}, + {file = "numpy-1.26.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:f8db2f125746e44dce707dd44d4f4efeea8d7e2b43aace3f8d1f235cfa2733dd"}, + {file = "numpy-1.26.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:0621f7daf973d34d18b4e4bafb210bbaf1ef5e0100b5fa750bd9cde84c7ac292"}, + {file = "numpy-1.26.0-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:51be5f8c349fdd1a5568e72713a21f518e7d6707bcf8503b528b88d33b57dc68"}, + {file = "numpy-1.26.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:767254ad364991ccfc4d81b8152912e53e103ec192d1bb4ea6b1f5a7117040be"}, + {file = "numpy-1.26.0-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:436c8e9a4bdeeee84e3e59614d38c3dbd3235838a877af8c211cfcac8a80b8d3"}, + {file = "numpy-1.26.0-cp310-cp310-win32.whl", hash = "sha256:c2e698cb0c6dda9372ea98a0344245ee65bdc1c9dd939cceed6bb91256837896"}, + {file = "numpy-1.26.0-cp310-cp310-win_amd64.whl", hash = "sha256:09aaee96c2cbdea95de76ecb8a586cb687d281c881f5f17bfc0fb7f5890f6b91"}, + {file = "numpy-1.26.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:637c58b468a69869258b8ae26f4a4c6ff8abffd4a8334c830ffb63e0feefe99a"}, + {file = "numpy-1.26.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:306545e234503a24fe9ae95ebf84d25cba1fdc27db971aa2d9f1ab6bba19a9dd"}, + {file = "numpy-1.26.0-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:8c6adc33561bd1d46f81131d5352348350fc23df4d742bb246cdfca606ea1208"}, + {file = "numpy-1.26.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:e062aa24638bb5018b7841977c360d2f5917268d125c833a686b7cbabbec496c"}, + {file = "numpy-1.26.0-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:546b7dd7e22f3c6861463bebb000646fa730e55df5ee4a0224408b5694cc6148"}, + {file = "numpy-1.26.0-cp311-cp311-win32.whl", hash = "sha256:c0b45c8b65b79337dee5134d038346d30e109e9e2e9d43464a2970e5c0e93229"}, + {file = "numpy-1.26.0-cp311-cp311-win_amd64.whl", hash = "sha256:eae430ecf5794cb7ae7fa3808740b015aa80747e5266153128ef055975a72b99"}, + {file = "numpy-1.26.0-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:166b36197e9debc4e384e9c652ba60c0bacc216d0fc89e78f973a9760b503388"}, + {file = "numpy-1.26.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:f042f66d0b4ae6d48e70e28d487376204d3cbf43b84c03bac57e28dac6151581"}, + {file = "numpy-1.26.0-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:e5e18e5b14a7560d8acf1c596688f4dfd19b4f2945b245a71e5af4ddb7422feb"}, + {file = "numpy-1.26.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:7f6bad22a791226d0a5c7c27a80a20e11cfe09ad5ef9084d4d3fc4a299cca505"}, + {file = "numpy-1.26.0-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:4acc65dd65da28060e206c8f27a573455ed724e6179941edb19f97e58161bb69"}, + {file = "numpy-1.26.0-cp312-cp312-win32.whl", hash = "sha256:bb0d9a1aaf5f1cb7967320e80690a1d7ff69f1d47ebc5a9bea013e3a21faec95"}, + {file = "numpy-1.26.0-cp312-cp312-win_amd64.whl", hash = "sha256:ee84ca3c58fe48b8ddafdeb1db87388dce2c3c3f701bf447b05e4cfcc3679112"}, + {file = "numpy-1.26.0-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:4a873a8180479bc829313e8d9798d5234dfacfc2e8a7ac188418189bb8eafbd2"}, + {file = "numpy-1.26.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:914b28d3215e0c721dc75db3ad6d62f51f630cb0c277e6b3bcb39519bed10bd8"}, + {file = "numpy-1.26.0-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:c78a22e95182fb2e7874712433eaa610478a3caf86f28c621708d35fa4fd6e7f"}, + {file = "numpy-1.26.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:86f737708b366c36b76e953c46ba5827d8c27b7a8c9d0f471810728e5a2fe57c"}, + {file = "numpy-1.26.0-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:b44e6a09afc12952a7d2a58ca0a2429ee0d49a4f89d83a0a11052da696440e49"}, + {file = "numpy-1.26.0-cp39-cp39-win32.whl", hash = "sha256:5671338034b820c8d58c81ad1dafc0ed5a00771a82fccc71d6438df00302094b"}, + {file = "numpy-1.26.0-cp39-cp39-win_amd64.whl", hash = "sha256:020cdbee66ed46b671429c7265cf00d8ac91c046901c55684954c3958525dab2"}, + {file = "numpy-1.26.0-pp39-pypy39_pp73-macosx_10_9_x86_64.whl", hash = "sha256:0792824ce2f7ea0c82ed2e4fecc29bb86bee0567a080dacaf2e0a01fe7654369"}, + {file = "numpy-1.26.0-pp39-pypy39_pp73-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:7d484292eaeb3e84a51432a94f53578689ffdea3f90e10c8b203a99be5af57d8"}, + {file = "numpy-1.26.0-pp39-pypy39_pp73-win_amd64.whl", hash = "sha256:186ba67fad3c60dbe8a3abff3b67a91351100f2661c8e2a80364ae6279720299"}, + {file = "numpy-1.26.0.tar.gz", hash = "sha256:f93fc78fe8bf15afe2b8d6b6499f1c73953169fad1e9a8dd086cdff3190e7fdf"}, +] packaging = [ {file = "packaging-23.2-py3-none-any.whl", hash = "sha256:8c491190033a9af7e1d931d0b5dacc2ef47509b34dd0de67ed209b5203fc88c7"}, {file = "packaging-23.2.tar.gz", hash = "sha256:048fb0e9405036518eaaf48a55953c750c11e1a1b68e0dd1a9d62ed0c092cfc5"}, diff --git a/pyproject.toml b/pyproject.toml index 7aa449481..c1333e869 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,6 +37,23 @@ deprecation = ">=2.1" # https://github.com/microsoft/tracelogging/issues/58 - traceloggingdynamic raises TypeError with Python 3.8 traceloggingdynamic = {version = ">=1.0", platform = "win32", python = "^3.9"} python-decouple = ">=3.8" +nidaqmx = {version = ">=0.8.0", extras = ["grpc"], optional = true} +nidcpower = {version = ">=1.4.4", extras = ["grpc"], optional = true} +nidigital = {version = ">=1.4.4", extras = ["grpc"], optional = true} +nidmm = {version = ">=1.4.4", extras = ["grpc"], optional = true} +nifgen = {version = ">=1.4.4", extras = ["grpc"], optional = true} +niscope = {version = ">=1.4.4", extras = ["grpc"], optional = true} +niswitch = {version = ">=1.4.4", extras = ["grpc"], optional = true} + +[tool.poetry.extras] +drivers = ["nidaqmx", "nidcpower", "nidigital", "nidmm", "nifgen", "niscope", "niswitch"] +nidaqmx = ["nidaqmx"] +nidcpower = ["nidcpower"] +nidigital = ["nidigital"] +nidmm = ["nidmm"] +nifgen = ["nifgen"] +niscope = ["niscope"] +niswitch = ["niswitch"] [tool.poetry.group.dev.dependencies] pytest = ">=7.2.0" @@ -57,6 +74,12 @@ types-pywin32 = ">=304" grpc-stubs = "^1.53" psutil = ">=5.9" types-psutil = ">=5.9" +# NumPy dropped support for Python 3.8 before adding support for Python 3.12, so +# we need to include multiple NumPy versions in poetry.lock. +numpy = [ + {version = ">=1.22", python = ">=3.8,<3.12"}, + {version = ">=1.26", python = ">=3.12,<3.13"}, +] [tool.poetry.group.docs] optional = true @@ -88,6 +111,7 @@ markers = [ [tool.mypy] disallow_untyped_defs = true namespace_packages = true +plugins = "numpy.typing.mypy_plugin" warn_redundant_casts = true warn_unused_configs = true warn_unused_ignores = true @@ -101,6 +125,15 @@ module = [ "grpc.framework.foundation.*", # 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 + "nidaqmx", + # https://github.com/ni/nimi-python/issues/1887 - Support type annotations + "nidcpower", + "nidigital", + "nidmm", + "nifgen", + "niscope", + "niswitch", ] ignore_missing_imports = true diff --git a/tests/unit/_drivers/_driver_utils.py b/tests/unit/_drivers/_driver_utils.py new file mode 100644 index 000000000..f7e36393f --- /dev/null +++ b/tests/unit/_drivers/_driver_utils.py @@ -0,0 +1,32 @@ +"""Driver-related unit test utilities.""" +from typing import List, Type, TypeVar +from unittest.mock import Mock, create_autospec + +from pytest_mock import MockerFixture + +from ni_measurementlink_service._configuration import MIDriverOptions + +TSession = TypeVar("TSession") + + +def create_mock_session(session_type: Type[TSession]) -> Mock: + """Create a single mock session object.""" + mock = create_autospec(session_type) + mock.__enter__.return_value = mock + mock.__exit__.return_value = None + return mock + + +def create_mock_sessions(session_type: Type[TSession], count: int) -> List[Mock]: + """Create multiple mock session objects.""" + return [create_mock_session(session_type) for _ in range(count)] + + +def set_simulation_options( + driver_name: str, mocker: MockerFixture, simulate: bool, board_type: str, model: str +) -> None: + """Set simulation options for the specified driver.""" + mocker.patch( + f"ni_measurementlink_service._drivers._{driver_name}.{driver_name.upper()}_OPTIONS", + MIDriverOptions(driver_name, simulate, board_type, model), + ) diff --git a/tests/unit/_drivers/test_nidcpower.py b/tests/unit/_drivers/test_nidcpower.py new file mode 100644 index 000000000..dc232127d --- /dev/null +++ b/tests/unit/_drivers/test_nidcpower.py @@ -0,0 +1,143 @@ +import functools +from unittest.mock import ANY, Mock + +import pytest +from pytest_mock import MockerFixture + +from ni_measurementlink_service.session_management import ( + INSTRUMENT_TYPE_NI_DCPOWER, + MultiSessionReservation, + SessionInitializationBehavior, +) +from tests.unit._drivers._driver_utils import ( + create_mock_session, + create_mock_sessions, + set_simulation_options, +) +from tests.unit._reservation_utils import create_grpc_session_infos + +try: + import nidcpower +except ImportError: + nidcpower = None + +pytestmark = pytest.mark.skipif(nidcpower is None, reason="Requires 'nidcpower' package.") + +if nidcpower: + # Note: this reads the Session type before it is patched. + create_mock_nidcpower_session = functools.partial(create_mock_session, nidcpower.Session) + create_mock_nidcpower_sessions = functools.partial(create_mock_sessions, nidcpower.Session) + create_nidcpower_session_infos = functools.partial( + create_grpc_session_infos, INSTRUMENT_TYPE_NI_DCPOWER + ) + set_nidcpower_simulation_options = functools.partial(set_simulation_options, "nidcpower") + + +def test___single_session_info___create_nidcpower_session___session_created( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation( + session_management_client, create_nidcpower_session_infos(1) + ) + session = create_mock_nidcpower_session() + session_type.side_effect = [session] + + with reservation.create_nidcpower_session() as session_info: + assert session_info.session is session + + session_type.assert_called_once_with( + resource_name="Dev0", reset=False, options={}, grpc_options=ANY + ) + + +def test___multiple_session_infos___create_nidcpower_sessions___sessions_created( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation( + session_management_client, create_nidcpower_session_infos(2) + ) + sessions = create_mock_nidcpower_sessions(3) + session_type.side_effect = sessions + + with reservation.create_nidcpower_sessions() as session_info: + assert session_info[0].session == sessions[0] + assert session_info[1].session == sessions[1] + + session_type.assert_any_call(resource_name="Dev0", reset=False, options={}, grpc_options=ANY) + session_type.assert_any_call(resource_name="Dev1", reset=False, options={}, grpc_options=ANY) + + +def test___optional_args___create_nidcpower_session___optional_args_passed( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation( + session_management_client, create_nidcpower_session_infos(1) + ) + session = create_mock_nidcpower_session() + session_type.side_effect = [session] + + with reservation.create_nidcpower_session( + reset=True, + options={"simulate": False}, + initialization_behavior=SessionInitializationBehavior.INITIALIZE_SERVER_SESSION, + ): + pass + + session_type.assert_called_once_with( + resource_name="Dev0", reset=True, options={"simulate": False}, grpc_options=ANY + ) + assert ( + session_type.call_args.kwargs["grpc_options"].initialization_behavior + == nidcpower.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION + ) + + +def test___simulation_configured___create_nidcpower_session___simulation_options_passed( + mocker: MockerFixture, + session_type: Mock, + session_management_client: Mock, +) -> None: + set_nidcpower_simulation_options(mocker, True, "PXIe", "4147") + reservation = MultiSessionReservation( + session_management_client, create_nidcpower_session_infos(1) + ) + session = create_mock_nidcpower_session() + session_type.side_effect = [session] + + with reservation.create_nidcpower_session(): + pass + + expected_options = {"simulate": True, "driver_setup": {"BoardType": "PXIe", "Model": "4147"}} + session_type.assert_called_once_with( + resource_name="Dev0", reset=False, options=expected_options, grpc_options=ANY + ) + + +def test___optional_args_and_simulation_configured___create_nidcpower_session___optional_args_passed( + mocker: MockerFixture, + session_type: Mock, + session_management_client: Mock, +) -> None: + set_nidcpower_simulation_options(mocker, True, "PXIe", "4147") + reservation = MultiSessionReservation( + session_management_client, create_nidcpower_session_infos(1) + ) + session = create_mock_nidcpower_session() + session_type.side_effect = [session] + + with reservation.create_nidcpower_session(reset=True, options={"simulate": False}): + pass + + expected_options = {"simulate": False} + session_type.assert_called_once_with( + resource_name="Dev0", reset=True, options=expected_options, grpc_options=ANY + ) + + +@pytest.fixture +def session_type(mocker: MockerFixture) -> Mock: + """A test fixture that replaces the Session class with a mock.""" + return mocker.patch("nidcpower.Session", autospec=True) diff --git a/tests/unit/_drivers/test_nidigital.py b/tests/unit/_drivers/test_nidigital.py new file mode 100644 index 000000000..b23dc9b7b --- /dev/null +++ b/tests/unit/_drivers/test_nidigital.py @@ -0,0 +1,159 @@ +import functools +from unittest.mock import ANY, Mock + +import pytest +from pytest_mock import MockerFixture + +from ni_measurementlink_service.session_management import ( + INSTRUMENT_TYPE_NI_DIGITAL_PATTERN, + MultiSessionReservation, + SessionInitializationBehavior, +) +from tests.unit._drivers._driver_utils import ( + create_mock_session, + create_mock_sessions, + set_simulation_options, +) +from tests.unit._reservation_utils import create_grpc_session_infos + +try: + import nidigital +except ImportError: + nidigital = None + +pytestmark = pytest.mark.skipif(nidigital is None, reason="Requires 'nidigital' package.") + +if nidigital: + # Note: this reads the Session type before it is patched. + create_mock_nidigital_session = functools.partial(create_mock_session, nidigital.Session) + create_mock_nidigital_sessions = functools.partial(create_mock_sessions, nidigital.Session) + create_nidigital_session_infos = functools.partial( + create_grpc_session_infos, INSTRUMENT_TYPE_NI_DIGITAL_PATTERN + ) + set_nidigital_simulation_options = functools.partial(set_simulation_options, "nidigital") + + +def test___single_session_info___create_nidigital_session___session_created( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation( + session_management_client, create_nidigital_session_infos(1) + ) + session = create_mock_nidigital_session() + session_type.side_effect = [session] + + with reservation.create_nidigital_session() as session_info: + assert session_info.session is session + + session_type.assert_called_once_with( + resource_name="Dev0", reset_device=False, options={}, grpc_options=ANY + ) + + +def test___multiple_session_infos___create_nidigital_sessions___sessions_created( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation( + session_management_client, create_nidigital_session_infos(2) + ) + sessions = create_mock_nidigital_sessions(3) + session_type.side_effect = sessions + + with reservation.create_nidigital_sessions() as session_info: + assert session_info[0].session == sessions[0] + assert session_info[1].session == sessions[1] + + session_type.assert_any_call( + resource_name="Dev0", reset_device=False, options={}, grpc_options=ANY + ) + session_type.assert_any_call( + resource_name="Dev1", reset_device=False, options={}, grpc_options=ANY + ) + + +def test___optional_args___create_nidigital_session___optional_args_passed( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation( + session_management_client, create_nidigital_session_infos(1) + ) + session = create_mock_nidigital_session() + session_type.side_effect = [session] + + with reservation.create_nidigital_session( + reset_device=True, + options={"simulate": False}, + initialization_behavior=SessionInitializationBehavior.INITIALIZE_SERVER_SESSION, + ): + pass + + session_type.assert_called_once_with( + resource_name="Dev0", + reset_device=True, + options={"simulate": False}, + grpc_options=ANY, + ) + assert ( + session_type.call_args.kwargs["grpc_options"].initialization_behavior + == nidigital.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION + ) + + +def test___simulation_configured___create_nidigital_session___simulation_options_passed( + mocker: MockerFixture, + session_type: Mock, + session_management_client: Mock, +) -> None: + set_nidigital_simulation_options(mocker, True, "PXIe", "6570") + reservation = MultiSessionReservation( + session_management_client, create_nidigital_session_infos(1) + ) + session = create_mock_nidigital_session() + session_type.side_effect = [session] + + with reservation.create_nidigital_session(): + pass + + expected_options = { + "simulate": True, + "driver_setup": {"BoardType": "PXIe", "Model": "6570"}, + } + session_type.assert_called_once_with( + resource_name="Dev0", + reset_device=False, + options=expected_options, + grpc_options=ANY, + ) + + +def test___optional_args_and_simulation_configured___create_nidigital_session___optional_args_passed( + mocker: MockerFixture, + session_type: Mock, + session_management_client: Mock, +) -> None: + set_nidigital_simulation_options(mocker, True, "PXIe", "6570") + reservation = MultiSessionReservation( + session_management_client, create_nidigital_session_infos(1) + ) + session = create_mock_nidigital_session() + session_type.side_effect = [session] + + with reservation.create_nidigital_session(reset_device=True, options={"simulate": False}): + pass + + expected_options = {"simulate": False} + session_type.assert_called_once_with( + resource_name="Dev0", + reset_device=True, + options=expected_options, + grpc_options=ANY, + ) + + +@pytest.fixture +def session_type(mocker: MockerFixture) -> Mock: + """A test fixture that replaces the Session class with a mock.""" + return mocker.patch("nidigital.Session", autospec=True) diff --git a/tests/unit/_drivers/test_nidmm.py b/tests/unit/_drivers/test_nidmm.py new file mode 100644 index 000000000..10cf8aacd --- /dev/null +++ b/tests/unit/_drivers/test_nidmm.py @@ -0,0 +1,149 @@ +import functools +from unittest.mock import ANY, Mock + +import pytest +from pytest_mock import MockerFixture + +from ni_measurementlink_service.session_management import ( + INSTRUMENT_TYPE_NI_DMM, + MultiSessionReservation, + SessionInitializationBehavior, +) +from tests.unit._drivers._driver_utils import ( + create_mock_session, + create_mock_sessions, + set_simulation_options, +) +from tests.unit._reservation_utils import create_grpc_session_infos + +try: + import nidmm +except ImportError: + nidmm = None + +pytestmark = pytest.mark.skipif(nidmm is None, reason="Requires 'nidmm' package.") + +if nidmm: + # Note: this reads the Session type before it is patched. + create_mock_nidmm_session = functools.partial(create_mock_session, nidmm.Session) + create_mock_nidmm_sessions = functools.partial(create_mock_sessions, nidmm.Session) + create_nidmm_session_infos = functools.partial( + create_grpc_session_infos, INSTRUMENT_TYPE_NI_DMM + ) + set_nidmm_simulation_options = functools.partial(set_simulation_options, "nidmm") + + +def test___single_session_info___create_nidmm_session___session_created( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation(session_management_client, create_nidmm_session_infos(1)) + session = create_mock_nidmm_session() + session_type.side_effect = [session] + + with reservation.create_nidmm_session() as session_info: + assert session_info.session is session + + session_type.assert_called_once_with( + resource_name="Dev0", reset_device=False, options={}, grpc_options=ANY + ) + + +def test___multiple_session_infos___create_nidmm_sessions___sessions_created( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation(session_management_client, create_nidmm_session_infos(2)) + sessions = create_mock_nidmm_sessions(3) + session_type.side_effect = sessions + + with reservation.create_nidmm_sessions() as session_info: + assert session_info[0].session == sessions[0] + assert session_info[1].session == sessions[1] + + session_type.assert_any_call( + resource_name="Dev0", reset_device=False, options={}, grpc_options=ANY + ) + session_type.assert_any_call( + resource_name="Dev1", reset_device=False, options={}, grpc_options=ANY + ) + + +def test___optional_args___create_nidmm_session___optional_args_passed( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation(session_management_client, create_nidmm_session_infos(1)) + session = create_mock_nidmm_session() + session_type.side_effect = [session] + + with reservation.create_nidmm_session( + reset_device=True, + options={"simulate": False}, + initialization_behavior=SessionInitializationBehavior.INITIALIZE_SERVER_SESSION, + ): + pass + + session_type.assert_called_once_with( + resource_name="Dev0", + reset_device=True, + options={"simulate": False}, + grpc_options=ANY, + ) + assert ( + session_type.call_args.kwargs["grpc_options"].initialization_behavior + == nidmm.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION + ) + + +def test___simulation_configured___create_nidmm_session___simulation_options_passed( + mocker: MockerFixture, + session_type: Mock, + session_management_client: Mock, +) -> None: + set_nidmm_simulation_options(mocker, True, "PXIe", "4081") + reservation = MultiSessionReservation(session_management_client, create_nidmm_session_infos(1)) + session = create_mock_nidmm_session() + session_type.side_effect = [session] + + with reservation.create_nidmm_session(): + pass + + expected_options = { + "simulate": True, + "driver_setup": {"BoardType": "PXIe", "Model": "4081"}, + } + session_type.assert_called_once_with( + resource_name="Dev0", + reset_device=False, + options=expected_options, + grpc_options=ANY, + ) + + +def test___optional_args_and_simulation_configured___create_nidmm_session___optional_args_passed( + mocker: MockerFixture, + session_type: Mock, + session_management_client: Mock, +) -> None: + set_nidmm_simulation_options(mocker, True, "PXIe", "4081") + reservation = MultiSessionReservation(session_management_client, create_nidmm_session_infos(1)) + session = create_mock_nidmm_session() + session_type.side_effect = [session] + + with reservation.create_nidmm_session(reset_device=True, options={"simulate": False}): + pass + + expected_options = {"simulate": False} + session_type.assert_called_once_with( + resource_name="Dev0", + reset_device=True, + options=expected_options, + grpc_options=ANY, + ) + + +@pytest.fixture +def session_type(mocker: MockerFixture) -> Mock: + """A test fixture that replaces the Session class with a mock.""" + return mocker.patch("nidmm.Session", autospec=True) diff --git a/tests/unit/_drivers/test_nifgen.py b/tests/unit/_drivers/test_nifgen.py new file mode 100644 index 000000000..16a57f7fa --- /dev/null +++ b/tests/unit/_drivers/test_nifgen.py @@ -0,0 +1,140 @@ +import functools +from unittest.mock import ANY, Mock + +import pytest +from pytest_mock import MockerFixture + +from ni_measurementlink_service.session_management import ( + INSTRUMENT_TYPE_NI_FGEN, + MultiSessionReservation, + SessionInitializationBehavior, +) +from tests.unit._drivers._driver_utils import ( + create_mock_session, + create_mock_sessions, + set_simulation_options, +) +from tests.unit._reservation_utils import create_grpc_session_infos + +try: + import nifgen +except ImportError: + nifgen = None + +pytestmark = pytest.mark.skipif(nifgen is None, reason="Requires 'nifgen' package.") + +if nifgen: + # Note: this reads the Session type before it is patched. + create_mock_nifgen_session = functools.partial(create_mock_session, nifgen.Session) + create_mock_nifgen_sessions = functools.partial(create_mock_sessions, nifgen.Session) + create_nifgen_session_infos = functools.partial( + create_grpc_session_infos, INSTRUMENT_TYPE_NI_FGEN + ) + set_nifgen_simulation_options = functools.partial(set_simulation_options, "nifgen") + + +def test___single_session_info___create_nifgen_session___session_created( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation(session_management_client, create_nifgen_session_infos(1)) + session = create_mock_nifgen_session() + session_type.side_effect = [session] + + with reservation.create_nifgen_session() as session_info: + assert session_info.session is session + + session_type.assert_called_once_with( + resource_name="Dev0", reset_device=False, options={}, grpc_options=ANY + ) + + +def test___multiple_session_infos___create_nifgen_sessions___sessions_created( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation(session_management_client, create_nifgen_session_infos(2)) + sessions = create_mock_nifgen_sessions(3) + session_type.side_effect = sessions + + with reservation.create_nifgen_sessions() as session_info: + assert session_info[0].session == sessions[0] + assert session_info[1].session == sessions[1] + + session_type.assert_any_call( + resource_name="Dev0", reset_device=False, options={}, grpc_options=ANY + ) + session_type.assert_any_call( + resource_name="Dev1", reset_device=False, options={}, grpc_options=ANY + ) + + +def test___optional_args___create_nifgen_session___optional_args_passed( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation(session_management_client, create_nifgen_session_infos(1)) + session = create_mock_nifgen_session() + session_type.side_effect = [session] + + with reservation.create_nifgen_session( + reset_device=True, + options={"simulate": False}, + initialization_behavior=SessionInitializationBehavior.INITIALIZE_SERVER_SESSION, + ): + pass + + session_type.assert_called_once_with( + resource_name="Dev0", reset_device=True, options={"simulate": False}, grpc_options=ANY + ) + assert ( + session_type.call_args.kwargs["grpc_options"].initialization_behavior + == nifgen.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION + ) + + +def test___simulation_configured___create_nifgen_session___simulation_options_passed( + mocker: MockerFixture, + session_type: Mock, + session_management_client: Mock, +) -> None: + set_nifgen_simulation_options(mocker, True, "PXIe", "5423 (2CH)") + reservation = MultiSessionReservation(session_management_client, create_nifgen_session_infos(1)) + session = create_mock_nifgen_session() + session_type.side_effect = [session] + + with reservation.create_nifgen_session(): + pass + + expected_options = { + "simulate": True, + "driver_setup": {"BoardType": "PXIe", "Model": "5423 (2CH)"}, + } + session_type.assert_called_once_with( + resource_name="Dev0", reset_device=False, options=expected_options, grpc_options=ANY + ) + + +def test___optional_args_and_simulation_configured___create_nifgen_session___optional_args_passed( + mocker: MockerFixture, + session_type: Mock, + session_management_client: Mock, +) -> None: + set_nifgen_simulation_options(mocker, True, "PXIe", "5423 (2CH)") + reservation = MultiSessionReservation(session_management_client, create_nifgen_session_infos(1)) + session = create_mock_nifgen_session() + session_type.side_effect = [session] + + with reservation.create_nifgen_session(reset_device=True, options={"simulate": False}): + pass + + expected_options = {"simulate": False} + session_type.assert_called_once_with( + resource_name="Dev0", reset_device=True, options=expected_options, grpc_options=ANY + ) + + +@pytest.fixture +def session_type(mocker: MockerFixture) -> Mock: + """A test fixture that replaces the Session class with a mock.""" + return mocker.patch("nifgen.Session", autospec=True) diff --git a/tests/unit/_drivers/test_niscope.py b/tests/unit/_drivers/test_niscope.py new file mode 100644 index 000000000..ee45e4bf5 --- /dev/null +++ b/tests/unit/_drivers/test_niscope.py @@ -0,0 +1,159 @@ +import functools +from unittest.mock import ANY, Mock + +import pytest +from pytest_mock import MockerFixture + +from ni_measurementlink_service.session_management import ( + INSTRUMENT_TYPE_NI_SCOPE, + MultiSessionReservation, + SessionInitializationBehavior, +) +from tests.unit._drivers._driver_utils import ( + create_mock_session, + create_mock_sessions, + set_simulation_options, +) +from tests.unit._reservation_utils import create_grpc_session_infos + +try: + import niscope +except ImportError: + niscope = None + +pytestmark = pytest.mark.skipif(niscope is None, reason="Requires 'niscope' package.") + +if niscope: + # Note: this reads the Session type before it is patched. + create_mock_niscope_session = functools.partial(create_mock_session, niscope.Session) + create_mock_niscope_sessions = functools.partial(create_mock_sessions, niscope.Session) + create_niscope_session_infos = functools.partial( + create_grpc_session_infos, INSTRUMENT_TYPE_NI_SCOPE + ) + set_niscope_simulation_options = functools.partial(set_simulation_options, "niscope") + + +def test___single_session_info___create_niscope_session___session_created( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation( + session_management_client, create_niscope_session_infos(1) + ) + session = create_mock_niscope_session() + session_type.side_effect = [session] + + with reservation.create_niscope_session() as session_info: + assert session_info.session is session + + session_type.assert_called_once_with( + resource_name="Dev0", reset_device=False, options={}, grpc_options=ANY + ) + + +def test___multiple_session_infos___create_niscope_sessions___sessions_created( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation( + session_management_client, create_niscope_session_infos(2) + ) + sessions = create_mock_niscope_sessions(3) + session_type.side_effect = sessions + + with reservation.create_niscope_sessions() as session_info: + assert session_info[0].session == sessions[0] + assert session_info[1].session == sessions[1] + + session_type.assert_any_call( + resource_name="Dev0", reset_device=False, options={}, grpc_options=ANY + ) + session_type.assert_any_call( + resource_name="Dev1", reset_device=False, options={}, grpc_options=ANY + ) + + +def test___optional_args___create_niscope_session___optional_args_passed( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation( + session_management_client, create_niscope_session_infos(1) + ) + session = create_mock_niscope_session() + session_type.side_effect = [session] + + with reservation.create_niscope_session( + reset_device=True, + options={"simulate": False}, + initialization_behavior=SessionInitializationBehavior.INITIALIZE_SERVER_SESSION, + ): + pass + + session_type.assert_called_once_with( + resource_name="Dev0", + reset_device=True, + options={"simulate": False}, + grpc_options=ANY, + ) + assert ( + session_type.call_args.kwargs["grpc_options"].initialization_behavior + == niscope.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION + ) + + +def test___simulation_configured___create_niscope_session___simulation_options_passed( + mocker: MockerFixture, + session_type: Mock, + session_management_client: Mock, +) -> None: + set_niscope_simulation_options(mocker, True, "PXIe", "5162 (4CH)") + reservation = MultiSessionReservation( + session_management_client, create_niscope_session_infos(1) + ) + session = create_mock_niscope_session() + session_type.side_effect = [session] + + with reservation.create_niscope_session(): + pass + + expected_options = { + "simulate": True, + "driver_setup": {"BoardType": "PXIe", "Model": "5162 (4CH)"}, + } + session_type.assert_called_once_with( + resource_name="Dev0", + reset_device=False, + options=expected_options, + grpc_options=ANY, + ) + + +def test___optional_args_and_simulation_configured___create_niscope_session___optional_args_passed( + mocker: MockerFixture, + session_type: Mock, + session_management_client: Mock, +) -> None: + set_niscope_simulation_options(mocker, True, "PXIe", "5162 (4CH)") + reservation = MultiSessionReservation( + session_management_client, create_niscope_session_infos(1) + ) + session = create_mock_niscope_session() + session_type.side_effect = [session] + + with reservation.create_niscope_session(reset_device=True, options={"simulate": False}): + pass + + expected_options = {"simulate": False} + session_type.assert_called_once_with( + resource_name="Dev0", + reset_device=True, + options=expected_options, + grpc_options=ANY, + ) + + +@pytest.fixture +def session_type(mocker: MockerFixture) -> Mock: + """A test fixture that replaces the Session class with a mock.""" + return mocker.patch("niscope.Session", autospec=True) diff --git a/tests/unit/_drivers/test_niswitch.py b/tests/unit/_drivers/test_niswitch.py new file mode 100644 index 000000000..13d297d19 --- /dev/null +++ b/tests/unit/_drivers/test_niswitch.py @@ -0,0 +1,178 @@ +import functools +from unittest.mock import ANY, Mock + +import pytest +from pytest_mock import MockerFixture + +from ni_measurementlink_service._configuration import NISwitchOptions +from ni_measurementlink_service.session_management import ( + INSTRUMENT_TYPE_NI_RELAY_DRIVER, + MultiSessionReservation, + SessionInitializationBehavior, +) +from tests.unit._drivers._driver_utils import ( + create_mock_session, + create_mock_sessions, +) +from tests.unit._reservation_utils import create_grpc_session_infos + +try: + import niswitch +except ImportError: + niswitch = None + +pytestmark = pytest.mark.skipif(niswitch is None, reason="Requires 'niswitch' package.") + +if niswitch: + # Note: this reads the Session type before it is patched. + create_mock_niswitch_session = functools.partial(create_mock_session, niswitch.Session) + create_mock_niswitch_sessions = functools.partial(create_mock_sessions, niswitch.Session) + create_niswitch_session_infos = functools.partial( + create_grpc_session_infos, INSTRUMENT_TYPE_NI_RELAY_DRIVER + ) + + +def test___single_session_info___create_niswitch_session___session_created( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation( + session_management_client, create_niswitch_session_infos(1) + ) + session = create_mock_niswitch_session() + session_type.side_effect = [session] + + with reservation.create_niswitch_session() as session_info: + assert session_info.session is session + + session_type.assert_called_once_with( + resource_name="Dev0", + topology="Configured Topology", + simulate=False, + reset_device=False, + grpc_options=ANY, + ) + + +def test___multiple_session_infos___create_niswitch_sessions___sessions_created( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation( + session_management_client, create_niswitch_session_infos(2) + ) + sessions = create_mock_niswitch_sessions(3) + session_type.side_effect = sessions + + with reservation.create_niswitch_sessions() as session_info: + assert session_info[0].session == sessions[0] + assert session_info[1].session == sessions[1] + + session_type.assert_any_call( + resource_name="Dev0", + topology="Configured Topology", + simulate=False, + reset_device=False, + grpc_options=ANY, + ) + session_type.assert_any_call( + resource_name="Dev1", + topology="Configured Topology", + simulate=False, + reset_device=False, + grpc_options=ANY, + ) + + +def test___optional_args___create_niswitch_session___optional_args_passed( + session_type: Mock, + session_management_client: Mock, +) -> None: + reservation = MultiSessionReservation( + session_management_client, create_niswitch_session_infos(1) + ) + session = create_mock_niswitch_session() + session_type.side_effect = [session] + + with reservation.create_niswitch_session( + topology="2567/Independent", + simulate=True, + reset_device=True, + initialization_behavior=SessionInitializationBehavior.INITIALIZE_SERVER_SESSION, + ): + pass + + session_type.assert_called_once_with( + resource_name="Dev0", + topology="2567/Independent", + simulate=True, + reset_device=True, + grpc_options=ANY, + ) + assert ( + session_type.call_args.kwargs["grpc_options"].initialization_behavior + == niswitch.SessionInitializationBehavior.INITIALIZE_SERVER_SESSION + ) + + +def test___simulation_configured___create_niswitch_session___simulation_options_passed( + mocker: MockerFixture, + session_type: Mock, + session_management_client: Mock, +) -> None: + _set_niswitch_simulation_options(mocker, True, "2567/Independent") + reservation = MultiSessionReservation( + session_management_client, create_niswitch_session_infos(1) + ) + session = create_mock_niswitch_session() + session_type.side_effect = [session] + + with reservation.create_niswitch_session(): + pass + + session_type.assert_called_once_with( + resource_name="Dev0", + topology="2567/Independent", + simulate=True, + reset_device=False, + grpc_options=ANY, + ) + + +def test___optional_args_and_simulation_configured___create_niswitch_session___optional_args_passed( + mocker: MockerFixture, + session_type: Mock, + session_management_client: Mock, +) -> None: + _set_niswitch_simulation_options(mocker, True, "2567/Independent") + reservation = MultiSessionReservation( + session_management_client, create_niswitch_session_infos(1) + ) + session = create_mock_niswitch_session() + session_type.side_effect = [session] + + with reservation.create_niswitch_session( + topology="2529/2-Wire 4x32 Matrix", simulate=False, reset_device=True + ): + pass + + session_type.assert_called_once_with( + resource_name="Dev0", + topology="2529/2-Wire 4x32 Matrix", + simulate=False, + reset_device=True, + grpc_options=ANY, + ) + + +@pytest.fixture +def session_type(mocker: MockerFixture) -> Mock: + """A test fixture that replaces the Session class with a mock.""" + return mocker.patch("niswitch.Session", autospec=True) + + +def _set_niswitch_simulation_options(mocker: MockerFixture, simulate: bool, topology: str) -> None: + mocker.patch( + "ni_measurementlink_service._drivers._niswitch.NISWITCH_OPTIONS", + NISwitchOptions("niswitch", simulate, topology), + ) diff --git a/tests/unit/_reservation_utils.py b/tests/unit/_reservation_utils.py new file mode 100644 index 000000000..5a43efd06 --- /dev/null +++ b/tests/unit/_reservation_utils.py @@ -0,0 +1,22 @@ +"""Reservation-related unit test utilities.""" +from typing import List + +from ni_measurementlink_service._internal.stubs import session_pb2 +from ni_measurementlink_service._internal.stubs.ni.measurementlink.sessionmanagement.v1 import ( + session_management_service_pb2, +) + + +def create_grpc_session_infos( + instrument_type_id: str, + session_count: int, +) -> List[session_management_service_pb2.SessionInformation]: + """Create a list of gRPC SessionInformation messages.""" + return [ + session_management_service_pb2.SessionInformation( + session=session_pb2.Session(name=f"MySession{i}"), + resource_name=f"Dev{i}", + instrument_type_id=instrument_type_id, + ) + for i in range(session_count) + ] diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index d76e35e8c..10b75eff0 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -74,10 +74,16 @@ def multi_session_reservation(mocker: MockerFixture) -> Mock: @pytest.fixture def session_management_client( - mocker: MockerFixture, multi_session_reservation: Mock, single_session_reservation: Mock + discovery_client: Mock, + grpc_channel_pool: Mock, + mocker: MockerFixture, + multi_session_reservation: Mock, + single_session_reservation: Mock, ) -> Mock: """Test fixture that creates a mock SessionManagementClient.""" mock = mocker.create_autospec(SessionManagementClient) + mock._discovery_client = discovery_client + mock._grpc_channel_pool = grpc_channel_pool mock.reserve_session.return_value = single_session_reservation mock.reserve_sessions.return_value = multi_session_reservation mock.reserve_all_registered_sessions.return_value = multi_session_reservation diff --git a/tests/unit/test_reservation.py b/tests/unit/test_reservation.py index c299c5eb3..dd1e8913b 100644 --- a/tests/unit/test_reservation.py +++ b/tests/unit/test_reservation.py @@ -1,25 +1,23 @@ -from typing import List +import functools from unittest.mock import Mock import pytest -from ni_measurementlink_service._internal.stubs import session_pb2 -from ni_measurementlink_service._internal.stubs.ni.measurementlink.sessionmanagement.v1 import ( - session_management_service_pb2, -) from ni_measurementlink_service.session_management import ( MultiSessionReservation, SessionInformation, ) +from tests.unit._reservation_utils import create_grpc_session_infos from tests.utilities import fake_driver +create_nifake_session_infos = functools.partial(create_grpc_session_infos, "nifake") +create_nifoo_session_infos = functools.partial(create_grpc_session_infos, "nifoo") + def test___single_session_info___create_session___session_info_yielded( session_management_client: Mock, ) -> None: - reservation = MultiSessionReservation( - session_management_client, _create_grpc_session_infos(1, "nifake") - ) + reservation = MultiSessionReservation(session_management_client, create_nifake_session_infos(1)) with reservation.create_session(_construct_session, "nifake") as session_info: assert session_info.session_name == "MySession0" @@ -30,9 +28,7 @@ def test___single_session_info___create_session___session_info_yielded( def test___single_session_info___create_session___session_created( session_management_client: Mock, ) -> None: - reservation = MultiSessionReservation( - session_management_client, _create_grpc_session_infos(1, "nifake") - ) + reservation = MultiSessionReservation(session_management_client, create_nifake_session_infos(1)) with reservation.create_session(_construct_session, "nifake") as session_info: assert isinstance(session_info.session, fake_driver.Session) @@ -42,9 +38,7 @@ def test___single_session_info___create_session___session_created( def test___single_session_info___create_session___session_lifetime_tracked( session_management_client: Mock, ) -> None: - reservation = MultiSessionReservation( - session_management_client, _create_grpc_session_infos(1, "nifake") - ) + reservation = MultiSessionReservation(session_management_client, create_nifake_session_infos(1)) with reservation.create_session(_construct_session, "nifake") as session_info: assert reservation._session_cache["MySession0"] is session_info.session @@ -57,9 +51,7 @@ def test___single_session_info___create_session___session_lifetime_tracked( def test___empty_instrument_type_id___create_session___value_error_raised( session_management_client: Mock, ) -> None: - reservation = MultiSessionReservation( - session_management_client, _create_grpc_session_infos(1, "nifake") - ) + reservation = MultiSessionReservation(session_management_client, create_nifake_session_infos(1)) with pytest.raises(ValueError) as exc_info: with reservation.create_session(_construct_session, ""): @@ -71,37 +63,33 @@ def test___empty_instrument_type_id___create_session___value_error_raised( def test___no_session_infos___create_session___value_error_raised( session_management_client: Mock, ) -> None: - reservation = MultiSessionReservation( - session_management_client, _create_grpc_session_infos(0, "nifake") - ) + reservation = MultiSessionReservation(session_management_client, create_nifake_session_infos(0)) with pytest.raises(ValueError) as exc_info: with reservation.create_session(_construct_session, "nifake"): pass - assert "No sessions matched instrument type ID 'nifake'." in exc_info.value.args[0] + assert "No reserved sessions matched instrument type ID 'nifake'." in exc_info.value.args[0] def test___multi_session_infos___create_session___value_error_raised( session_management_client: Mock, ) -> None: - reservation = MultiSessionReservation( - session_management_client, _create_grpc_session_infos(2, "nifake") - ) + reservation = MultiSessionReservation(session_management_client, create_nifake_session_infos(2)) with pytest.raises(ValueError) as exc_info: with reservation.create_session(_construct_session, "nifake"): pass - assert "Too many sessions matched instrument type ID 'nifake'." in exc_info.value.args[0] + assert ( + "Too many reserved sessions matched instrument type ID 'nifake'." in exc_info.value.args[0] + ) def test___session_already_exists___create_session___runtime_error_raised( session_management_client: Mock, ) -> None: - reservation = MultiSessionReservation( - session_management_client, _create_grpc_session_infos(1, "nifake") - ) + reservation = MultiSessionReservation(session_management_client, create_nifake_session_infos(1)) with reservation.create_session(_construct_session, "nifake"): with pytest.raises(RuntimeError) as exc_info: @@ -114,7 +102,7 @@ def test___session_already_exists___create_session___runtime_error_raised( def test___heterogenous_session_infos___create_session___grouped_by_instrument_type( session_management_client: Mock, ) -> None: - grpc_session_infos = _create_grpc_session_infos(2, "nifoo") + grpc_session_infos = create_nifoo_session_infos(2) grpc_session_infos[1].instrument_type_id = "nibar" reservation = MultiSessionReservation(session_management_client, grpc_session_infos) @@ -130,9 +118,7 @@ def test___heterogenous_session_infos___create_session___grouped_by_instrument_t def test___multi_session_infos___create_sessions___session_infos_yielded( session_management_client: Mock, ) -> None: - reservation = MultiSessionReservation( - session_management_client, _create_grpc_session_infos(3, "nifake") - ) + reservation = MultiSessionReservation(session_management_client, create_nifake_session_infos(3)) with reservation.create_sessions(_construct_session, "nifake") as session_infos: assert [info.session_name for info in session_infos] == [ @@ -147,9 +133,7 @@ def test___multi_session_infos___create_sessions___session_infos_yielded( def test___multi_session_infos___create_sessions___sessions_created( session_management_client: Mock, ) -> None: - reservation = MultiSessionReservation( - session_management_client, _create_grpc_session_infos(3, "nifake") - ) + reservation = MultiSessionReservation(session_management_client, create_nifake_session_infos(3)) with reservation.create_sessions(_construct_session, "nifake") as session_infos: assert all([isinstance(info.session, fake_driver.Session) for info in session_infos]) @@ -159,9 +143,7 @@ def test___multi_session_infos___create_sessions___sessions_created( def test___multi_session_infos___create_sessions___session_lifetime_tracked( session_management_client: Mock, ) -> None: - reservation = MultiSessionReservation( - session_management_client, _create_grpc_session_infos(3, "nifake") - ) + reservation = MultiSessionReservation(session_management_client, create_nifake_session_infos(3)) with reservation.create_sessions(_construct_session, "nifake") as session_infos: assert reservation._session_cache["MySession0"] is session_infos[0].session @@ -176,9 +158,7 @@ def test___multi_session_infos___create_sessions___session_lifetime_tracked( def test___empty_instrument_type_id___create_sessions___value_error_raised( session_management_client: Mock, ) -> None: - reservation = MultiSessionReservation( - session_management_client, _create_grpc_session_infos(3, "nifake") - ) + reservation = MultiSessionReservation(session_management_client, create_nifake_session_infos(3)) with pytest.raises(ValueError) as exc_info: with reservation.create_sessions(_construct_session, ""): @@ -190,23 +170,19 @@ def test___empty_instrument_type_id___create_sessions___value_error_raised( def test___no_session_infos___create_sessions___value_error_raised( session_management_client: Mock, ) -> None: - reservation = MultiSessionReservation( - session_management_client, _create_grpc_session_infos(0, "nifake") - ) + reservation = MultiSessionReservation(session_management_client, []) with pytest.raises(ValueError) as exc_info: with reservation.create_sessions(_construct_session, "nifake"): pass - assert "No sessions matched instrument type ID 'nifake'." in exc_info.value.args[0] + assert "No reserved sessions matched instrument type ID 'nifake'." in exc_info.value.args[0] def test___session_already_exists___create_sessions___runtime_error_raised( session_management_client: Mock, ) -> None: - reservation = MultiSessionReservation( - session_management_client, _create_grpc_session_infos(3, "nifake") - ) + reservation = MultiSessionReservation(session_management_client, create_nifake_session_infos(3)) with reservation.create_sessions(_construct_session, "nifake"): with pytest.raises(RuntimeError) as exc_info: @@ -219,7 +195,7 @@ def test___session_already_exists___create_sessions___runtime_error_raised( def test___heterogenous_session_infos___create_sessions___grouped_by_instrument_type( session_management_client: Mock, ) -> None: - grpc_session_infos = _create_grpc_session_infos(3, "nifoo") + grpc_session_infos = create_nifoo_session_infos(3) grpc_session_infos[1].instrument_type_id = "nibar" reservation = MultiSessionReservation(session_management_client, grpc_session_infos) @@ -234,17 +210,3 @@ def test___heterogenous_session_infos___create_sessions___grouped_by_instrument_ def _construct_session(session_info: SessionInformation) -> fake_driver.Session: return fake_driver.Session(session_info.resource_name) - - -def _create_grpc_session_infos( - session_count: int, - instrument_type_id: str, -) -> List[session_management_service_pb2.SessionInformation]: - return [ - session_management_service_pb2.SessionInformation( - session=session_pb2.Session(name=f"MySession{i}"), - resource_name=f"Dev{i}", - instrument_type_id=instrument_type_id, - ) - for i in range(session_count) - ]