scripts/qmp-shell: add mypy types

As per my usual, this patch is annotations only. Any changes with side
effects are done elsewhere.

Note: pylint does not understand the subscripts for Collection in Python 3.6,
so use the stronger Sequence type as a workaround.
Signed-off-by: John Snow <jsnow@redhat.com>
Message-id: 20210607200649.1840382-28-jsnow@redhat.com
Signed-off-by: John Snow <jsnow@redhat.com>
This commit is contained in:
John Snow 2021-06-07 16:06:34 -04:00
parent 1acde76328
commit 1eab88723c
1 changed files with 41 additions and 26 deletions

View File

@ -72,10 +72,18 @@ import os
import re import re
import readline import readline
import sys import sys
from typing import (
Iterator,
List,
NoReturn,
Optional,
Sequence,
)
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'python')) sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'python'))
from qemu import qmp from qemu import qmp
from qemu.qmp import QMPMessage
class QMPCompleter: class QMPCompleter:
@ -122,25 +130,26 @@ class FuzzyJSON(ast.NodeTransformer):
# TODO: QMPShell's interface is a bit ugly (eg. _fill_completion() and # TODO: QMPShell's interface is a bit ugly (eg. _fill_completion() and
# _execute_cmd()). Let's design a better one. # _execute_cmd()). Let's design a better one.
class QMPShell(qmp.QEMUMonitorProtocol): class QMPShell(qmp.QEMUMonitorProtocol):
def __init__(self, address, pretty=False, verbose=False): def __init__(self, address: str, pretty: bool = False,
verbose: bool = False):
super().__init__(self.parse_address(address)) super().__init__(self.parse_address(address))
self._greeting = None self._greeting: Optional[QMPMessage] = None
self._completer = QMPCompleter() self._completer = QMPCompleter()
self._pretty = pretty self._pretty = pretty
self._transmode = False self._transmode = False
self._actions = list() self._actions: List[QMPMessage] = []
self._histfile = os.path.join(os.path.expanduser('~'), self._histfile = os.path.join(os.path.expanduser('~'),
'.qmp-shell_history') '.qmp-shell_history')
self.verbose = verbose self.verbose = verbose
def _fill_completion(self): def _fill_completion(self) -> None:
cmds = self.cmd('query-commands') cmds = self.cmd('query-commands')
if 'error' in cmds: if 'error' in cmds:
return return
for cmd in cmds['return']: for cmd in cmds['return']:
self._completer.append(cmd['name']) self._completer.append(cmd['name'])
def __completer_setup(self): def __completer_setup(self) -> None:
self._completer = QMPCompleter() self._completer = QMPCompleter()
self._fill_completion() self._fill_completion()
readline.set_history_length(1024) readline.set_history_length(1024)
@ -157,14 +166,14 @@ class QMPShell(qmp.QEMUMonitorProtocol):
print(f"Failed to read history '{self._histfile}': {err!s}") print(f"Failed to read history '{self._histfile}': {err!s}")
atexit.register(self.__save_history) atexit.register(self.__save_history)
def __save_history(self): def __save_history(self) -> None:
try: try:
readline.write_history_file(self._histfile) readline.write_history_file(self._histfile)
except IOError as err: except IOError as err:
print(f"Failed to save history file '{self._histfile}': {err!s}") print(f"Failed to save history file '{self._histfile}': {err!s}")
@classmethod @classmethod
def __parse_value(cls, val): def __parse_value(cls, val: str) -> object:
try: try:
return int(val) return int(val)
except ValueError: except ValueError:
@ -189,7 +198,9 @@ class QMPShell(qmp.QEMUMonitorProtocol):
pass pass
return val return val
def __cli_expr(self, tokens, parent): def __cli_expr(self,
tokens: Sequence[str],
parent: qmp.QMPObject) -> None:
for arg in tokens: for arg in tokens:
(key, sep, val) = arg.partition('=') (key, sep, val) = arg.partition('=')
if sep != '=': if sep != '=':
@ -215,7 +226,7 @@ class QMPShell(qmp.QEMUMonitorProtocol):
raise QMPShellError(f'Cannot set "{key}" multiple times') raise QMPShellError(f'Cannot set "{key}" multiple times')
parent[optpath[-1]] = value parent[optpath[-1]] = value
def __build_cmd(self, cmdline): def __build_cmd(self, cmdline: str) -> Optional[QMPMessage]:
""" """
Build a QMP input object from a user provided command-line in the Build a QMP input object from a user provided command-line in the
following format: following format:
@ -224,6 +235,7 @@ class QMPShell(qmp.QEMUMonitorProtocol):
""" """
argument_regex = r'''(?:[^\s"']|"(?:\\.|[^"])*"|'(?:\\.|[^'])*')+''' argument_regex = r'''(?:[^\s"']|"(?:\\.|[^"])*"|'(?:\\.|[^'])*')+'''
cmdargs = re.findall(argument_regex, cmdline) cmdargs = re.findall(argument_regex, cmdline)
qmpcmd: QMPMessage
# Transactional CLI entry/exit: # Transactional CLI entry/exit:
if cmdargs[0] == 'transaction(': if cmdargs[0] == 'transaction(':
@ -261,14 +273,14 @@ class QMPShell(qmp.QEMUMonitorProtocol):
self.__cli_expr(cmdargs[1:], qmpcmd['arguments']) self.__cli_expr(cmdargs[1:], qmpcmd['arguments'])
return qmpcmd return qmpcmd
def _print(self, qmp_message): def _print(self, qmp_message: object) -> None:
indent = None indent = None
if self._pretty: if self._pretty:
indent = 4 indent = 4
jsobj = json.dumps(qmp_message, indent=indent, sort_keys=self._pretty) jsobj = json.dumps(qmp_message, indent=indent, sort_keys=self._pretty)
print(str(jsobj)) print(str(jsobj))
def _execute_cmd(self, cmdline): def _execute_cmd(self, cmdline: str) -> bool:
try: try:
qmpcmd = self.__build_cmd(cmdline) qmpcmd = self.__build_cmd(cmdline)
except Exception as err: except Exception as err:
@ -288,11 +300,12 @@ class QMPShell(qmp.QEMUMonitorProtocol):
self._print(resp) self._print(resp)
return True return True
def connect(self, negotiate: bool = True): def connect(self, negotiate: bool = True) -> None:
self._greeting = super().connect(negotiate) self._greeting = super().connect(negotiate)
self.__completer_setup() self.__completer_setup()
def show_banner(self, msg='Welcome to the QMP low-level shell!'): def show_banner(self,
msg: str = 'Welcome to the QMP low-level shell!') -> None:
print(msg) print(msg)
if not self._greeting: if not self._greeting:
print('Connected') print('Connected')
@ -301,12 +314,12 @@ class QMPShell(qmp.QEMUMonitorProtocol):
print("Connected to QEMU {major}.{minor}.{micro}\n".format(**version)) print("Connected to QEMU {major}.{minor}.{micro}\n".format(**version))
@property @property
def prompt(self): def prompt(self) -> str:
if self._transmode: if self._transmode:
return 'TRANS> ' return 'TRANS> '
return '(QEMU) ' return '(QEMU) '
def read_exec_command(self): def read_exec_command(self) -> bool:
""" """
Read and execute a command. Read and execute a command.
@ -326,7 +339,7 @@ class QMPShell(qmp.QEMUMonitorProtocol):
return self._execute_cmd(cmdline) return self._execute_cmd(cmdline)
def repl(self): def repl(self) -> Iterator[None]:
self.show_banner() self.show_banner()
while self.read_exec_command(): while self.read_exec_command():
yield yield
@ -334,11 +347,12 @@ class QMPShell(qmp.QEMUMonitorProtocol):
class HMPShell(QMPShell): class HMPShell(QMPShell):
def __init__(self, address, pretty=False, verbose=False): def __init__(self, address: str,
pretty: bool = False, verbose: bool = False):
super().__init__(address, pretty, verbose) super().__init__(address, pretty, verbose)
self.__cpu_index = 0 self.__cpu_index = 0
def __cmd_completion(self): def __cmd_completion(self) -> None:
for cmd in self.__cmd_passthrough('help')['return'].split('\r\n'): for cmd in self.__cmd_passthrough('help')['return'].split('\r\n'):
if cmd and cmd[0] != '[' and cmd[0] != '\t': if cmd and cmd[0] != '[' and cmd[0] != '\t':
name = cmd.split()[0] # drop help text name = cmd.split()[0] # drop help text
@ -355,21 +369,22 @@ class HMPShell(QMPShell):
self._completer.append(name) self._completer.append(name)
self._completer.append('help ' + name) # help completion self._completer.append('help ' + name) # help completion
def __info_completion(self): def __info_completion(self) -> None:
for cmd in self.__cmd_passthrough('info')['return'].split('\r\n'): for cmd in self.__cmd_passthrough('info')['return'].split('\r\n'):
if cmd: if cmd:
self._completer.append('info ' + cmd.split()[1]) self._completer.append('info ' + cmd.split()[1])
def __other_completion(self): def __other_completion(self) -> None:
# special cases # special cases
self._completer.append('help info') self._completer.append('help info')
def _fill_completion(self): def _fill_completion(self) -> None:
self.__cmd_completion() self.__cmd_completion()
self.__info_completion() self.__info_completion()
self.__other_completion() self.__other_completion()
def __cmd_passthrough(self, cmdline, cpu_index=0): def __cmd_passthrough(self, cmdline: str,
cpu_index: int = 0) -> QMPMessage:
return self.cmd_obj({ return self.cmd_obj({
'execute': 'human-monitor-command', 'execute': 'human-monitor-command',
'arguments': { 'arguments': {
@ -378,7 +393,7 @@ class HMPShell(QMPShell):
} }
}) })
def _execute_cmd(self, cmdline): def _execute_cmd(self, cmdline: str) -> bool:
if cmdline.split()[0] == "cpu": if cmdline.split()[0] == "cpu":
# trap the cpu command, it requires special setting # trap the cpu command, it requires special setting
try: try:
@ -404,16 +419,16 @@ class HMPShell(QMPShell):
print('%s: %s' % (resp['error']['class'], resp['error']['desc'])) print('%s: %s' % (resp['error']['class'], resp['error']['desc']))
return True return True
def show_banner(self, msg='Welcome to the HMP shell!'): def show_banner(self, msg: str = 'Welcome to the HMP shell!') -> None:
QMPShell.show_banner(self, msg) QMPShell.show_banner(self, msg)
def die(msg): def die(msg: str) -> NoReturn:
sys.stderr.write('ERROR: %s\n' % msg) sys.stderr.write('ERROR: %s\n' % msg)
sys.exit(1) sys.exit(1)
def main(): def main() -> None:
parser = argparse.ArgumentParser() parser = argparse.ArgumentParser()
parser.add_argument('-H', '--hmp', action='store_true', parser.add_argument('-H', '--hmp', action='store_true',
help='Use HMP interface') help='Use HMP interface')