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>

1 files changed, 41 insertions, 26 deletions

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