1""" 2QOM 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 18import argparse 19import os 20import sys 21from typing import ( 22 Any, 23 Dict, 24 List, 25 Optional, 26 Type, 27 TypeVar, 28) 29 30from qemu.qmp import QMPError 31from qemu.qmp.legacy import QEMUMonitorProtocol 32 33 34class 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 68class 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 92class 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 114CommandT = TypeVar('CommandT', bound='QOMCommand') 115 116 117class 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) 231