local_agent/translation_layer/command_handlers/base.py

# Copyright 2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Module for base command handler."""
import abc
import inflection
from typing import Any, Dict

from local_agent import errors as agent_errors


class BaseCommandHandler:
    """Base handler class for all command handlers."""

    SUPPORTED_METHODS = set()

    def __init__(self, dut: Any) -> None:
        """
        Base command handler constructor.

        Args:
            dut: Gazoo device instance.
        """
        self.dut = dut

    def handle_request(self, rpc_request: Dict[str, Any]) -> Dict[str, Any]:
        """
        The required method for handling the RPC request,
        the corresponding GDM device class operation should be called
        accordingly. Note that we don't return the failure JSON-RPC response
        explicitly since it's already handled by the exception handler in
        the local agent process.

        Args:
            rpc_request: JSON-RPC request.

        Returns:
            RPC response, which is the result of executing the command.
        """
        rpc_id = rpc_request['id']
        method = rpc_request['method']
        params = rpc_request['params']
        method_snake_case = f'_{inflection.underscore(method)}'
        func = getattr(self, method_snake_case, None)
        if func is None:
            raise agent_errors.InvalidRPCError(f'Unknown method {method}.')
        func_result = func(params)
        if func_result is not None:
            result = {'value': func_result}
        else:
            result = {}
        return {'id': rpc_id, 'jsonrpc': '2.0', 'result': result}

    @staticmethod
    def validate_key_in_params(
        params: Dict[str, Any], param_key: str, expected_type: Any) -> None:
        """Verifies params contains the key with the expected type.

        Args:
            params: RPC request parameters.
            param_key: Key of the parameter dict.
            expected_type: Expected type of the given key.

        Raises:
            ValueError: param_key does not exist in params or the expected_type
            does not match.
        """
        if param_key not in params:
            raise ValueError(f'Missing field {param_key} from RPC request.')
        param_value = params[param_key]
        if not isinstance(param_value, expected_type):
            raise ValueError(
                f'Invalid type for {param_key}. '
                f'Expecting {expected_type} while receiving {param_value}.')