| 1 | """ |
| 2 | QOM Command abstractions. |
| 3 | """ |
| 4 | ## |
| 5 | # Copyright John Snow 2020, for Red Hat, Inc. |
| 6 | # Copyright IBM, Corp. 2011 |
| 7 | # |
| 8 | # Authors: |
| 9 | # John Snow <jsnow@redhat.com> |
| 10 | # Anthony Liguori <aliguori@amazon.com> |
| 11 | # |
| 12 | # This work is licensed under the terms of the GNU GPL, version 2 or later. |
| 13 | # See the COPYING file in the top-level directory. |
| 14 | # |
| 15 | # Based on ./scripts/qmp/qom-[set|get|tree|list] |
| 16 | ## |
| 17 | |
| 18 | import argparse |
| 19 | import os |
| 20 | import sys |
| 21 | from typing import ( |
| 22 | Any, |
| 23 | Dict, |
| 24 | List, |
| 25 | Optional, |
| 26 | Type, |
| 27 | TypeVar, |
| 28 | ) |
| 29 | |
| 30 | from qemu.qmp import QMPError |
| 31 | from qemu.qmp.legacy import QEMUMonitorProtocol |
| 32 | |
| 33 | |
| 34 | class ObjectPropertyInfo: |
| 35 | """ |
| 36 | Represents the return type from e.g. qom-list. |
| 37 | """ |
| 38 | def __init__(self, name: str, type_: str, |
| 39 | description: Optional[str] = None, |
| 40 | default_value: Optional[object] = None): |
| 41 | self.name = name |
| 42 | self.type = type_ |
| 43 | self.description = description |
| 44 | self.default_value = default_value |
| 45 | |
| 46 | @classmethod |
| 47 | def make(cls, value: Dict[str, Any]) -> 'ObjectPropertyInfo': |
| 48 | """ |
| 49 | Build an ObjectPropertyInfo from a Dict with an unknown shape. |
| 50 | """ |
| 51 | assert value.keys() >= {'name', 'type'} |
| 52 | assert value.keys() <= {'name', 'type', 'description', 'default-value'} |
| 53 | return cls(value['name'], value['type'], |
| 54 | value.get('description'), |
| 55 | value.get('default-value')) |
| 56 | |
| 57 | @property |
| 58 | def child(self) -> bool: |
| 59 | """Is this property a child property?""" |
| 60 | return self.type.startswith('child<') |
| 61 | |
| 62 | @property |
| 63 | def link(self) -> bool: |
| 64 | """Is this property a link property?""" |
| 65 | return self.type.startswith('link<') |
| 66 | |
| 67 | |
| 68 | class ObjectPropertyValue: |
| 69 | """ |
| 70 | Represents a property return from e.g. qom-tree-get |
| 71 | """ |
| 72 | def __init__(self, name: str, type_: str, value: object): |
| 73 | self.name = name |
| 74 | self.type = type_ |
| 75 | self.value = value |
| 76 | |
| 77 | @classmethod |
| 78 | def make(cls, value: Dict[str, Any]) -> 'ObjectPropertyValue': |
| 79 | """ |
| 80 | Build an ObjectPropertyValue from a Dict with an unknown shape. |
| 81 | """ |
| 82 | assert value.keys() >= {'name', 'type'} |
| 83 | assert value.keys() <= {'name', 'type', 'value'} |
| 84 | return cls(value['name'], value['type'], value.get('value')) |
| 85 | |
| 86 | @property |
| 87 | def child(self) -> bool: |
| 88 | """Is this property a child property?""" |
| 89 | return self.type.startswith('child<') |
| 90 | |
| 91 | |
| 92 | class ObjectPropertiesValues: |
| 93 | """ |
| 94 | Represents the return type from e.g. qom-list-get |
| 95 | """ |
| 96 | # pylint: disable=too-few-public-methods |
| 97 | |
| 98 | def __init__(self, properties: List[ObjectPropertyValue]) -> None: |
| 99 | self.properties = properties |
| 100 | |
| 101 | @classmethod |
| 102 | def make(cls, value: Dict[str, Any]) -> 'ObjectPropertiesValues': |
| 103 | """ |
| 104 | Build an ObjectPropertiesValues from a Dict with an unknown shape. |
| 105 | """ |
| 106 | assert value.keys() == {'properties'} |
| 107 | props = [ObjectPropertyValue(item['name'], |
| 108 | item['type'], |
| 109 | item.get('value')) |
| 110 | for item in value['properties']] |
| 111 | return cls(props) |
| 112 | |
| 113 | |
| 114 | CommandT = TypeVar('CommandT', bound='QOMCommand') |
| 115 | |
| 116 | |
| 117 | class QOMCommand: |
| 118 | """ |
| 119 | Represents a QOM sub-command. |
| 120 | |
| 121 | :param args: Parsed arguments, as returned from parser.parse_args. |
| 122 | """ |
| 123 | name: str |
| 124 | help: str |
| 125 | |
| 126 | def __init__(self, args: argparse.Namespace): |
| 127 | if args.socket is None: |
| 128 | raise QMPError("No QMP socket path or address given") |
| 129 | self.qmp = QEMUMonitorProtocol( |
| 130 | QEMUMonitorProtocol.parse_address(args.socket) |
| 131 | ) |
| 132 | self.qmp.connect() |
| 133 | |
| 134 | @classmethod |
| 135 | def register(cls, subparsers: Any) -> None: |
| 136 | """ |
| 137 | Register this command with the argument parser. |
| 138 | |
| 139 | :param subparsers: argparse subparsers object, from "add_subparsers". |
| 140 | """ |
| 141 | subparser = subparsers.add_parser(cls.name, help=cls.help, |
| 142 | description=cls.help) |
| 143 | cls.configure_parser(subparser) |
| 144 | |
| 145 | @classmethod |
| 146 | def configure_parser(cls, parser: argparse.ArgumentParser) -> None: |
| 147 | """ |
| 148 | Configure a parser with this command's arguments. |
| 149 | |
| 150 | :param parser: argparse parser or subparser object. |
| 151 | """ |
| 152 | default_path = os.environ.get('QMP_SOCKET') |
| 153 | parser.add_argument( |
| 154 | '--socket', '-s', |
| 155 | dest='socket', |
| 156 | action='store', |
| 157 | help='QMP socket path or address (addr:port).' |
| 158 | ' May also be set via QMP_SOCKET environment variable.', |
| 159 | default=default_path |
| 160 | ) |
| 161 | parser.set_defaults(cmd_class=cls) |
| 162 | |
| 163 | @classmethod |
| 164 | def add_path_prop_arg(cls, parser: argparse.ArgumentParser) -> None: |
| 165 | """ |
| 166 | Add the <path>.<proptery> positional argument to this command. |
| 167 | |
| 168 | :param parser: The parser to add the argument to. |
| 169 | """ |
| 170 | parser.add_argument( |
| 171 | 'path_prop', |
| 172 | metavar='<path>.<property>', |
| 173 | action='store', |
| 174 | help="QOM path and property, separated by a period '.'" |
| 175 | ) |
| 176 | |
| 177 | def run(self) -> int: |
| 178 | """ |
| 179 | Run this command. |
| 180 | |
| 181 | :return: 0 on success, 1 otherwise. |
| 182 | """ |
| 183 | raise NotImplementedError |
| 184 | |
| 185 | def qom_list(self, path: str) -> List[ObjectPropertyInfo]: |
| 186 | """ |
| 187 | :return: a strongly typed list from the 'qom-list' command. |
| 188 | """ |
| 189 | rsp = self.qmp.cmd('qom-list', path=path) |
| 190 | # qom-list returns List[ObjectPropertyInfo] |
| 191 | assert isinstance(rsp, list) |
| 192 | return [ObjectPropertyInfo.make(x) for x in rsp] |
| 193 | |
| 194 | def qom_list_get(self, paths: List[str]) -> List[ObjectPropertiesValues]: |
| 195 | """ |
| 196 | :return: a strongly typed list from the 'qom-list-get' command. |
| 197 | """ |
| 198 | rsp = self.qmp.cmd('qom-list-get', paths=paths) |
| 199 | # qom-list-get returns List[ObjectPropertiesValues] |
| 200 | assert isinstance(rsp, list) |
| 201 | return [ObjectPropertiesValues.make(x) for x in rsp] |
| 202 | |
| 203 | @classmethod |
| 204 | def command_runner( |
| 205 | cls: Type[CommandT], |
| 206 | args: argparse.Namespace |
| 207 | ) -> int: |
| 208 | """ |
| 209 | Run a fully-parsed subcommand, with error-handling for the CLI. |
| 210 | |
| 211 | :return: The return code from `run()`. |
| 212 | """ |
| 213 | try: |
| 214 | cmd = cls(args) |
| 215 | return cmd.run() |
| 216 | except QMPError as err: |
| 217 | print(f"{type(err).__name__}: {err!s}", file=sys.stderr) |
| 218 | return -1 |
| 219 | |
| 220 | @classmethod |
| 221 | def entry_point(cls) -> int: |
| 222 | """ |
| 223 | Build this command's parser, parse arguments, and run the command. |
| 224 | |
| 225 | :return: `run`'s return code. |
| 226 | """ |
| 227 | parser = argparse.ArgumentParser(description=cls.help) |
| 228 | cls.configure_parser(parser) |
| 229 | args = parser.parse_args() |
| 230 | return cls.command_runner(args) |