python/qemu/utils/qom_common.py

"""
QOM Command abstractions.
"""
##
# Copyright John Snow 2020, for Red Hat, Inc.
# Copyright IBM, Corp. 2011
#
# Authors:
#  John Snow <jsnow@redhat.com>
#  Anthony Liguori <aliguori@amazon.com>
#
# This work is licensed under the terms of the GNU GPL, version 2 or later.
# See the COPYING file in the top-level directory.
#
# Based on ./scripts/qmp/qom-[set|get|tree|list]
##

import argparse
import os
import sys
from typing import (
    Any,
    Dict,
    List,
    Optional,
    Type,
    TypeVar,
)

from qemu.qmp import QMPError
from qemu.qmp.legacy import QEMUMonitorProtocol


class ObjectPropertyInfo:
    """
    Represents the return type from e.g. qom-list.
    """
    def __init__(self, name: str, type_: str,
                 description: Optional[str] = None,
                 default_value: Optional[object] = None):
        self.name = name
        self.type = type_
        self.description = description
        self.default_value = default_value

    @classmethod
    def make(cls, value: Dict[str, Any]) -> 'ObjectPropertyInfo':
        """
        Build an ObjectPropertyInfo from a Dict with an unknown shape.
        """
        assert value.keys() >= {'name', 'type'}
        assert value.keys() <= {'name', 'type', 'description', 'default-value'}
        return cls(value['name'], value['type'],
                   value.get('description'),
                   value.get('default-value'))

    @property
    def child(self) -> bool:
        """Is this property a child property?"""
        return self.type.startswith('child<')

    @property
    def link(self) -> bool:
        """Is this property a link property?"""
        return self.type.startswith('link<')


CommandT = TypeVar('CommandT', bound='QOMCommand')


class QOMCommand:
    """
    Represents a QOM sub-command.

    :param args: Parsed arguments, as returned from parser.parse_args.
    """
    name: str
    help: str

    def __init__(self, args: argparse.Namespace):
        if args.socket is None:
            raise QMPError("No QMP socket path or address given")
        self.qmp = QEMUMonitorProtocol(
            QEMUMonitorProtocol.parse_address(args.socket)
        )
        self.qmp.connect()

    @classmethod
    def register(cls, subparsers: Any) -> None:
        """
        Register this command with the argument parser.

        :param subparsers: argparse subparsers object, from "add_subparsers".
        """
        subparser = subparsers.add_parser(cls.name, help=cls.help,
                                          description=cls.help)
        cls.configure_parser(subparser)

    @classmethod
    def configure_parser(cls, parser: argparse.ArgumentParser) -> None:
        """
        Configure a parser with this command's arguments.

        :param parser: argparse parser or subparser object.
        """
        default_path = os.environ.get('QMP_SOCKET')
        parser.add_argument(
            '--socket', '-s',
            dest='socket',
            action='store',
            help='QMP socket path or address (addr:port).'
            ' May also be set via QMP_SOCKET environment variable.',
            default=default_path
        )
        parser.set_defaults(cmd_class=cls)

    @classmethod
    def add_path_prop_arg(cls, parser: argparse.ArgumentParser) -> None:
        """
        Add the <path>.<proptery> positional argument to this command.

        :param parser: The parser to add the argument to.
        """
        parser.add_argument(
            'path_prop',
            metavar='<path>.<property>',
            action='store',
            help="QOM path and property, separated by a period '.'"
        )

    def run(self) -> int:
        """
        Run this command.

        :return: 0 on success, 1 otherwise.
        """
        raise NotImplementedError

    def qom_list(self, path: str) -> List[ObjectPropertyInfo]:
        """
        :return: a strongly typed list from the 'qom-list' command.
        """
        rsp = self.qmp.cmd('qom-list', path=path)
        # qom-list returns List[ObjectPropertyInfo]
        assert isinstance(rsp, list)
        return [ObjectPropertyInfo.make(x) for x in rsp]

    @classmethod
    def command_runner(
            cls: Type[CommandT],
            args: argparse.Namespace
    ) -> int:
        """
        Run a fully-parsed subcommand, with error-handling for the CLI.

        :return: The return code from `run()`.
        """
        try:
            cmd = cls(args)
            return cmd.run()
        except QMPError as err:
            print(f"{type(err).__name__}: {err!s}", file=sys.stderr)
            return -1

    @classmethod
    def entry_point(cls) -> int:
        """
        Build this command's parser, parse arguments, and run the command.

        :return: `run`'s return code.
        """
        parser = argparse.ArgumentParser(description=cls.help)
        cls.configure_parser(parser)
        args = parser.parse_args()
        return cls.command_runner(args)