This patch adds type annotations for aqmp-tui using the mypy library. Signed-off-by: G S Niteesh Babu <niteesh...@gmail.com> --- python/qemu/aqmp/aqmp_tui.py | 79 ++++++++++++++++++++---------------- python/setup.cfg | 3 -- 2 files changed, 43 insertions(+), 39 deletions(-)
diff --git a/python/qemu/aqmp/aqmp_tui.py b/python/qemu/aqmp/aqmp_tui.py index ec9eba0aa7..ab9ada793a 100644 --- a/python/qemu/aqmp/aqmp_tui.py +++ b/python/qemu/aqmp/aqmp_tui.py @@ -9,8 +9,15 @@ import argparse import asyncio import logging -from logging import Handler +from logging import Handler, LogRecord import signal +from typing import ( + Any, + List, + Optional, + Tuple, + Union, +) import urwid import urwid_readline @@ -22,13 +29,13 @@ from .util import create_task, pretty_traceback -UPDATE_MSG = 'UPDATE_MSG' +UPDATE_MSG: str = 'UPDATE_MSG' # Using root logger to enable all loggers under qemu and asyncio LOGGER = logging.getLogger() -def format_json(msg): +def format_json(msg: str) -> str: """ Formats given multiline JSON message into a single line message. Converting into single line is more asthetically pleasing when looking @@ -43,17 +50,17 @@ def format_json(msg): class App(QMPClient): - def __init__(self, address): + def __init__(self, address: Union[str, Tuple[str, int]]) -> None: urwid.register_signal(type(self), UPDATE_MSG) self.window = Window(self) self.address = address - self.aloop = None + self.aloop: Optional[Any] = None # FIXME: Use more concrete type. super().__init__() - def add_to_history(self, msg): + def add_to_history(self, msg: str) -> None: urwid.emit_signal(self, UPDATE_MSG, msg) - def _cb_outbound(self, msg): + def _cb_outbound(self, msg: Message) -> Message: # FIXME: I think the ideal way to omit these messages during in-TUI # logging will be to add a filter to the logger. We can use regex to # filter out messages starting with 'Request:' or 'Response:' but I @@ -67,25 +74,25 @@ def _cb_outbound(self, msg): self.add_to_history('<-- ' + str(msg)) return msg - def _cb_inbound(self, msg): + def _cb_inbound(self, msg: Message) -> Message: handler = LOGGER.handlers[0] if not isinstance(handler, TUILogHandler): LOGGER.debug('Response: %s', str(msg)) self.add_to_history('--> ' + str(msg)) return msg - async def wait_for_events(self): + async def wait_for_events(self) -> None: async for event in self.events: self.handle_event(event) - async def _send_to_server(self, raw_msg): + async def _send_to_server(self, raw_msg: str) -> None: # FIXME: Format the raw_msg in history view to one line. It is not # pleasing to see multiple lines JSON object with an error statement. try: msg = Message(bytes(raw_msg, encoding='utf-8')) # Format multiline json into a single line JSON, since it is more # pleasing to look along with err message in TUI. - raw_msg = self.format_json(raw_msg) + raw_msg = format_json(raw_msg) await self._raw(msg, assign_id='id' not in msg) except (ValueError, TypeError) as err: LOGGER.info('Invalid message: %s', str(err)) @@ -102,18 +109,18 @@ def _cb_inbound(self, msg): LOGGER.error('Exception from _send_to_server: %s', str(err)) raise err - def cb_send_to_server(self, msg): + def cb_send_to_server(self, msg: str) -> None: create_task(self._send_to_server(msg)) - def unhandled_input(self, key): + def unhandled_input(self, key: str) -> None: if key == 'esc': self.kill_app() - def kill_app(self): + def kill_app(self) -> None: # TODO: Work on the disconnect logic create_task(self._kill_app()) - async def _kill_app(self): + async def _kill_app(self) -> None: # It is ok to call disconnect even in disconnect state try: await self.disconnect() @@ -124,7 +131,7 @@ def kill_app(self): raise err raise urwid.ExitMainLoop() - def handle_event(self, event): + def handle_event(self, event: Message) -> None: # FIXME: Consider all states present in qapi/run-state.json if event['event'] == 'SHUTDOWN': self._set_status('Server shutdown') @@ -139,7 +146,7 @@ def _get_formatted_address(self) -> str: addr = f'{host}:{port}' return addr - async def connect_server(self): + async def connect_server(self) -> None: try: await self.connect(self.address) addr = self._get_formatted_address() @@ -148,7 +155,7 @@ def _get_formatted_address(self) -> str: LOGGER.info('connect_server: ConnectError %s', str(err)) self._set_status('Server shutdown') - def run(self, debug=False): + def run(self, debug: bool = False) -> None: self.aloop = asyncio.get_event_loop() self.aloop.set_debug(debug) @@ -176,7 +183,7 @@ class StatusBar(urwid.Text): """ A simple Text widget that currently only shows connection status. """ - def __init__(self, text=''): + def __init__(self, text: str = ''): super().__init__(text, align='right') @@ -185,14 +192,14 @@ class Editor(urwid_readline.ReadlineEdit): Support urwid_readline features along with history support which lacks in urwid_readline """ - def __init__(self, master): + def __init__(self, master: App) -> None: super().__init__(caption='> ', multiline=True) self.master = master - self.history = [] - self.last_index = -1 - self.show_history = False + self.history: List[str] = [] + self.last_index: int = -1 + self.show_history: bool = False - def keypress(self, size, key): + def keypress(self, size: Tuple[int, int], key: str) -> Optional[str]: # TODO: Add some logic for down key and clean up logic if possible. # Returning None means the key has been handled by this widget # which otherwise is propogated to the parent widget to be @@ -223,7 +230,7 @@ def keypress(self, size, key): else: self.show_history = False self.last_index = 0 - return super().keypress(size, key) + return super().keypress(size, key) # type: ignore return None @@ -231,7 +238,7 @@ class EditorWidget(urwid.Filler): """ Wraps CustomEdit """ - def __init__(self, master): + def __init__(self, master: App) -> None: super().__init__(Editor(master), valign='top') @@ -239,12 +246,12 @@ class HistoryBox(urwid.ListBox): """ Shows all the QMP message transmitted/received """ - def __init__(self, master): + def __init__(self, master: App) -> None: self.master = master self.history = urwid.SimpleFocusListWalker([]) super().__init__(self.history) - def add_to_history(self, history): + def add_to_history(self, history: str) -> None: self.history.append(urwid.Text(history)) if self.history: self.history.set_focus(len(self.history) - 1) @@ -254,7 +261,7 @@ class HistoryWindow(urwid.Frame): """ Composes the HistoryBox and EditorWidget """ - def __init__(self, master): + def __init__(self, master: App) -> None: self.master = master self.editor_widget = EditorWidget(master) self.editor = urwid.LineBox(self.editor_widget) @@ -264,7 +271,7 @@ def __init__(self, master): super().__init__(self.body) urwid.connect_signal(self.master, UPDATE_MSG, self.cb_add_to_history) - def cb_add_to_history(self, msg): + def cb_add_to_history(self, msg: str) -> None: self.history.add_to_history(msg) @@ -275,7 +282,7 @@ class Window(urwid.Frame): future when we will have multiple windows and want to the switch between them and display overlays """ - def __init__(self, master): + def __init__(self, master: App) -> None: self.master = master footer = StatusBar() body = HistoryWindow(master) @@ -283,18 +290,18 @@ def __init__(self, master): class TUILogHandler(Handler): - def __init__(self, tui): + def __init__(self, tui: App) -> None: super().__init__() self.tui = tui - def emit(self, record): + def emit(self, record: LogRecord) -> None: level = record.levelname msg = record.getMessage() msg = f'[{level}]: {msg}' self.tui.add_to_history(msg) -def main(): +def main() -> None: parser = argparse.ArgumentParser(description='AQMP TUI') parser.add_argument('qmp_server', help='Address of the QMP server' '< UNIX socket path | TCP addr:port >') @@ -311,7 +318,7 @@ def main(): try: address = QEMUMonitorProtocol.parse_address(args.qmp_server) except QMPBadPortError as err: - parser.error(err) + parser.error(str(err)) app = App(address) @@ -330,4 +337,4 @@ def main(): if __name__ == '__main__': - main() # type: ignore + main() diff --git a/python/setup.cfg b/python/setup.cfg index 8cd9ac0d81..11c6240aba 100644 --- a/python/setup.cfg +++ b/python/setup.cfg @@ -83,9 +83,6 @@ namespace_packages = True allow_subclassing_any = True [mypy-qemu.aqmp.aqmp_tui] -disallow_untyped_defs = False -disallow_incomplete_defs = False -check_untyped_defs = False # urwid and urwid_readline have no type stubs: allow_subclassing_any = True -- 2.17.1