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 68CommandT = TypeVar('CommandT', bound='QOMCommand') 69 70 71class QOMCommand: 72 """ 73 Represents a QOM sub-command. 74 75 :param args: Parsed arguments, as returned from parser.parse_args. 76 """ 77 name: str 78 help: str 79 80 def __init__(self, args: argparse.Namespace): 81 if args.socket is None: 82 raise QMPError("No QMP socket path or address given") 83 self.qmp = QEMUMonitorProtocol( 84 QEMUMonitorProtocol.parse_address(args.socket) 85 ) 86 self.qmp.connect() 87 88 @classmethod 89 def register(cls, subparsers: Any) -> None: 90 """ 91 Register this command with the argument parser. 92 93 :param subparsers: argparse subparsers object, from "add_subparsers". 94 """ 95 subparser = subparsers.add_parser(cls.name, help=cls.help, 96 description=cls.help) 97 cls.configure_parser(subparser) 98 99 @classmethod 100 def configure_parser(cls, parser: argparse.ArgumentParser) -> None: 101 """ 102 Configure a parser with this command's arguments. 103 104 :param parser: argparse parser or subparser object. 105 """ 106 default_path = os.environ.get('QMP_SOCKET') 107 parser.add_argument( 108 '--socket', '-s', 109 dest='socket', 110 action='store', 111 help='QMP socket path or address (addr:port).' 112 ' May also be set via QMP_SOCKET environment variable.', 113 default=default_path 114 ) 115 parser.set_defaults(cmd_class=cls) 116 117 @classmethod 118 def add_path_prop_arg(cls, parser: argparse.ArgumentParser) -> None: 119 """ 120 Add the <path>.<proptery> positional argument to this command. 121 122 :param parser: The parser to add the argument to. 123 """ 124 parser.add_argument( 125 'path_prop', 126 metavar='<path>.<property>', 127 action='store', 128 help="QOM path and property, separated by a period '.'" 129 ) 130 131 def run(self) -> int: 132 """ 133 Run this command. 134 135 :return: 0 on success, 1 otherwise. 136 """ 137 raise NotImplementedError 138 139 def qom_list(self, path: str) -> List[ObjectPropertyInfo]: 140 """ 141 :return: a strongly typed list from the 'qom-list' command. 142 """ 143 rsp = self.qmp.cmd('qom-list', path=path) 144 # qom-list returns List[ObjectPropertyInfo] 145 assert isinstance(rsp, list) 146 return [ObjectPropertyInfo.make(x) for x in rsp] 147 148 @classmethod 149 def command_runner( 150 cls: Type[CommandT], 151 args: argparse.Namespace 152 ) -> int: 153 """ 154 Run a fully-parsed subcommand, with error-handling for the CLI. 155 156 :return: The return code from `run()`. 157 """ 158 try: 159 cmd = cls(args) 160 return cmd.run() 161 except QMPError as err: 162 print(f"{type(err).__name__}: {err!s}", file=sys.stderr) 163 return -1 164 165 @classmethod 166 def entry_point(cls) -> int: 167 """ 168 Build this command's parser, parse arguments, and run the command. 169 170 :return: `run`'s return code. 171 """ 172 parser = argparse.ArgumentParser(description=cls.help) 173 cls.configure_parser(parser) 174 args = parser.parse_args() 175 return cls.command_runner(args) 176