src/antlion/controllers/fuchsia_lib/base_lib.py

#!/usr/bin/env python3
#
# Copyright 2022 The Fuchsia Authors
#
# 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.

import json
import socket

from typing import Any, Mapping
from urllib.request import Request, urlopen

from antlion import logger

DEFAULT_SL4F_RESPONSE_TIMEOUT_SEC = 30


class DeviceOffline(Exception):
    """Exception if the device is no longer reachable via the network."""


class SL4FCommandFailed(Exception):
    """A SL4F command to the server failed."""


class BaseLib:
    def __init__(self, addr: str, logger_tag: str) -> None:
        self.address = addr
        self.log = logger.create_tagged_trace_logger(
            f"SL4F | {self.address} | {logger_tag}"
        )

    def send_command(
        self,
        cmd: str,
        args: Mapping[str, Any],
        response_timeout: int = DEFAULT_SL4F_RESPONSE_TIMEOUT_SEC,
    ) -> Mapping[str, Any]:
        """Builds and sends a JSON command to SL4F server.

        Args:
            cmd: SL4F method name of command.
            args: Arguments required to execute cmd.
            response_timeout: Seconds to wait for a response before
                throwing an exception.

        Returns:
            Response from SL4F server.

        Throws:
            TimeoutError: The HTTP request timed out waiting for a response
        """
        data = {
            "jsonrpc": "2.0",
            # id is required by the SL4F server to parse test_data but is not
            # currently used.
            "id": "",
            "method": cmd,
            "params": args,
        }
        data_json = json.dumps(data).encode("utf-8")
        req = Request(
            self.address,
            data=data_json,
            headers={
                "Content-Type": "application/json; charset=utf-8",
                "Content-Length": len(data_json),
            },
        )

        self.log.debug(f'Sending request "{cmd}" with {args}')
        try:
            response = urlopen(req, timeout=response_timeout)
        except socket.timeout as e:
            # socket.timeout was aliased to TimeoutError in Python 3.10. For
            # older versions of Python, we need to cast to TimeoutError to
            # provide a version-agnostic API.
            raise TimeoutError("socket timeout") from e

        response_body = response.read().decode("utf-8")
        try:
            response_json = json.loads(response_body)
            self.log.debug(f'Received response for "{cmd}": {response_json}')
        except json.JSONDecodeError as e:
            raise SL4FCommandFailed(response_body) from e

        # If the SL4F command fails it returns a str, without an 'error' field
        # to get.
        if not isinstance(response_json, dict):
            raise SL4FCommandFailed(response_json)

        return response_json