From a9ce32b5c5a140d4fa1e9f43cba7996d6d193943 Mon Sep 17 00:00:00 2001 From: Miruna Serian Date: Wed, 22 Jul 2026 18:08:12 +0100 Subject: [PATCH 1/9] Implement ClientInterceptor in SessionManager --- .../session_manager/session_manager_driver.py | 51 ++------- src/drunc/utils/grpc_utils.py | 59 ++++++++++ tests/session_manager/conftest.py | 80 +++++++++---- .../test_session_manager_driver.py | 97 ++++++---------- .../test_session_manager_rich_errors.py | 107 ++++++++++-------- 5 files changed, 223 insertions(+), 171 deletions(-) diff --git a/src/drunc/session_manager/session_manager_driver.py b/src/drunc/session_manager/session_manager_driver.py index 2a9ec7594..999bb0696 100644 --- a/src/drunc/session_manager/session_manager_driver.py +++ b/src/drunc/session_manager/session_manager_driver.py @@ -8,9 +8,8 @@ from druncschema.token_pb2 import Token from drunc.utils.grpc_utils import ( + RichErrorClientInterceptor, copy_token, - extract_grpc_rich_error, - handle_grpc_error, ) from drunc.utils.utils import get_logger @@ -35,7 +34,9 @@ def __init__(self, address: str, token: Token): options = [ ("grpc.keepalive_time_ms", 60000) # pings the server every 60 seconds ] - self.channel = grpc.insecure_channel(self.address, options=options) + raw_channel = grpc.insecure_channel(self.address, options=options) + rich_interceptor = RichErrorClientInterceptor(logger=self.log) + self.channel = grpc.intercept_channel(raw_channel, rich_interceptor) self.stub = SessionManagerStub(self.channel) self.token = copy_token(token) self.log = get_logger("session_manager_driver", rich_handler=True) @@ -51,19 +52,7 @@ def describe(self, timeout: int | float = 60) -> Description: """ request = Request(token=copy_token(self.token)) - try: - response: Description = self.stub.describe(request, timeout=timeout) - except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - self.log.error(error_details) - except Exception as extraction_error: - self.log.debug( - f"Could not extract rich error details from gRPC error: {extraction_error}", - exc_info=True, - ) - - handle_grpc_error(e) + response: Description = self.stub.describe(request, timeout=timeout) return response @@ -78,19 +67,9 @@ def list_all_sessions(self, timeout: int | float = 60) -> AllActiveSessions: """ request = Request(token=copy_token(self.token)) - try: - response: AllActiveSessions = self.stub.list_all_sessions(request, timeout=timeout) - except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - self.log.error(error_details) - except Exception as extraction_error: - self.log.debug( - f"Could not extract rich error details from gRPC error: {extraction_error}", - exc_info=True, - ) - - handle_grpc_error(e) + response: AllActiveSessions = self.stub.list_all_sessions( + request, timeout=timeout + ) return response @@ -105,18 +84,6 @@ def list_all_configs(self, timeout: int | float = 60) -> AllConfigKeys: """ request = Request(token=copy_token(self.token)) - try: - response: AllConfigKeys = self.stub.list_all_configs(request, timeout=timeout) - except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - self.log.error(error_details) - except Exception as extraction_error: - self.log.debug( - f"Could not extract rich error details from gRPC error: {extraction_error}", - exc_info=True, - ) - - handle_grpc_error(e) + response: AllConfigKeys = self.stub.list_all_configs(request, timeout=timeout) return response diff --git a/src/drunc/utils/grpc_utils.py b/src/drunc/utils/grpc_utils.py index a228de8a7..c39fd38d7 100644 --- a/src/drunc/utils/grpc_utils.py +++ b/src/drunc/utils/grpc_utils.py @@ -521,3 +521,62 @@ def error_wrapper(request: object, context: grpc.ServicerContext) -> object: response_serializer=handler.response_serializer, ) return handler + + +class _CallWrapper: + """Wraps the gRPC Call/Future object to catch exceptions.""" + + def __init__(self, call, method, logger): + self._call = call + self._method = method + self._logger = logger + + def _handle_error(self, exception): + if isinstance(exception, grpc.RpcError): + self._logger.error(f"gRPC Call Failed on method: {self._method}") + try: + error_details = extract_grpc_rich_error(exception) + self._logger.error(error_details) + except Exception as extraction_error: + self._logger.debug( + f"Could not extract rich error details: {extraction_error}", + exc_info=True, + ) + handle_grpc_error(exception) + + def __getattr__(self, attr): + # intercept the .result() call and process the error + # to be used in tests if a mock passes an exception directly + if attr == "result" and isinstance(self._call, Exception): + + def handle_mocked_error(*args, **kwargs): + self._handle_error(self._call) + raise self._call + + return handle_mocked_error + + # Wrap the Future's methods to catch the error when it resolves. + val = getattr(self._call, attr) + if attr in ("result", "exception"): + + def wrapper(*args, **kwargs): + try: + res = val(*args, **kwargs) + if attr == "exception" and res is not None: + self._handle_error(res) + return res + except Exception as e: + self._handle_error(e) + raise + + return wrapper + return val + + +class RichErrorClientInterceptor(grpc.UnaryUnaryClientInterceptor): + def __init__(self, logger): + self.log = logger + + def intercept_unary_unary(self, continuation, client_call_details, request): + response_call = continuation(client_call_details, request) + return _CallWrapper(response_call, client_call_details.method, self.log) diff --git a/tests/session_manager/conftest.py b/tests/session_manager/conftest.py index a4f18185d..5f4152d63 100644 --- a/tests/session_manager/conftest.py +++ b/tests/session_manager/conftest.py @@ -2,6 +2,7 @@ from unittest.mock import MagicMock, patch import google.protobuf.any_pb2 +import grpc import grpc_testing import pytest from druncschema.description_pb2 import CommandDescription, Description @@ -148,33 +149,70 @@ def mock_logger_driver(): yield mock_logger_instance +class MockGrpcCall: + """A mock of the gRPC Future (Call) object.""" + + def __init__(self, response=None): + self._response = response + + def result(self): + return self._response + + +class FakeMultiCallable: + """Simulates a gRPC method endpoint.""" + + def __init__(self, channel): + self.channel = channel + + def with_call(self, request, *args, **kwargs): + if self.channel.error: + raise self.channel.error + + call = MockGrpcCall(response=self.channel.response) + return (self.channel.response, call) + + +class FakeChannel(grpc.Channel): + """A fake gRPC channel.""" + + def __init__(self): + self.response = None + self.error = None + + def unary_unary(self, method, *args, **kwargs): + return FakeMultiCallable(self) + + def unary_stream(self, *args, **kwargs): + pass + + def stream_unary(self, *args, **kwargs): + pass + + def stream_stream(self, *args, **kwargs): + pass + + def close(self): + pass + + def subscribe(self, *args, **kwargs): + pass + + def unsubscribe(self, *args, **kwargs): + pass + + @pytest.fixture(scope="function") def mock_driver(mock_logger_driver): - """ - This fixture creates a driver instance where the underlying gRPC channel - and stub are mocked. + fake_channel = FakeChannel() - Returns: - SessionManagerDriver: Driver instance with mocked dependencies - """ - with ( - patch("drunc.session_manager.session_manager_driver.grpc.insecure_channel"), - patch( - "drunc.session_manager.session_manager_driver.SessionManagerStub" - ) as mock_stub_class, + with patch( + "drunc.session_manager.session_manager_driver.grpc.insecure_channel", + return_value=fake_channel, ): - # Create mock stub instance that will be returned by SessionManagerStub() - mock_stub = MagicMock() - mock_stub_class.return_value = mock_stub - - # Initialize driver with mocked dependencies driver = SessionManagerDriver(address="localhost:50051", token=Token()) - - # Attach mock stub for easy access in tests - driver._mock_stub = mock_stub - driver.log = mock_logger_driver - + driver._fake_channel = fake_channel return driver diff --git a/tests/session_manager/test_session_manager_driver.py b/tests/session_manager/test_session_manager_driver.py index 4e4a5bbd2..d34195cc1 100644 --- a/tests/session_manager/test_session_manager_driver.py +++ b/tests/session_manager/test_session_manager_driver.py @@ -1,9 +1,11 @@ """ This module tests that the SessionManagerDriver correctly invokes the underlying gRPC stub methods and properly handles gRPC exceptions. + +They use a real stub and interceptor and a mocked channel. """ -from unittest.mock import MagicMock, patch +from unittest.mock import patch import grpc import pytest @@ -12,49 +14,15 @@ @pytest.mark.parametrize( - "method_name, expected_response", - [ - ("describe", "describe_response"), - ("list_all_sessions", "all_active_sessions_response"), - ("list_all_configs", "all_config_keys_response"), - ], - indirect=[ - "expected_response" - ], # Tells pytest to treat 'expected_response' as fixture names -) -def test_grpc_success(mock_driver, method_name, expected_response): - """ - Test that the methods correctly call the stub and return response. - """ - # Configure mock stub to return expected response - getattr(mock_driver._mock_stub, method_name).return_value = expected_response - - # Call the method under test - response = getattr(mock_driver, method_name)() - - getattr(mock_driver._mock_stub, method_name).assert_called_once() - - call_args = getattr(mock_driver._mock_stub, method_name).call_args - request = call_args[0][0] - - assert hasattr(request, "token") - assert response == expected_response - - -@pytest.mark.parametrize( - "method_name", - [ - "describe", - "list_all_sessions", - "list_all_configs", - ], + "method_name", ["describe", "list_all_sessions", "list_all_configs"] ) def test_grpc_error_handling(mock_driver, method_name): """ Test that gRPC errors are handled and logged. """ + grpc_error = grpc.RpcError("Simulated gRPC failure") - getattr(mock_driver.stub, method_name).side_effect = grpc_error + mock_driver._fake_channel.error = grpc_error error_details = GrpcErrorDetails( code="INVALID_ARGUMENT", @@ -63,62 +31,69 @@ def test_grpc_error_handling(mock_driver, method_name): ) with ( - patch( - "drunc.session_manager.session_manager_driver.extract_grpc_rich_error" - ) as mock_extract, - patch( - "drunc.session_manager.session_manager_driver.handle_grpc_error" - ) as mock_handler, - patch("grpc_status.rpc_status.from_call", return_value=MagicMock()), + patch("drunc.utils.grpc_utils.extract_grpc_rich_error") as mock_extract, + patch("drunc.utils.grpc_utils.handle_grpc_error") as mock_handler, ): mock_extract.return_value = error_details mock_handler.side_effect = grpc_error + # Execute driver method with pytest.raises(grpc.RpcError): getattr(mock_driver, method_name)() + # Assert the interceptor successfully caught the error mock_extract.assert_called_once_with(grpc_error) - mock_driver.log.error.assert_called_once_with(error_details) mock_handler.assert_called_once_with(grpc_error) - logged = mock_driver.log.error.call_args[0][0] + assert mock_driver.log.error.call_count == 2 + logged = mock_driver.log.error.call_args_list[1][0][0] assert logged == error_details @pytest.mark.parametrize( - "method_name", + "method_name, expected_response", [ - "describe", - "list_all_sessions", - "list_all_configs", + ("describe", "describe_response"), + ("list_all_sessions", "all_active_sessions_response"), + ("list_all_configs", "all_config_keys_response"), ], + indirect=["expected_response"], +) +def test_grpc_success(mock_driver, method_name, expected_response): + mock_driver._fake_channel.response = expected_response + + response = getattr(mock_driver, method_name)() + + assert response == expected_response + + +@pytest.mark.parametrize( + "method_name", ["describe", "list_all_sessions", "list_all_configs"] ) def test_grpc_error_fallback(mock_driver, method_name): """ Test that the client correctly handles a gRPC error when no rich error details are available. """ grpc_error = grpc.RpcError("Basic gRPC error") - getattr(mock_driver.stub, method_name).side_effect = grpc_error + mock_driver._fake_channel.error = grpc_error + error_details = GrpcErrorDetails( code="UNKNOWN", message="Basic gRPC error", details=[] ) with ( - patch("grpc_status.rpc_status.from_call", return_value=None), - patch( - "drunc.session_manager.session_manager_driver.extract_grpc_rich_error" - ) as mock_extract, - patch( - "drunc.session_manager.session_manager_driver.handle_grpc_error" - ) as mock_handler, + patch("drunc.utils.grpc_utils.extract_grpc_rich_error") as mock_extract, + patch("drunc.utils.grpc_utils.handle_grpc_error") as mock_handler, ): mock_extract.return_value = error_details mock_handler.side_effect = grpc_error + # Execute driver method with pytest.raises(grpc.RpcError): getattr(mock_driver, method_name)() + # Assert fallback logic mock_extract.assert_called_once() - mock_driver.log.error.assert_called_once() - logged = mock_driver.log.error.call_args[0][0] + assert mock_driver.log.error.call_count == 2 + logged = mock_driver.log.error.call_args_list[1][0][0] assert logged == error_details diff --git a/tests/session_manager/test_session_manager_rich_errors.py b/tests/session_manager/test_session_manager_rich_errors.py index cd0db4b6b..03b7d2096 100644 --- a/tests/session_manager/test_session_manager_rich_errors.py +++ b/tests/session_manager/test_session_manager_rich_errors.py @@ -8,10 +8,12 @@ SessionManagerStub, add_SessionManagerServicer_to_server, ) -from google.rpc import error_details_pb2, status_pb2 from drunc.session_manager.session_manager import SessionManager -from drunc.utils.grpc_utils import RichErrorServerInterceptor +from drunc.utils.grpc_utils import ( + RichErrorClientInterceptor, + RichErrorServerInterceptor, +) class SessionManagerRichErrorTestSuite: @@ -49,8 +51,12 @@ def setup_server_and_client(self): self.server.add_insecure_port(listen_addr) self.server.start() + self.mock_client_logger = MagicMock() + client_interceptor = RichErrorClientInterceptor(logger=self.mock_client_logger) + # Create client channel and stub - self.channel = grpc.insecure_channel(self.server_address) + raw_channel = grpc.insecure_channel(self.server_address) + self.channel = grpc.intercept_channel(raw_channel, client_interceptor) self.stub = SessionManagerStub(self.channel) def teardown_server_and_client(self): @@ -82,6 +88,7 @@ def test_list_all_configs_no_config_files_rich_error( session_manager_rich_error_test_suite.setup_server_and_client() stub = session_manager_rich_error_test_suite.stub + mock_logger = session_manager_rich_error_test_suite.mock_client_logger # Remove the DUNEDAQ_DB_PATH from the environment to simulate it's not set monkeypatch.delenv("DUNEDAQ_DB_PATH", raising=False) @@ -94,19 +101,21 @@ def test_list_all_configs_no_config_files_rich_error( assert err.code() == grpc.StatusCode.FAILED_PRECONDITION assert "DUNEDAQ_DB_PATH" in err.details() - # Unpack rich error metadata - status = status_pb2.Status() - for key, value in err.trailing_metadata(): - if key == "grpc-status-details-bin": - status.ParseFromString(value) + # The interceptor calls log.error twice (once for the method, once for the details) + assert mock_logger.error.call_count == 2 + + # Extract the GrpcErrorDetails object that the interceptor logged + logged_error_details = mock_logger.error.call_args_list[1][0][0] - # There should be a PreconditionFailure detail - precond = error_details_pb2.PreconditionFailure() - status.details[0].Unpack(precond) + assert logged_error_details.code == "FAILED_PRECONDITION" - violation = precond.violations[0] - assert violation.type == "MISSING OR INVALID" - assert "DUNEDAQ_DB_PATH env variable not set" in violation.description + # Access the PreconditionFailure detail object directly from the list + precond_detail = logged_error_details.details[0] + assert precond_detail.violations[0].type == "MISSING OR INVALID" + assert ( + "DUNEDAQ_DB_PATH env variable not set" + in precond_detail.violations[0].description + ) def test_no_config_files_rich_error( @@ -114,6 +123,7 @@ def test_no_config_files_rich_error( ): session_manager_rich_error_test_suite.setup_server_and_client() stub = session_manager_rich_error_test_suite.stub + mock_logger = session_manager_rich_error_test_suite.mock_client_logger monkeypatch.setenv("DUNEDAQ_DB_PATH", "/fake_path") @@ -125,18 +135,21 @@ def test_no_config_files_rich_error( assert err.code() == grpc.StatusCode.FAILED_PRECONDITION assert "Config files" in err.details() - # Unpack rich error metadata - status = status_pb2.Status() - for key, value in err.trailing_metadata(): - if key == "grpc-status-details-bin": - status.ParseFromString(value) + # The interceptor calls log.error twice (once for the method, once for the details) + assert mock_logger.error.call_count == 2 - precond = error_details_pb2.PreconditionFailure() - status.details[0].Unpack(precond) + # Extract the GrpcErrorDetails object that the interceptor logged + logged_error_details = mock_logger.error.call_args_list[1][0][0] - violation = precond.violations[0] - assert violation.type == "MISSING OR INVALID" - assert "No configuration files found in /fake_path" in violation.description + assert logged_error_details.code == "FAILED_PRECONDITION" + + # Access the PreconditionFailure detail object directly from the list + precond_detail = logged_error_details.details[0] + assert precond_detail.violations[0].type == "MISSING OR INVALID" + assert ( + "No configuration files found in /fake_path" + in precond_detail.violations[0].description + ) def test_config_parse_failure( @@ -144,6 +157,7 @@ def test_config_parse_failure( ): session_manager_rich_error_test_suite.setup_server_and_client() stub = session_manager_rich_error_test_suite.stub + mock_logger = session_manager_rich_error_test_suite.mock_client_logger # Set env var so search_paths is non-empty monkeypatch.setenv("DUNEDAQ_DB_PATH", "valid_path/") @@ -163,18 +177,19 @@ def test_config_parse_failure( err = excinfo.value assert err.code() == grpc.StatusCode.FAILED_PRECONDITION - # Unpack rich error metadata - status = status_pb2.Status() - for key, value in err.trailing_metadata(): - if key == "grpc-status-details-bin": - status.ParseFromString(value) - precond = error_details_pb2.PreconditionFailure() - status.details[0].Unpack(precond) + # The interceptor calls log.error twice (once for the method, once for the details) + assert mock_logger.error.call_count == 2 + + # Extract the GrpcErrorDetails object that the interceptor logged + logged_error_details = mock_logger.error.call_args_list[1][0][0] - violation = precond.violations[0] - assert violation.type == "MISSING OR INVALID" - assert "Config files" in violation.subject - assert "Failed to parse configuration file" in violation.description + # Access the PreconditionFailure detail object directly from the list + precond_detail = logged_error_details.details[0] + assert precond_detail.violations[0].type == "MISSING OR INVALID" + assert "Config files" in precond_detail.violations[0].subject + assert ( + "Failed to parse configuration file" in precond_detail.violations[0].description + ) def test_dals_missing_or_invalid( @@ -182,6 +197,7 @@ def test_dals_missing_or_invalid( ): session_manager_rich_error_test_suite.setup_server_and_client() stub = session_manager_rich_error_test_suite.stub + mock_logger = session_manager_rich_error_test_suite.mock_client_logger # Set env var so search_paths is non-empty monkeypatch.setenv("DUNEDAQ_DB_PATH", "valid_path/") @@ -201,16 +217,13 @@ def test_dals_missing_or_invalid( err = excinfo.value assert err.code() == grpc.StatusCode.FAILED_PRECONDITION + assert mock_logger.error.call_count == 2 + + # Extract the GrpcErrorDetails object that the interceptor logged + logged_error_details = mock_logger.error.call_args_list[1][0][0] - # Unpack rich error metadata - status = status_pb2.Status() - for key, value in err.trailing_metadata(): - if key == "grpc-status-details-bin": - status.ParseFromString(value) - precond = error_details_pb2.PreconditionFailure() - status.details[0].Unpack(precond) - - violation = precond.violations[0] - assert violation.type == "MISSING OR INVALID" - assert "Session DALs" in violation.subject - assert "DALs missing or invalid" in violation.description + # Access the PreconditionFailure detail object directly from the list + precond_detail = logged_error_details.details[0] + assert precond_detail.violations[0].type == "MISSING OR INVALID" + assert "Session DALs" in precond_detail.violations[0].subject + assert "DALs missing or invalid" in precond_detail.violations[0].description From 8f19f7cf6bbc18e0a750153a1e757ccd84c77a6a Mon Sep 17 00:00:00 2001 From: Miruna Serian Date: Thu, 23 Jul 2026 17:25:06 +0100 Subject: [PATCH 2/9] Implement client interceptor in process manager --- .../process_manager/process_manager_driver.py | 178 +++---------- src/drunc/utils/grpc_utils.py | 4 + .../test_process_manager_driver.py | 246 +++++------------- 3 files changed, 105 insertions(+), 323 deletions(-) diff --git a/src/drunc/process_manager/process_manager_driver.py b/src/drunc/process_manager/process_manager_driver.py index 3bae291e3..1fb9a9f31 100644 --- a/src/drunc/process_manager/process_manager_driver.py +++ b/src/drunc/process_manager/process_manager_driver.py @@ -37,9 +37,8 @@ from drunc.process_manager.oks_parser import get_full_db_path from drunc.process_manager.utils import format_hostname, get_log_path, get_rte_script from drunc.utils.grpc_utils import ( + RichErrorClientInterceptor, copy_token, - extract_grpc_rich_error, - handle_grpc_error, ) from drunc.utils.utils import ( file_is_read_only, @@ -63,7 +62,9 @@ def __init__(self, address: str, token: Token): options = [ ("grpc.keepalive_time_ms", 60000) # pings the server every 60 seconds ] - self.channel = grpc.insecure_channel(self.address, options=options) + raw_channel = grpc.insecure_channel(self.address, options=options) + rich_interceptor = RichErrorClientInterceptor(logger=self.log) + self.channel = grpc.intercept_channel(raw_channel, rich_interceptor) self.stub = ProcessManagerStub(self.channel) self.token = copy_token(token) @@ -98,19 +99,7 @@ def send_msg(self, msg): timeout = 10 - try: - response = self.stub.send_msg(request, timeout=timeout) - except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - self.log.error(error_details) - except Exception as extraction_error: - self.log.critical( - f"Could not extract rich error details from gRPC error: {extraction_error}", - exc_info=True, - ) - handle_grpc_error(e) - + response = self.stub.send_msg(request, timeout=timeout) return response # ----- Boot workflow ----- @@ -213,20 +202,8 @@ def boot( ) touch_and_chmod(opmon_file) - try: - response = self.stub.boot(request, timeout=timeout) - yield response - - except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - self.log.error(error_details) - except Exception as extraction_error: - self.log.debug( - f"Could not extract rich error details from gRPC error: {extraction_error}", - exc_info=True, - ) - handle_grpc_error(e) + response = self.stub.boot(request, timeout=timeout) + yield response # Step 7: discover segment root controller self._discover_controller( @@ -322,9 +299,9 @@ def _build_boot_request( data_path = app.get("data_path") env["DUNE_DAQ_BASE_RELEASE"] = os.getenv("DUNE_DAQ_BASE_RELEASE") env["SPACK_RELEASES_DIR"] = os.getenv("SPACK_RELEASES_DIR") - # Some edge cases throw issues with DISPLAY being set, so we remove it from the + # Some edge cases throw issues with DISPLAY being set, so we remove it from the # environment - env.pop('DISPLAY', None) + env.pop("DISPLAY", None) tree_id = app["tree_id"] # The following line is required to provide an independent method of injecting @@ -834,20 +811,8 @@ def dummy_boot( ) self.log.debug(f"{request=}\n\n") - try: - response = self.stub.boot(request, timeout=timeout) - yield response - - except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - self.log.error(error_details) - except Exception as extraction_error: - self.log.debug( - f"Could not extract rich error details from gRPC error: {extraction_error}", - exc_info=True, - ) - handle_grpc_error(e) + response = self.stub.boot(request, timeout=timeout) + yield response def _prepare_exec_and_args_dummy_boot(self, sleep: int, n_sleeps: int) -> list: args = [ @@ -890,18 +855,7 @@ def terminate( ) -> ProcessInstanceList: request = Request(token=copy_token(self.token)) - try: - response = self.stub.terminate(request, timeout=timeout) - except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - self.log.error(error_details) - except Exception as extraction_error: - self.log.debug( - f"Could not extract rich error details from gRPC error: {extraction_error}", - exc_info=True, - ) - handle_grpc_error(e) + response = self.stub.terminate(request, timeout=timeout) return response @@ -910,72 +864,36 @@ def kill( ) -> ProcessInstanceList: request.token.CopyFrom(self.token) - try: - response = self.stub.kill(request, timeout=timeout) - except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - self.log.error(error_details) - except Exception as extraction_error: - self.log.debug( - f"Could not extract rich error details from gRPC error: {extraction_error}", - exc_info=True, - ) - handle_grpc_error(e) + response = self.stub.kill(request, timeout=timeout) return response def logs(self, request: LogRequest, timeout: int | float = 60) -> LogLines | None: request.token.CopyFrom(self.token) - try: - response = self.stub.logs(request, timeout=timeout) - - # Check if the response indicates a BadQuery error - if response.flag == ResponseFlag.NOT_EXECUTED_BAD_REQUEST_FORMAT: - lines = response.lines - if len(lines) == 1: - lines = lines[0] - self.log.warning(f"Bad query for logs: {lines}") - return None - - # Check for other error flags - if response.flag == ResponseFlag.DRUNC_EXCEPTION_THROWN: - self.log.error(f"Exception occurred on server: {response.lines}") - return None + response = self.stub.logs(request, timeout=timeout) - return response + # Check if the response indicates a BadQuery error + if response.flag == ResponseFlag.NOT_EXECUTED_BAD_REQUEST_FORMAT: + lines = response.lines + if len(lines) == 1: + lines = lines[0] + self.log.warning(f"Bad query for logs: {lines}") + return None - except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - self.log.error(error_details) - except Exception as extraction_error: - self.log.debug( - f"Could not extract rich error details from gRPC error: {extraction_error}", - exc_info=True, - ) - handle_grpc_error(e) + # Check for other error flags + if response.flag == ResponseFlag.DRUNC_EXCEPTION_THROWN: + self.log.error(f"Exception occurred on server: {response.lines}") return None + return response + def ps( self, request: ProcessQuery, timeout: int | float = 60 ) -> ProcessInstanceList: request.token.CopyFrom(self.token) - try: - response = self.stub.ps(request, timeout=timeout) - except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - self.log.error(error_details) - except Exception as extraction_error: - self.log.debug( - f"Could not extract rich error details from gRPC error: {extraction_error}", - exc_info=True, - ) - - handle_grpc_error(e) + response = self.stub.ps(request, timeout=timeout) return response @@ -984,19 +902,7 @@ def flush( ) -> ProcessInstanceList: request.token.CopyFrom(self.token) - try: - response = self.stub.flush(request, timeout=timeout) - except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - self.log.error(error_details) - except Exception as extraction_error: - self.log.debug( - f"Could not extract rich error details from gRPC error: {extraction_error}", - exc_info=True, - ) - - handle_grpc_error(e) + response = self.stub.flush(request, timeout=timeout) return response @@ -1005,38 +911,14 @@ def restart( ) -> ProcessInstanceList: request.token.CopyFrom(self.token) - try: - response = self.stub.restart(request, timeout=timeout) - except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - self.log.error(error_details) - except Exception as extraction_error: - self.log.debug( - f"Could not extract rich error details from gRPC error: {extraction_error}", - exc_info=True, - ) - - handle_grpc_error(e) + response = self.stub.restart(request, timeout=timeout) return response def describe(self, timeout: int | float = 60) -> Description: request = Request(token=copy_token(self.token)) - try: - response = self.stub.describe(request, timeout=timeout) - except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - self.log.error(error_details) - except Exception as extraction_error: - self.log.debug( - f"Could not extract rich error details from gRPC error: {extraction_error}", - exc_info=True, - ) - - handle_grpc_error(e) + response = self.stub.describe(request, timeout=timeout) return response diff --git a/src/drunc/utils/grpc_utils.py b/src/drunc/utils/grpc_utils.py index c39fd38d7..a968d6ea8 100644 --- a/src/drunc/utils/grpc_utils.py +++ b/src/drunc/utils/grpc_utils.py @@ -580,3 +580,7 @@ def __init__(self, logger): def intercept_unary_unary(self, continuation, client_call_details, request): response_call = continuation(client_call_details, request) return _CallWrapper(response_call, client_call_details.method, self.log) + + def intercept_unary_stream(self, continuation, client_call_details, request): + response_iterator = continuation(client_call_details, request) + return _CallWrapper(response_iterator, client_call_details.method, self.log) diff --git a/tests/process_manager/test_process_manager_driver.py b/tests/process_manager/test_process_manager_driver.py index 415afef0e..17f3813d1 100644 --- a/tests/process_manager/test_process_manager_driver.py +++ b/tests/process_manager/test_process_manager_driver.py @@ -4,6 +4,8 @@ This module tests that the ProcessManagerDriver correctly invokes the underlying gRPC stub methods and properly handles gRPC exceptions. +They use a real stub and interceptor and a mocked channel. + if any of these tests fail it is likely that the driver method implementations have changed. The tests should be checked to see if they need to be updated or if a bug was introduced. @@ -26,7 +28,6 @@ from drunc.connectivity_service.exceptions import ApplicationLookupUnsuccessful from drunc.exceptions import DruncSetupException, DruncShellException from drunc.process_manager.process_manager_driver import ProcessManagerDriver -from drunc.utils.grpc_utils import GrpcErrorDetails @pytest.fixture(scope="function") @@ -43,7 +44,6 @@ def mock_logger(): def mock_driver(mock_logger): """ Create a ProcessManagerDriver instance with a mocked gRPC stub. - This fixture creates a driver instance where the underlying gRPC channel and stub are mocked, allowing tests to verify method invocations without requiring a real gRPC server. @@ -63,10 +63,10 @@ def mock_driver(mock_logger): # Initialise driver with mocked dependencies driver = ProcessManagerDriver(address="localhost:50051", token=Token()) + driver.log = mock_logger - # Attach mock stub for easy access in tests + # Initialise driver with mocked dependencies driver._mock_stub = mock_stub - return driver @@ -104,9 +104,9 @@ def _setup(*, is_ready=True, grpc_error=None): # Configure the boot stub to either return a response or raise an error if grpc_error: - mock_driver.stub.boot = MagicMock(side_effect=grpc_error) + mock_driver._mock_stub.boot.side_effect = grpc_error else: - mock_driver.stub.boot = MagicMock(return_value="boot_response") + mock_driver._mock_stub.boot.return_value = "boot_response" return mock_request, csc_mock @@ -135,7 +135,7 @@ def test_collect_all_apps_merges_correctly(mock_infra_apps, mock_apps, mock_driv ) # Assert the stub wasn't used - assert mock_driver._mock_stub.method_calls == [] + assert mock_driver._mock_stub.boot.call_count == 0 # Assert the result is a merged list assert result == [ {"tree_id": "1.0", "name": "infra_app_1"}, @@ -277,42 +277,19 @@ def test_boot_handles_grpc_exception(mock_driver, boot_test_setup): grpc_error = grpc.RpcError("Connection failed") boot_test_setup(grpc_error=grpc_error) - error_details = GrpcErrorDetails( - code="INVALID_ARGUMENT", - message="Invalid request parameters", - details=["field_violations: field=token, description=Invalid token format"], - ) - - with ( - patch( - "drunc.process_manager.process_manager_driver.extract_grpc_rich_error" - ) as mock_extract, - patch( - "drunc.process_manager.process_manager_driver.handle_grpc_error" - ) as mock_handler, - patch("grpc_status.rpc_status.from_call", return_value=MagicMock()), - ): - mock_extract.return_value = error_details - mock_handler.side_effect = grpc_error - - # Expect the exception to be raised after error handling - with pytest.raises(grpc.RpcError): - list( - mock_driver.boot( - conf_file="conf.yaml", - conf_id="conf1", - user="u", - session_name="s", - log_level="INFO", - ) + with pytest.raises(grpc.RpcError) as excinfo: + list( + mock_driver.boot( + conf_file="conf.yaml", + conf_id="conf1", + user="u", + session_name="s", + log_level="INFO", ) + ) - mock_extract.assert_called_once_with(grpc_error) - mock_driver.log.error.assert_called_with(error_details) - mock_handler.assert_called_once_with(grpc_error) - - logged = mock_driver.log.error.call_args[0][0] - assert logged == error_details + assert excinfo.value is grpc_error + mock_driver._mock_stub.boot.assert_called_once() @patch("drunc.process_manager.process_manager_driver.get_log_path") @@ -622,21 +599,18 @@ def test_discover_controller_without_connectivity_service( @patch("drunc.process_manager.process_manager_driver.copy_token", return_value=Token()) -@patch("drunc.process_manager.process_manager_driver.handle_grpc_error") @patch( "drunc.process_manager.process_manager_driver.os.getcwd", return_value="/mocked/path", ) -def test_dummy_boot_success( - mock_getcwd, mock_handle_error, mock_copy_token, mock_driver -): +def test_dummy_boot_success(mock_getcwd, mock_copy_token, mock_driver): """ Test that `dummy_boot` creates and sends correct BootRequests and yields responses. """ mock_driver.token = Token() # Simulate gRPC stub returning two different responses for each process - mock_driver.stub.boot.side_effect = ["response_0", "response_1"] + mock_driver._mock_stub.boot.side_effect = ["response_0", "response_1"] result = list( mock_driver.dummy_boot( @@ -649,10 +623,10 @@ def test_dummy_boot_success( ) ) assert result == ["response_0", "response_1"] - assert mock_driver.stub.boot.call_count == 2 + assert mock_driver._mock_stub.boot.call_count == 2 # Assert each BootRequest sent to the stub - for i, call in enumerate(mock_driver.stub.boot.call_args_list): + for i, call in enumerate(mock_driver._mock_stub.boot.call_args_list): args, _ = call request = args[0] assert isinstance(request, BootRequest) @@ -674,52 +648,30 @@ def test_dummy_boot_success( ) def test_dummy_boot_grpc_error_handling(mock_getcwd, mock_copy_token, mock_driver): """ - Test that dummy_boot handles grpc.RpcError using handle_grpc_error(). - Simulates a gRPC failure during stub.boot and verifies that the error handler is invoked. + Test that dummy_boot propagates grpc.RpcError from the boot call. """ - # Setup mock driver and stub mock_driver.token = Token() mock_driver._prepare_exec_and_args_dummy_boot = MagicMock( return_value=[{"exec": "binary", "args": ["--arg1"]}] ) mock_driver._build_boot_request_dummy_boot = MagicMock() grpc_error = grpc.RpcError() - mock_driver.stub.boot.side_effect = grpc_error + mock_driver._mock_stub.boot.side_effect = grpc_error - error_details = GrpcErrorDetails( - code="INVALID_ARGUMENT", - message="Invalid request parameters", - details=["field_violations: field=token, description=Invalid token format"], - ) - with ( - patch( - "drunc.process_manager.process_manager_driver.extract_grpc_rich_error" - ) as mock_extract, - patch( - "drunc.process_manager.process_manager_driver.handle_grpc_error" - ) as mock_handler, - patch("grpc_status.rpc_status.from_call", return_value=MagicMock()), - ): - mock_extract.return_value = error_details - mock_handler.side_effect = grpc_error - with pytest.raises(grpc.RpcError): - list( - mock_driver.dummy_boot( - user="test_user", - session_name="test_session", - n_processes=1, - sleep=1, - n_sleeps=1, - timeout=30, - ) + with pytest.raises(grpc.RpcError) as excinfo: + list( + mock_driver.dummy_boot( + user="test_user", + session_name="test_session", + n_processes=1, + sleep=1, + n_sleeps=1, + timeout=30, ) + ) - mock_extract.assert_called_once_with(grpc_error) - mock_driver.log.error.assert_called_with(error_details) - mock_handler.assert_called_once_with(grpc_error) - - logged = mock_driver.log.error.call_args[0][0] - assert logged == error_details + assert excinfo.value is grpc_error + mock_driver._mock_stub.boot.assert_called_once() def test_prepare_exec_and_args_dummy_boot(mock_driver): @@ -819,58 +771,26 @@ def test_terminate_success(mock_driver, terminate_response): def test_terminate_grpc_error(mock_driver): """ Test that terminate method properly handles gRPC exceptions. - - Verifies that when the gRPC stub raises an exception, the driver - calls the error handling utility function which then re-raises. """ - # Configure mock stub to raise gRPC error grpc_error = grpc.RpcError("Connection failed") mock_driver._mock_stub.terminate.side_effect = grpc_error - with patch( - "drunc.process_manager.process_manager_driver.handle_grpc_error" - ) as mock_handler: - # Configure mock handler to re-raise as the real function does - mock_handler.side_effect = grpc_error + with pytest.raises(grpc.RpcError) as excinfo: + mock_driver.terminate() - # Expect the exception to be raised after error handling - with pytest.raises(grpc.RpcError): - mock_driver.terminate() - - # Verify error handler was called with the exception - mock_handler.assert_called_once_with(grpc_error) + assert excinfo.value is grpc_error + mock_driver._mock_stub.terminate.assert_called_once() -def test_terminate_error_no_grpc(mock_driver, mock_logger): - """ - Test that terminate method properly handles exceptions that are not gRPC. - """ - # Outer exception - grpc_err = grpc.RpcError("gRPC Failed") - mock_driver._mock_stub.terminate.side_effect = grpc_err +def test_terminate_non_grpc_error(mock_driver): + error = Exception("boom") + mock_driver._mock_stub.terminate.side_effect = error - # Inner exception for when extract_grpc_rich_error_fails - extract_grpc_rich_error_path = ( - "drunc.process_manager.process_manager_driver.extract_grpc_rich_error" - ) - handle_grpc_error_path = ( - "drunc.process_manager.process_manager_driver.handle_grpc_error" - ) + with pytest.raises(Exception) as excinfo: + mock_driver.terminate() - with ( - patch(extract_grpc_rich_error_path) as mock_extract_grpc_rich_error, - patch(handle_grpc_error_path) as mock_handler, - ): - # Simulate a TypeError - mock_extract_grpc_rich_error.side_effect = TypeError("Test TypeError") - mock_handler.side_effect = grpc_err - - with pytest.raises(grpc.RpcError): - mock_driver.terminate() - - args, kwargs = mock_logger.debug.call_args - assert "Could not extract rich error details" in args[0] - assert kwargs.get("exc_info") is True + assert excinfo.value is error + mock_driver._mock_stub.terminate.assert_called_once() def test_kill_success(mock_driver, process_query_request, kill_response): @@ -898,15 +818,11 @@ def test_kill_grpc_error(mock_driver, process_query_request): grpc_error = grpc.RpcError("Service unavailable") mock_driver._mock_stub.kill.side_effect = grpc_error - with patch( - "drunc.process_manager.process_manager_driver.handle_grpc_error" - ) as mock_handler: - mock_handler.side_effect = grpc_error + with pytest.raises(grpc.RpcError) as excinfo: + mock_driver.kill(process_query_request) - with pytest.raises(grpc.RpcError): - mock_driver.kill(process_query_request) - - mock_handler.assert_called_once_with(grpc_error) + assert excinfo.value is grpc_error + mock_driver._mock_stub.kill.assert_called_once() def test_logs_success(mock_driver, log_request, logs_response): @@ -934,15 +850,11 @@ def test_logs_grpc_error(mock_driver, log_request): grpc_error = grpc.RpcError("Authentication failed") mock_driver._mock_stub.logs.side_effect = grpc_error - with patch( - "drunc.process_manager.process_manager_driver.handle_grpc_error" - ) as mock_handler: - mock_handler.side_effect = grpc_error + with pytest.raises(grpc.RpcError) as excinfo: + mock_driver.logs(log_request) - with pytest.raises(grpc.RpcError): - mock_driver.logs(log_request) - - mock_handler.assert_called_once_with(grpc_error) + assert excinfo.value is grpc_error + mock_driver._mock_stub.logs.assert_called_once() def test_logs_bad_query_target_not_found(mock_driver, mock_logger, log_request): @@ -1000,15 +912,11 @@ def test_ps_grpc_error(mock_driver, process_query_request): grpc_error = grpc.RpcError("Request timeout") mock_driver._mock_stub.ps.side_effect = grpc_error - with patch( - "drunc.process_manager.process_manager_driver.handle_grpc_error" - ) as mock_handler: - mock_handler.side_effect = grpc_error - - with pytest.raises(grpc.RpcError): - mock_driver.ps(process_query_request) + with pytest.raises(grpc.RpcError) as excinfo: + mock_driver.ps(process_query_request) - mock_handler.assert_called_once_with(grpc_error) + assert excinfo.value is grpc_error + mock_driver._mock_stub.ps.assert_called_once() def test_flush_success(mock_driver, process_query_request, flush_response): @@ -1036,15 +944,11 @@ def test_flush_grpc_error(mock_driver, process_query_request): grpc_error = grpc.RpcError("Server error") mock_driver._mock_stub.flush.side_effect = grpc_error - with patch( - "drunc.process_manager.process_manager_driver.handle_grpc_error" - ) as mock_handler: - mock_handler.side_effect = grpc_error - - with pytest.raises(grpc.RpcError): - mock_driver.flush(process_query_request) + with pytest.raises(grpc.RpcError) as excinfo: + mock_driver.flush(process_query_request) - mock_handler.assert_called_once_with(grpc_error) + assert excinfo.value is grpc_error + mock_driver._mock_stub.flush.assert_called_once() def test_restart_success(mock_driver, process_query_request, restart_response): @@ -1072,15 +976,11 @@ def test_restart_grpc_error(mock_driver, process_query_request): grpc_error = grpc.RpcError("Network unreachable") mock_driver._mock_stub.restart.side_effect = grpc_error - with patch( - "drunc.process_manager.process_manager_driver.handle_grpc_error" - ) as mock_handler: - mock_handler.side_effect = grpc_error - - with pytest.raises(grpc.RpcError): - mock_driver.restart(process_query_request) + with pytest.raises(grpc.RpcError) as excinfo: + mock_driver.restart(process_query_request) - mock_handler.assert_called_once_with(grpc_error) + assert excinfo.value is grpc_error + mock_driver._mock_stub.restart.assert_called_once() def test_describe_success(mock_driver, describe_response): @@ -1109,12 +1009,8 @@ def test_describe_grpc_error(mock_driver): grpc_error = grpc.RpcError("Service not found") mock_driver._mock_stub.describe.side_effect = grpc_error - with patch( - "drunc.process_manager.process_manager_driver.handle_grpc_error" - ) as mock_handler: - mock_handler.side_effect = grpc_error - - with pytest.raises(grpc.RpcError): - mock_driver.describe() + with pytest.raises(grpc.RpcError) as excinfo: + mock_driver.describe() - mock_handler.assert_called_once_with(grpc_error) + assert excinfo.value is grpc_error + mock_driver._mock_stub.describe.assert_called_once() From fb18860e1ffa3b398ba8875ae505f1ffa3de8364 Mon Sep 17 00:00:00 2001 From: Miruna Serian Date: Tue, 4 Aug 2026 15:57:32 +0100 Subject: [PATCH 3/9] Update session manager tests --- src/drunc/utils/grpc_utils.py | 18 +++++++++--------- .../test_session_manager_driver.py | 6 ++---- .../test_session_manager_rich_errors.py | 10 ---------- 3 files changed, 11 insertions(+), 23 deletions(-) diff --git a/src/drunc/utils/grpc_utils.py b/src/drunc/utils/grpc_utils.py index e816a3b99..88c20e0ba 100644 --- a/src/drunc/utils/grpc_utils.py +++ b/src/drunc/utils/grpc_utils.py @@ -474,8 +474,10 @@ def error_wrapper(request: object, context: grpc.ServicerContext) -> object: return handler -class _CallWrapper: - """Wraps the gRPC Call/Future object to catch exceptions.""" +class _GRPCCallWrapper: + """Wraps the gRPC Call/Future object to catch exceptions. + This is because the gRPC errors happen when the response is read, not when the + request is made. So we need to wrap the Future object and catch the exception when the result is read.""" def __init__(self, call, method, logger): self._call = call @@ -484,7 +486,6 @@ def __init__(self, call, method, logger): def _handle_error(self, exception): if isinstance(exception, grpc.RpcError): - self._logger.error(f"gRPC Call Failed on method: {self._method}") try: error_details = extract_grpc_rich_error(exception) self._logger.error(error_details) @@ -498,6 +499,7 @@ def _handle_error(self, exception): def __getattr__(self, attr): # intercept the .result() call and process the error # to be used in tests if a mock passes an exception directly + # to do: make mocks pass a Future if attr == "result" and isinstance(self._call, Exception): def handle_mocked_error(*args, **kwargs): @@ -529,9 +531,7 @@ def __init__(self, logger): self.log = logger def intercept_unary_unary(self, continuation, client_call_details, request): - response_call = continuation(client_call_details, request) - return _CallWrapper(response_call, client_call_details.method, self.log) - - def intercept_unary_stream(self, continuation, client_call_details, request): - response_iterator = continuation(client_call_details, request) - return _CallWrapper(response_iterator, client_call_details.method, self.log) + response_call = continuation( + client_call_details, request + ) # continue the RPC call on the underlying channel + return _GRPCCallWrapper(response_call, client_call_details.method, self.log) diff --git a/tests/session_manager/test_session_manager_driver.py b/tests/session_manager/test_session_manager_driver.py index d34195cc1..ec2a67977 100644 --- a/tests/session_manager/test_session_manager_driver.py +++ b/tests/session_manager/test_session_manager_driver.py @@ -45,8 +45,7 @@ def test_grpc_error_handling(mock_driver, method_name): mock_extract.assert_called_once_with(grpc_error) mock_handler.assert_called_once_with(grpc_error) - assert mock_driver.log.error.call_count == 2 - logged = mock_driver.log.error.call_args_list[1][0][0] + logged = mock_driver.log.error.call_args_list[0][0][0] assert logged == error_details @@ -94,6 +93,5 @@ def test_grpc_error_fallback(mock_driver, method_name): # Assert fallback logic mock_extract.assert_called_once() - assert mock_driver.log.error.call_count == 2 - logged = mock_driver.log.error.call_args_list[1][0][0] + logged = mock_driver.log.error.call_args_list[0][0][0] assert logged == error_details diff --git a/tests/session_manager/test_session_manager_rich_errors.py b/tests/session_manager/test_session_manager_rich_errors.py index 0ff9fbc31..5255af2c5 100644 --- a/tests/session_manager/test_session_manager_rich_errors.py +++ b/tests/session_manager/test_session_manager_rich_errors.py @@ -89,7 +89,6 @@ def test_list_all_configs_no_config_files_rich_error( session_manager_rich_error_test_suite.setup_server_and_client() stub = session_manager_rich_error_test_suite.stub - mock_logger = session_manager_rich_error_test_suite.mock_client_logger # Remove the DUNEDAQ_DB_PATH from the environment to simulate it's not set monkeypatch.delenv("DUNEDAQ_DB_PATH", raising=False) @@ -102,9 +101,6 @@ def test_list_all_configs_no_config_files_rich_error( assert err.code() == grpc.StatusCode.FAILED_PRECONDITION assert "DUNEDAQ_DB_PATH" in err.details() - # The interceptor calls log.error twice (once for the method, once for the details) - assert mock_logger.error.call_count == 2 - # Unpack rich error metadata status = status_pb2.Status() for key, value in err.trailing_metadata(): @@ -142,7 +138,6 @@ def test_no_config_files_rich_error( ): session_manager_rich_error_test_suite.setup_server_and_client() stub = session_manager_rich_error_test_suite.stub - mock_logger = session_manager_rich_error_test_suite.mock_client_logger monkeypatch.setenv("DUNEDAQ_DB_PATH", "/fake_path") @@ -154,9 +149,6 @@ def test_no_config_files_rich_error( assert err.code() == grpc.StatusCode.FAILED_PRECONDITION assert "Config files" in err.details() - # The interceptor calls log.error twice (once for the method, once for the details) - assert mock_logger.error.call_count == 2 - # Unpack rich error metadata status = status_pb2.Status() for key, value in err.trailing_metadata(): @@ -252,7 +244,6 @@ def test_dals_missing_or_invalid( ): session_manager_rich_error_test_suite.setup_server_and_client() stub = session_manager_rich_error_test_suite.stub - mock_logger = session_manager_rich_error_test_suite.mock_client_logger # Set env var so search_paths is non-empty monkeypatch.setenv("DUNEDAQ_DB_PATH", "valid_path/") @@ -272,7 +263,6 @@ def test_dals_missing_or_invalid( err = excinfo.value assert err.code() == grpc.StatusCode.FAILED_PRECONDITION - assert mock_logger.error.call_count == 2 # Unpack rich error metadata status = status_pb2.Status() From 70572c4b91c93bdfa8fe6f1f508c02bf6a54bf0c Mon Sep 17 00:00:00 2001 From: Miruna Serian Date: Tue, 4 Aug 2026 16:34:33 +0100 Subject: [PATCH 4/9] Update markdown with client interceptor --- src/rich_error_handling_overview.md | 61 ++++++++++++++++++++--------- 1 file changed, 43 insertions(+), 18 deletions(-) diff --git a/src/rich_error_handling_overview.md b/src/rich_error_handling_overview.md index bd29e6158..e9acd3cb0 100644 --- a/src/rich_error_handling_overview.md +++ b/src/rich_error_handling_overview.md @@ -16,17 +16,18 @@ Every custom `drunc` exception (`DruncException`) is built to automatically gene * **`reason`:** A machine-readable string to categorise the error, such as `NOT_IMPLEMENTED` or `COMMAND_ERROR`. * **`domain`:** The component or part that generated the error (e.g., `drunc`, `ProcessManager.boot`). * **`details`:** Human-readable context (e.g., "DUNEDAQ_DB_PATH is not set"). + ### ii. Specialised Error Details -Sometimes, the standard `ErrorInfo` isn't structured enough. gRPC provides specialised Protobuf messages for specific failure states, such as `PreconditionFailure`, `BadRequest`, or `ResourceInfo` and [more](https://github.com/googleapis/googleapis/blob/master/google/rpc/error_details.proto). +Sometimes, the standard `ErrorInfo` isn't structured enough. gRPC provides specialised Protobuf messages for specific failure states, such as `PreconditionFailure`, `BadRequest`, or `ResourceInfo` and [more](https://github.com/googleapis/googleapis/blob/master/google/rpc/error_details.proto). You can append these specialised Protobuf objects to the error by overriding the `@property def specialised_details(self)` method in the custom exception. #### Example: PreconditionFailure + If a service fails to start you could use `PreconditionFailure. ```python - class DruncSetupException(DruncException): grpc_error_code = code_pb2.FAILED_PRECONDITION reason = "SETUP_FAILED" @@ -44,13 +45,14 @@ class DruncSetupException(DruncException): ] ) return [precond] - ``` + ## 2. Server-Side - Throwing Errors You do not need to manually pack Protobuf objects. Simply raise a subclass of `DruncException`. ### Example: Raising an existing exception + ```python from drunc.exceptions import DruncSetupException @@ -58,31 +60,54 @@ from drunc.exceptions import DruncSetupException if not config_files: raise DruncSetupException( message="Config files missing", - details=f"No configuration files found in {search_paths}" + details=f"No configuration files found in {search_paths}", ) ``` + You can overwrite any of the `message`, `grpc_error_code`,`details`,`reason` and `domain`fields. If not explicitly provided, they will automatically fall back to the standard default values defined in the base error. -## 3. Client-Side - Catching Errors +## 3. Server-Side - Server Interceptor -**Provisional: this will change with the implementation of a `ClientInterceptor`** +`RichErrorServerInterceptor` is a `grpc.ServerInterceptor` (defined in `src/drunc/utils/grpc_utils.py`) that wraps every unary-unary handler in a try/except block. When a `DruncException` is caught, it calls `abort_with_rich_details()` to pack the `ErrorInfo` and any specialised details into the gRPC trailing metadata before aborting the call. Non-`DruncException` errors and streaming calls are passed through unchanged. -Use `extract_grpc_rich_error` to unpack the rich metadata: +Implementing the interceptor: ```python -try: - response = stub.boot(request, timeout=60) -except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - log.error(error_details) - except Exception as extraction_error: - log.debug(f"Could not extract rich error: {extraction_error}") - - handle_grpc_error(e) +import grpc +from drunc.utils.grpc_utils import RichErrorServerInterceptor + +server = grpc.server( + futures.ThreadPoolExecutor(max_workers=10), + interceptors=[RichErrorServerInterceptor()], +) ``` -## 4. Testing gRPC errors +## 4. Client-Side - Catching Errors with a Client Interceptor + +Client-side rich error handling is done automatically by the `RichErrorClientInterceptor` (located in `src/drunc/utils/grpc_utils.py`). + +Because gRPC network calls are asynchronous, a failed request doesn't throw an error the moment it is sent. Instead, the `RpcError` is raised when the application actually tries to read the data by calling `.result()` on the response `Future`. `_GRPCCallWrapper` wraps the `Future` returned by `continuation` and delays error handling until `.result()` or `.exception()` is actually called by the application. + +If the interceptor can't find any rich metadata or if extraction fails, it simply logs a debug warning and passes the standard gRPC error along unchanged. + +### Setting up a client with the interceptor + +```python +self.log = get_logger("process_manager_driver", rich_handler=True) +self.address = address +options = [ + ("grpc.keepalive_time_ms", 60000) # pings the server every 60 seconds +] +raw_channel = grpc.insecure_channel(self.address, options=options) +rich_interceptor = RichErrorClientInterceptor(logger=self.log) +self.channel = grpc.intercept_channel(raw_channel, rich_interceptor) +self.stub = ProcessManagerStub(self.channel) +self.token = copy_token(token) +``` + +Once the channel is set up with the interceptor, **no extra try/except blocks are needed** to extract the rich error. Any `grpc.RpcError` that contains rich errors will be logged automatically before being re-raised. + +## 5. Testing gRPC errors Because gRPC status.details is a dynamic list, you need to iterate and check the type using `.Is()`. From f21e36de1214ea11c080f6dd8e175ff388d43e11 Mon Sep 17 00:00:00 2001 From: Miruna Serian Date: Tue, 4 Aug 2026 17:10:58 +0100 Subject: [PATCH 5/9] Update tests docstrings --- tests/session_manager/conftest.py | 23 ++++++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/tests/session_manager/conftest.py b/tests/session_manager/conftest.py index 5f4152d63..f5b2fb8f5 100644 --- a/tests/session_manager/conftest.py +++ b/tests/session_manager/conftest.py @@ -150,7 +150,12 @@ def mock_logger_driver(): class MockGrpcCall: - """A mock of the gRPC Future (Call) object.""" + """A mock of the gRPC Future (Call) object. + + The real gRPC channel returns a (response, call) tuple from `with_call`. + The interceptor uses the `call` object to read the response and any trailing metadata. + This mock allows that without requiring a real network connection. + """ def __init__(self, response=None): self._response = response @@ -160,7 +165,11 @@ def result(self): class FakeMultiCallable: - """Simulates a gRPC method endpoint.""" + """Simulates a single gRPC unary-unary method endpoint. + + The real `grpc.Channel.unary_unary()` returns a callable that, when called + with a request, performs the RPC. `FakeMultiCallable` replaces that callable. + """ def __init__(self, channel): self.channel = channel @@ -174,7 +183,7 @@ def with_call(self, request, *args, **kwargs): class FakeChannel(grpc.Channel): - """A fake gRPC channel.""" + """A fake gRPC channel used to test the SessionManagerDriver in isolation.""" def __init__(self): self.response = None @@ -204,6 +213,14 @@ def unsubscribe(self, *args, **kwargs): @pytest.fixture(scope="function") def mock_driver(mock_logger_driver): + """Create a SessionManagerDriver with a FakeChannel instead of a real gRPC connection. + + `grpc.insecure_channel` returns a `FakeChannel`, so the driver + initialises normally. + + The `FakeChannel` is attached as `driver._fake_channel` so individual tests + can set `.response` or `.error` on it to control what each stub call returns. + """ fake_channel = FakeChannel() with patch( From 8c9bec7c34f680e13a939eaddd982f88d8ebe956 Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Thu, 6 Aug 2026 11:32:37 +0200 Subject: [PATCH 6/9] Moving documentation to docs path --- {src => docs}/rich_error_handling_overview.md | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename {src => docs}/rich_error_handling_overview.md (100%) diff --git a/src/rich_error_handling_overview.md b/docs/rich_error_handling_overview.md similarity index 100% rename from src/rich_error_handling_overview.md rename to docs/rich_error_handling_overview.md From 4ffb73d5d7067d14493f06fc57fb471abac6514e Mon Sep 17 00:00:00 2001 From: Miruna Serian Date: Fri, 7 Aug 2026 17:54:32 +0100 Subject: [PATCH 7/9] Fix indentation in process manager driver --- .../process_manager/process_manager_driver.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/drunc/process_manager/process_manager_driver.py b/src/drunc/process_manager/process_manager_driver.py index 1dc8e0477..cb2141165 100644 --- a/src/drunc/process_manager/process_manager_driver.py +++ b/src/drunc/process_manager/process_manager_driver.py @@ -208,13 +208,13 @@ def boot( ) touch_and_chmod(opmon_file) - response = self.stub.boot(request, timeout=timeout) - self.log.info( - f"Booted '{request.process_description.metadata.name}' " - f"from session '{request.process_description.metadata.session}' " - f"with UUID {response.values[0].uuid.uuid} on host {request.process_description.metadata.hostname}" - ) - yield response + response = self.stub.boot(request, timeout=timeout) + self.log.info( + f"Booted '{request.process_description.metadata.name}' " + f"from session '{request.process_description.metadata.session}' " + f"with UUID {response.values[0].uuid.uuid} on host {request.process_description.metadata.hostname}" + ) + yield response # Step 7: discover segment root controller self._discover_controller( From d7f88f5ec1389dd23602c504f886255f2240b0ff Mon Sep 17 00:00:00 2001 From: Miruna Serian Date: Thu, 13 Aug 2026 07:40:57 +0100 Subject: [PATCH 8/9] Add more checks for RichErrorClientInterceptor --- .../test_process_manager_rich_errors.py | 226 +++++++++++++++--- .../test_session_manager_rich_errors.py | 57 ++++- 2 files changed, 243 insertions(+), 40 deletions(-) diff --git a/tests/process_manager/test_process_manager_rich_errors.py b/tests/process_manager/test_process_manager_rich_errors.py index 35f6b3bc0..c998732dc 100644 --- a/tests/process_manager/test_process_manager_rich_errors.py +++ b/tests/process_manager/test_process_manager_rich_errors.py @@ -1,3 +1,12 @@ +"""Test rich error handling with a real gRPC server with RichErrorServerInterceptor +and a real client stub with RichErrorClientInterceptor. + +These tests check: +- server-side exception mapping to gRPC status and rich details +- client-interceptor handling by asserting that `extract_grpc_rich_error` and +the interceptor logger are called. +""" + from concurrent import futures from unittest.mock import MagicMock, patch @@ -8,7 +17,11 @@ add_ProcessManagerServicer_to_server, ) -from drunc.utils.grpc_utils import RichErrorServerInterceptor, extract_grpc_rich_error +from drunc.utils.grpc_utils import ( + RichErrorClientInterceptor, + RichErrorServerInterceptor, + extract_grpc_rich_error, +) from tests.process_manager.process_manager_mock_impls import ( ConcreteProcessManager, ) @@ -24,6 +37,7 @@ def __init__(self): self.channel = None self.stub = None self.servicer = None + self.mock_client_logger = None def setup_server_and_client(self): """ @@ -55,8 +69,13 @@ def setup_server_and_client(self): self.server.add_insecure_port(listen_addr) self.server.start() + # Create a mock logger for the client interceptor + self.mock_client_logger = MagicMock() + client_interceptor = RichErrorClientInterceptor(logger=self.mock_client_logger) + # Create client channel and stub - self.channel = grpc.insecure_channel(self.server_address) + raw_channel = grpc.insecure_channel(self.server_address) + self.channel = grpc.intercept_channel(raw_channel, client_interceptor) self.stub = ProcessManagerStub(self.channel) def teardown_server_and_client(self): @@ -103,23 +122,48 @@ def process_manager_rich_error_test_suite(): suite.teardown_server_and_client() +@pytest.fixture(scope="function") +def request_by_method( + boot_request, process_query_request, generic_request, log_request +): + return { + "boot": boot_request, + "restart": process_query_request, + "kill": process_query_request, + "terminate": generic_request, + "ps": process_query_request, + "logs": log_request, + "flush": process_query_request, + } + + +METHODS_WITH_REQUEST = [ + ("boot", "_boot_impl"), + ("restart", "_restart_impl"), + ("kill", "_kill_impl"), + ("ps", "_ps_impl"), + ("logs", "_logs_impl"), + ("flush", "_flush_impl"), +] + +METHODS_WITHOUT_REQUEST = [ + ("terminate", "_terminate_impl"), +] + + @pytest.mark.parametrize( "method_name, impl_name", - [ - ("boot", "_boot_impl"), - ("restart", "_restart_impl"), - ("kill", "_kill_impl"), - ("terminate", "_terminate_impl"), - ("ps", "_ps_impl"), - ("logs", "_logs_impl"), - ], + METHODS_WITH_REQUEST, ) -def test_all_methods_not_implemented( - process_manager_rich_error_test_suite, ers_env, method_name, impl_name, boot_request +def test_methods_with_request_not_implemented( + process_manager_rich_error_test_suite, + ers_env, + request_by_method, + method_name, + impl_name, ): """ - Parametrized test to verify that all ProcessManager methods correctly - handle NotImplementedError by returning a Rich Error. + Test that methods correctly handle NotImplementedError by returning a Rich Error. """ # Setup the test suite @@ -128,12 +172,21 @@ def test_all_methods_not_implemented( # Mock the specific implementation method mock_impl = MagicMock(side_effect=NotImplementedError()) setattr(process_manager_rich_error_test_suite.servicer, impl_name, mock_impl) + request = request_by_method[method_name] # Call the method via the stub stub_method = getattr(process_manager_rich_error_test_suite.stub, method_name) - with pytest.raises(grpc.RpcError) as exc_info: - stub_method(boot_request) + # Patch but allow the real extract_grpc_rich_error to be called as this only happens + # when the ClientInterceptor catches the RpcError + with patch( + "drunc.utils.grpc_utils.extract_grpc_rich_error", + wraps=extract_grpc_rich_error, + ) as mock_extract_grpc_rich_error: + with pytest.raises(grpc.RpcError) as exc_info: + stub_method(request) + + mock_impl.assert_called_once_with(request) err = exc_info.value assert err.code() == grpc.StatusCode.UNIMPLEMENTED @@ -145,26 +198,75 @@ def test_all_methods_not_implemented( error_info = rich_error.details[0] assert error_info is not None + assert rich_error.code == "UNIMPLEMENTED" assert error_info.reason == "NOT_IMPLEMENTED" + assert error_info.domain == f"ProcessManager.{method_name}" + mock_extract_grpc_rich_error.assert_called_once() + process_manager_rich_error_test_suite.mock_client_logger.error.assert_called_once() @pytest.mark.parametrize( "method_name, impl_name", - [ - ("boot", "_boot_impl"), - ("restart", "_restart_impl"), - ("kill", "_kill_impl"), - ("terminate", "_terminate_impl"), - ("ps", "_ps_impl"), - ("logs", "_logs_impl"), - ], + METHODS_WITHOUT_REQUEST, ) -def test_all_methods_unhandled_exception( - process_manager_rich_error_test_suite, ers_env, method_name, impl_name, boot_request +def test_methods_without_request_not_implemented( + process_manager_rich_error_test_suite, + ers_env, + request_by_method, + method_name, + impl_name, ): """ - Parametrized test to verify that all ProcessManager methods correctly - handle DruncCommandExceptions by returning an INTERNAL error with ErrorInfo. + Check that methods for which the implementation takes no request argument + return rich errors when NotImplementedError is raised. + """ + + process_manager_rich_error_test_suite.setup_server_and_client() + + mock_impl = MagicMock(side_effect=NotImplementedError()) + setattr(process_manager_rich_error_test_suite.servicer, impl_name, mock_impl) + request = request_by_method[method_name] + + stub_method = getattr(process_manager_rich_error_test_suite.stub, method_name) + + with patch( + "drunc.utils.grpc_utils.extract_grpc_rich_error", + wraps=extract_grpc_rich_error, + ) as mock_extract_grpc_rich_error: + with pytest.raises(grpc.RpcError) as exc_info: + stub_method(request) + + mock_impl.assert_called_once_with() + + err = exc_info.value + assert err.code() == grpc.StatusCode.UNIMPLEMENTED + assert "Implementation missing" in err.details() + + rich_error = extract_grpc_rich_error(err) + error_info = rich_error.details[0] + + assert error_info is not None + assert rich_error.code == "UNIMPLEMENTED" + assert error_info.reason == "NOT_IMPLEMENTED" + assert error_info.domain == f"ProcessManager.{method_name}" + mock_extract_grpc_rich_error.assert_called_once() + process_manager_rich_error_test_suite.mock_client_logger.error.assert_called_once() + + +@pytest.mark.parametrize( + "method_name, impl_name", + METHODS_WITH_REQUEST, +) +def test_methods_with_request_unhandled_exception( + process_manager_rich_error_test_suite, + ers_env, + request_by_method, + method_name, + impl_name, +): + """ + Check that methods handle DruncCommandExceptions by returning an I + NTERNAL error with ErrorInfo. """ # Setup the test suite @@ -174,12 +276,19 @@ def test_all_methods_unhandled_exception( exception_msg = f"Unexpected error in {method_name}" mock_impl = MagicMock(side_effect=ValueError(exception_msg)) setattr(process_manager_rich_error_test_suite.servicer, impl_name, mock_impl) + request = request_by_method[method_name] # Call the method via the stub stub_method = getattr(process_manager_rich_error_test_suite.stub, method_name) - with pytest.raises(grpc.RpcError) as exc_info: - stub_method(boot_request) + with patch( + "drunc.utils.grpc_utils.extract_grpc_rich_error", + wraps=extract_grpc_rich_error, + ) as mock_extract_grpc_rich_error: + with pytest.raises(grpc.RpcError) as exc_info: + stub_method(request) + + mock_impl.assert_called_once_with(request) err = exc_info.value err_msg = f"Unhandled exception in ProcessManager.{method_name}" @@ -193,6 +302,59 @@ def test_all_methods_unhandled_exception( error_info = rich_error.details[0] assert error_info is not None + assert rich_error.code == "INTERNAL" + assert error_info.reason == "COMMAND_ERROR" + assert error_info.domain == f"ProcessManager.{method_name}" + mock_extract_grpc_rich_error.assert_called_once() + process_manager_rich_error_test_suite.mock_client_logger.error.assert_called_once() + + +@pytest.mark.parametrize( + "method_name, impl_name", + METHODS_WITHOUT_REQUEST, +) +def test_methods_without_request_unhandled_exception( + process_manager_rich_error_test_suite, + ers_env, + request_by_method, + method_name, + impl_name, +): + """ + Check that the methods for which the implementation + takes no request argument return INTERNAL rich errors for unhandled exceptions. + """ + + process_manager_rich_error_test_suite.setup_server_and_client() + + exception_msg = f"Unexpected error in {method_name}" + mock_impl = MagicMock(side_effect=ValueError(exception_msg)) + setattr(process_manager_rich_error_test_suite.servicer, impl_name, mock_impl) + request = request_by_method[method_name] + + stub_method = getattr(process_manager_rich_error_test_suite.stub, method_name) + + with patch( + "drunc.utils.grpc_utils.extract_grpc_rich_error", + wraps=extract_grpc_rich_error, + ) as mock_extract_grpc_rich_error: + with pytest.raises(grpc.RpcError) as exc_info: + stub_method(request) + + mock_impl.assert_called_once_with() + + err = exc_info.value + err_msg = f"Unhandled exception in ProcessManager.{method_name}" + assert err.code() == grpc.StatusCode.INTERNAL + assert err_msg in err.details() + assert exception_msg in err.details() - assert f"ProcessManager.{method_name}" in error_info.domain - assert "" in error_info.domain + rich_error = extract_grpc_rich_error(err) + error_info = rich_error.details[0] + + assert error_info is not None + assert rich_error.code == "INTERNAL" + assert error_info.reason == "COMMAND_ERROR" + assert error_info.domain == f"ProcessManager.{method_name}" + mock_extract_grpc_rich_error.assert_called_once() + process_manager_rich_error_test_suite.mock_client_logger.error.assert_called_once() diff --git a/tests/session_manager/test_session_manager_rich_errors.py b/tests/session_manager/test_session_manager_rich_errors.py index 5255af2c5..b8c59d62c 100644 --- a/tests/session_manager/test_session_manager_rich_errors.py +++ b/tests/session_manager/test_session_manager_rich_errors.py @@ -1,3 +1,13 @@ +"""Test rich error handling with a real gRPC server with RichErrorServerInterceptor +and a real client stub with RichErrorClientInterceptor. + +These tests check: +- server-side exception mapping to gRPC status and rich details +- manually unpack grpc-status-details-bin to test specific ErrorInfo and PreconditionFailure fields +- client-interceptor handling by asserting that `extract_grpc_rich_error` and +the interceptor logger are called. +""" + from concurrent import futures from pathlib import Path from unittest.mock import MagicMock, patch @@ -14,6 +24,7 @@ from drunc.utils.grpc_utils import ( RichErrorClientInterceptor, RichErrorServerInterceptor, + extract_grpc_rich_error, ) @@ -93,8 +104,13 @@ def test_list_all_configs_no_config_files_rich_error( # Remove the DUNEDAQ_DB_PATH from the environment to simulate it's not set monkeypatch.delenv("DUNEDAQ_DB_PATH", raising=False) - with pytest.raises(grpc.RpcError) as excinfo: - stub.list_all_configs(generic_request) + # Patch but allow the real extract_grpc_rich_error to be called + with patch( + "drunc.utils.grpc_utils.extract_grpc_rich_error", + wraps=extract_grpc_rich_error, + ) as mock_extract_grpc_rich_error: + with pytest.raises(grpc.RpcError) as excinfo: + stub.list_all_configs(generic_request) err = excinfo.value @@ -132,6 +148,9 @@ def test_list_all_configs_no_config_files_rich_error( assert violation.type == "MISSING OR INVALID" assert "DUNEDAQ_DB_PATH env variable not set" in violation.description + mock_extract_grpc_rich_error.assert_called_once() + session_manager_rich_error_test_suite.mock_client_logger.error.assert_called_once() + def test_no_config_files_rich_error( session_manager_rich_error_test_suite, generic_request, monkeypatch @@ -141,8 +160,13 @@ def test_no_config_files_rich_error( monkeypatch.setenv("DUNEDAQ_DB_PATH", "/fake_path") - with pytest.raises(grpc.RpcError) as excinfo: - stub.list_all_configs(generic_request) + # Patch but allow the real extract_grpc_rich_error to be called + with patch( + "drunc.utils.grpc_utils.extract_grpc_rich_error", + wraps=extract_grpc_rich_error, + ) as mock_extract_grpc_rich_error: + with pytest.raises(grpc.RpcError) as excinfo: + stub.list_all_configs(generic_request) err = excinfo.value @@ -180,6 +204,9 @@ def test_no_config_files_rich_error( assert "No configuration files found in /fake_path" in violation.description assert "Config files" in violation.subject + mock_extract_grpc_rich_error.assert_called_once() + session_manager_rich_error_test_suite.mock_client_logger.error.assert_called_once() + def test_config_parse_failure( session_manager_rich_error_test_suite, generic_request, monkeypatch @@ -199,8 +226,12 @@ def test_config_parse_failure( "drunc.session_manager.session_manager.Configuration", side_effect=Exception("Config failed"), ): - with pytest.raises(grpc.RpcError) as excinfo: - stub.list_all_configs(generic_request) + with patch( + "drunc.utils.grpc_utils.extract_grpc_rich_error", + wraps=extract_grpc_rich_error, + ) as mock_extract_grpc_rich_error: + with pytest.raises(grpc.RpcError) as excinfo: + stub.list_all_configs(generic_request) err = excinfo.value assert err.code() == grpc.StatusCode.FAILED_PRECONDITION @@ -238,6 +269,9 @@ def test_config_parse_failure( assert "Failed to parse configuration file" in violation.description assert "Services could not start" in violation.subject + mock_extract_grpc_rich_error.assert_called_once() + session_manager_rich_error_test_suite.mock_client_logger.error.assert_called_once() + def test_dals_missing_or_invalid( session_manager_rich_error_test_suite, generic_request, monkeypatch @@ -258,8 +292,12 @@ def test_dals_missing_or_invalid( "drunc.session_manager.session_manager.Configuration", return_value=fake_config, ): - with pytest.raises(grpc.RpcError) as excinfo: - stub.list_all_configs(generic_request) + with patch( + "drunc.utils.grpc_utils.extract_grpc_rich_error", + wraps=extract_grpc_rich_error, + ) as mock_extract_grpc_rich_error: + with pytest.raises(grpc.RpcError) as excinfo: + stub.list_all_configs(generic_request) err = excinfo.value assert err.code() == grpc.StatusCode.FAILED_PRECONDITION @@ -295,3 +333,6 @@ def test_dals_missing_or_invalid( assert "Session DALs" in violation.subject assert violation.type == "MISSING OR INVALID" assert "DALs missing or invalid" in violation.description + + mock_extract_grpc_rich_error.assert_called_once() + session_manager_rich_error_test_suite.mock_client_logger.error.assert_called_once() From 00add64c3fc1d907ce170a35bd293383d3db83c8 Mon Sep 17 00:00:00 2001 From: Miruna Serian Date: Thu, 13 Aug 2026 09:01:34 +0100 Subject: [PATCH 9/9] Add missing request.token.CopyFrom() in log_on_server --- src/drunc/process_manager/process_manager_driver.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/drunc/process_manager/process_manager_driver.py b/src/drunc/process_manager/process_manager_driver.py index cb2141165..5b408e73f 100644 --- a/src/drunc/process_manager/process_manager_driver.py +++ b/src/drunc/process_manager/process_manager_driver.py @@ -1016,6 +1016,7 @@ def log_on_server( execute_along_path=False, execute_on_all_subsequent_children_in_path=False, ) + request.token.CopyFrom(self.token) response: LogOnServerResponse = self.stub.log_on_server( request, timeout=timeout )