Commit 15914ae1 authored by Waleed Akbar's avatar Waleed Akbar
Browse files

Enhance gRPC error handling in pluggable service routes and update test file formatting

parent 87c32736

src/nbi/service/dscm_oc/routes.py

+27 −3
Original line number Diff line number Diff line
@@ -13,6 +13,7 @@ 
# limitations under the License.



import logging

import grpc

from .enforce_header                           import require_accept, require_content_type

from .error                                    import _bad_request, _not_found, yang_json

from .json_to_proto_conversion                 import (
@@ -71,6 +72,13 @@ def rc_get(rc_path, device_uuid=None): 
        LOGGER.warning(f"Pluggable not found for device {device_uuid}: {e.details}")

        return _not_found(f"Pluggable not found: {e.details}", path=rc_path)



    except grpc.RpcError as e:

        code = e.code()  # type: ignore[union-attr]

        if code == grpc.StatusCode.NOT_FOUND:

            return _not_found(f"Pluggable not found: {e.details()}", path=rc_path)  # type: ignore[union-attr]

        LOGGER.error(f"gRPC error getting pluggable for device {device_uuid}: {e}", exc_info=True)

        return _bad_request(f"Failed to get pluggable: {e.details()}", path=rc_path)  # type: ignore[union-attr]



    except ServiceException as e:

        LOGGER.error(f"Unexpected error getting pluggable for device {device_uuid}: {str(e)}", exc_info=True)

        return _bad_request(f"Failed to get pluggable: {str(e)}", path=rc_path)
@@ -113,6 +121,14 @@ def rc_post(rc_path, device_uuid=None): 
        LOGGER.warning(f"Invalid argument creating pluggable for device {device_uuid}: {e.details}")

        return _bad_request(f"Invalid argument: {e.details}", path=rc_path)



    except grpc.RpcError as e:

        code = e.code()  # type: ignore[union-attr]

        if code == grpc.StatusCode.ALREADY_EXISTS:

            LOGGER.warning(f"Pluggable already exists (gRPC) for device {device_uuid}: {e.details()}")

            return _bad_request(f"Pluggable already exists: {e.details()}", path=rc_path)  # type: ignore[union-attr]

        LOGGER.error(f"gRPC error creating pluggable for device {device_uuid}: {e}", exc_info=True)

        return _bad_request(f"Failed to create pluggable: {e.details()}", path=rc_path)  # type: ignore[union-attr]



    except ServiceException as e:

        LOGGER.error(f"Unexpected error creating pluggable for device {device_uuid}: {str(e)}", exc_info=True)

        return _bad_request(f"Failed to create pluggable: {str(e)}", path=rc_path)
@@ -137,6 +153,14 @@ def rc_delete(rc_path, device_uuid=None): 
        LOGGER.warning(f"Pluggable not found for device {device_uuid}: {e.details} (already deleted or never existed)")

        return Response(status=204)



    except grpc.RpcError as e:

        code = e.code()  # type: ignore[union-attr]

        if code == grpc.StatusCode.NOT_FOUND:

            LOGGER.warning(f"Pluggable not found (gRPC) for device {device_uuid} (already deleted or never existed)")

            return Response(status=204)

        LOGGER.error(f"gRPC error deleting pluggable for device {device_uuid}: {e}", exc_info=True)

        return _bad_request(f"Failed to delete pluggable: {e.details()}", path=rc_path)  # type: ignore[union-attr]



    except ServiceException as e:

        LOGGER.error(f"Unexpected error deleting pluggable for device {device_uuid}: {str(e)}", exc_info=True)

        return _bad_request(f"Failed to delete pluggable: {str(e)}", path=rc_path)

src/nbi/tests/test_dscm_restconf.py

+11 −8
Original line number Diff line number Diff line
@@ -47,13 +47,13 @@ DSCMPLUGGABLE_SERVICE_PORT = get_service_port_grpc(ServiceNameEnum.PLUGGABLES) 
os.environ[get_env_var_name(ServiceNameEnum.PLUGGABLES, ENVVAR_SUFIX_SERVICE_HOST     )] = str(LOCAL_HOST)

os.environ[get_env_var_name(ServiceNameEnum.PLUGGABLES, ENVVAR_SUFIX_SERVICE_PORT_GRPC)] = str(DSCMPLUGGABLE_SERVICE_PORT)



class MockContextService(GenericGrpcService):

    def __init__(self, bind_port: Union[str, int]) -> None:

        super().__init__(bind_port, LOCAL_HOST, enable_health_servicer=False, cls_name='MockService')

# class MockContextService(GenericGrpcService):

#     def __init__(self, bind_port: Union[str, int]) -> None:

#         super().__init__(bind_port, LOCAL_HOST, enable_health_servicer=False, cls_name='MockService')



    def install_servicers(self):

        self.context_servicer = MockServicerImpl_Context()

        add_ContextServiceServicer_to_server(self.context_servicer, self.server)

#     def install_servicers(self):

#         self.context_servicer = MockServicerImpl_Context()

#         add_ContextServiceServicer_to_server(self.context_servicer, self.server)





@pytest.fixture(scope='session')
@@ -99,6 +99,9 @@ def log_each(request): 
    yield

    LOGGER.info(f"<<<<<< END   {request.node.name} <<<<<<")





# Actual test cases



def test_post_hub_optical_channel_frequency(nbi_service_rest, dscm_pluggable_client: PluggablesClient):

    """Test PATCH to update optical channel frequency."""

    device = "device=T1.1/"