tests: Add logging acceptance tests (#373)

* service: Rename SessionManagementClient

* tests: Add logging acceptance tests

* tests: Fix name conflict between the `request` fixture and a gRPC request variable

Author: bkeryan

ni_measurementlink_service/session_management.py

@@ -5,7 +5,17 @@
 import warnings
 from functools import cached_property
 from types import TracebackType
-from typing import Any, Iterable, List, Literal, NamedTuple, Optional, Sequence, Type, TypeVar
+from typing import (
+    Any,
+    Iterable,
+    List,
+    Literal,
+    NamedTuple,
+    Optional,
+    Sequence,
+    Type,
+    TypeVar,
+)
 
 import grpc
 from deprecation import DeprecatedWarning
@@ -235,7 +245,7 @@ def __getattr__(name: str) -> Any:
         raise AttributeError(f"module {__name__} has no attribute {name}")
 
 
-class Client(object):
+class SessionManagementClient(object):
     """Class that manages driver sessions."""
 
     def __init__(self, *, grpc_channel: grpc.Channel) -> None:
@@ -505,3 +515,7 @@ def reserve_all_registered_sessions(
             self._client.ReserveAllRegisteredSessions(request)
         )
         return MultiSessionReservation(session_manager=self, session_info=response.sessions)
+
+
+Client = SessionManagementClient
+"""Alias for compatibility with code that uses session_management.Client."""

pyproject.toml

@@ -72,6 +72,9 @@ build-backend = "poetry.core.masonry.api"
 addopts = "--doctest-modules --strict-markers"
 filterwarnings = ["always::ImportWarning", "always::ResourceWarning"]
 testpaths = ["tests"]
+markers = [
+  "service_class: specifies which test service to use.",
+]
 
 [tool.mypy]
 warn_unused_configs = true

tests/acceptance/test_logging.py

@@ -0,0 +1,135 @@
+import logging
+import re
+from typing import Generator
+
+import pytest
+from pytest import FixtureRequest, LogCaptureFixture
+
+from ni_measurementlink_service import session_management
+from ni_measurementlink_service._internal.discovery_client import DiscoveryClient
+from ni_measurementlink_service._internal.stubs.ni.measurementlink.measurement.v2 import (
+    measurement_service_pb2 as v2_measurement_service_pb2,
+    measurement_service_pb2_grpc as v2_measurement_service_pb2_grpc,
+)
+from ni_measurementlink_service.measurement.service import MeasurementService
+from ni_measurementlink_service.session_management import SessionManagementClient
+from tests.acceptance.test_streaming_data_measurement import (
+    _get_configuration_parameters as get_streaming_data_configuration_parameters,
+)
+from tests.utilities import loopback_measurement, streaming_data_measurement
+from tests.utilities.discovery_service_process import DiscoveryServiceProcess
+
+
+def test___discovery_client___call___client_call_logged(
+    caplog: LogCaptureFixture, discovery_client: DiscoveryClient
+) -> None:
+    with caplog.at_level(logging.DEBUG):
+        _ = discovery_client.resolve_service(
+            session_management.GRPC_SERVICE_INTERFACE_NAME, session_management.GRPC_SERVICE_CLASS
+        )
+
+    method_name = "/ni.measurementlink.discovery.v1.DiscoveryService/ResolveService"
+    debug_messages = [r.message for r in caplog.records if r.levelno == logging.DEBUG]
+    assert f"gRPC client call starting: {method_name}" in debug_messages
+    assert f"gRPC client call complete: {method_name}" in debug_messages
+
+
+def test___session_management_client___call___client_call_logged(
+    caplog: LogCaptureFixture, request: FixtureRequest
+) -> None:
+    with caplog.at_level(logging.DEBUG):
+        # HACK: set log level before constructing client
+        session_management_client: SessionManagementClient = request.getfixturevalue(
+            "session_management_client"
+        )
+
+        session_management_client.unregister_sessions(session_info=[])
+
+    method_name = (
+        "/ni.measurementlink.sessionmanagement.v1.SessionManagementService/UnregisterSessions"
+    )
+    debug_messages = [r.message for r in caplog.records if r.levelno == logging.DEBUG]
+    assert f"gRPC client call starting: {method_name}" in debug_messages
+    assert f"gRPC client call complete: {method_name}" in debug_messages
+
+
+@pytest.mark.service_class("ni.tests.LoopbackMeasurement_Python")
+def test___loopback_measurement___get_metadata___server_call_logged(
+    caplog: LogCaptureFixture, request: FixtureRequest
+) -> None:
+    with caplog.at_level(logging.DEBUG):
+        # HACK: set log level before constructing server
+        stub_v2: v2_measurement_service_pb2_grpc.MeasurementServiceStub = request.getfixturevalue(
+            "stub_v2"
+        )
+
+        _ = stub_v2.GetMetadata(v2_measurement_service_pb2.GetMetadataRequest())
+
+    method_name = "/ni.measurementlink.measurement.v2.MeasurementService/GetMetadata"
+    debug_messages = [r.message for r in caplog.records if r.levelno == logging.DEBUG]
+    assert f"gRPC server call starting: {method_name}" in debug_messages
+    assert f"gRPC server call complete: {method_name}" in debug_messages
+    info_messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
+    info_regex = re.compile(rf"gRPC server call {method_name} responded OK in [\d.]+ ms")
+    assert len(list(filter(info_regex.match, info_messages))) == 1
+
+
+@pytest.mark.service_class("ni.tests.StreamingDataMeasurement_Python")
+def test___streaming_data_measurement___measure___server_call_logged(
+    caplog: LogCaptureFixture, request: FixtureRequest
+) -> None:
+    with caplog.at_level(logging.DEBUG):
+        # HACK: set log level before constructing server
+        stub_v2: v2_measurement_service_pb2_grpc.MeasurementServiceStub = request.getfixturevalue(
+            "stub_v2"
+        )
+        num_responses = 10
+
+        measure_request = v2_measurement_service_pb2.MeasureRequest(
+            configuration_parameters=get_streaming_data_configuration_parameters(
+                num_responses=num_responses
+            )
+        )
+        response_iterator = stub_v2.Measure(measure_request)
+        for response in response_iterator:
+            pass
+
+    method_name = "/ni.measurementlink.measurement.v2.MeasurementService/Measure"
+    debug_messages = [r.message for r in caplog.records if r.levelno == logging.DEBUG]
+    assert f"gRPC server call starting: {method_name}" in debug_messages
+    assert f"gRPC server call complete: {method_name}" in debug_messages
+    debug_response_regex = re.compile(rf"gRPC server call streaming response: {method_name}")
+    assert len(list(filter(debug_response_regex.match, debug_messages))) == num_responses
+    info_messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
+    info_regex = re.compile(rf"gRPC server call {method_name} responded OK in [\d.]+ ms")
+    assert len(list(filter(info_regex.match, info_messages))) == 1
+
+
+@pytest.fixture
+def measurement_service(request: FixtureRequest) -> MeasurementService:
+    service_class_marker = request.node.get_closest_marker("service_class")
+    service_class = service_class_marker.args[0]
+    if service_class == "ni.tests.LoopbackMeasurement_Python":
+        return request.getfixturevalue("loopback_measurement_service")
+    elif service_class == "ni.tests.StreamingDataMeasurement_Python":
+        return request.getfixturevalue("streaming_data_measurement_service")
+    else:
+        raise ValueError(f"Unsupported service class: {service_class}")
+
+
+@pytest.fixture
+def loopback_measurement_service(
+    discovery_service_process: DiscoveryServiceProcess,
+) -> Generator[MeasurementService, None, None]:
+    """Test fixture that creates a loopback measurement."""
+    with loopback_measurement.measurement_service.host_service() as service:
+        yield service
+
+
+@pytest.fixture
+def streaming_data_measurement_service(
+    discovery_service_process: DiscoveryServiceProcess,
+) -> Generator[MeasurementService, None, None]:
+    """Test fixture that creates a loopback measurement."""
+    with streaming_data_measurement.measurement_service.host_service() as service:
+        yield service

tests/conftest.py

@@ -6,14 +6,22 @@
 import grpc
 import pytest
 
-from ni_measurementlink_service._internal.discovery_client import _get_registration_json_file_path
+from ni_measurementlink_service import session_management
+from ni_measurementlink_service._internal.discovery_client import (
+    DiscoveryClient,
+    _get_registration_json_file_path,
+)
 from ni_measurementlink_service._internal.stubs.ni.measurementlink.measurement.v1 import (
     measurement_service_pb2_grpc as v1_measurement_service_pb2_grpc,
 )
 from ni_measurementlink_service._internal.stubs.ni.measurementlink.measurement.v2 import (
     measurement_service_pb2_grpc as v2_measurement_service_pb2_grpc,
 )
-from ni_measurementlink_service.measurement.service import MeasurementService
+from ni_measurementlink_service.measurement.service import (
+    GrpcChannelPool,
+    MeasurementService,
+)
+from ni_measurementlink_service.session_management import SessionManagementClient
 from tests.utilities.discovery_service_process import DiscoveryServiceProcess
 
 
@@ -63,3 +71,30 @@ def discovery_service_process() -> Generator[DiscoveryServiceProcess, None, None
     else:
         with DiscoveryServiceProcess() as proc:
             yield proc
+
+
+@pytest.fixture(scope="session")
+def grpc_channel_pool() -> Generator[GrpcChannelPool, None, None]:
+    """Test fixture that creates a gRPC channel pool."""
+    with GrpcChannelPool() as grpc_channel_pool:
+        yield grpc_channel_pool
+
+
+@pytest.fixture
+def discovery_client(discovery_service_process: DiscoveryServiceProcess) -> DiscoveryClient:
+    """Test fixture that creates a discovery client."""
+    # TODO: DiscoveryClient should accept a GrpcChannelPool
+    return DiscoveryClient()
+
+
+@pytest.fixture
+def session_management_client(
+    discovery_client: DiscoveryClient, grpc_channel_pool: GrpcChannelPool
+) -> SessionManagementClient:
+    """Test fixture that creates a session management client."""
+    # TODO: SessionManagementClient should accept a GrpcChannelPool
+    location = discovery_client.resolve_service(
+        session_management.GRPC_SERVICE_INTERFACE_NAME, session_management.GRPC_SERVICE_CLASS
+    )
+    grpc_channel = grpc_channel_pool.get_channel(location.insecure_address)
+    return SessionManagementClient(grpc_channel=grpc_channel)