motorhead/qemu
1
0
Fork 0
mirror of https://github.com/Motorhead1991/qemu.git synced 2025-08-04 16:23:55 -06:00
Code Issues Projects Releases Packages Wiki Activity Actions

python/qemu: Add mypy type annotations

Browse source 
These should all be purely annotations with no changes in behavior at
all. You need to be in the python folder, but you should be able to
confirm that these annotations are correct (or at least self-consistent)
by running `mypy --strict qemu`.

Signed-off-by: John Snow <jsnow@redhat.com>
Reviewed-by: Kevin Wolf <kwolf@redhat.com>
Message-id: 20201006235817.3280413-12-jsnow@redhat.com
Signed-off-by: John Snow <jsnow@redhat.com>
This commit is contained in:
John Snow 2020-10-06 19:58:08 -04:00
parent 090744d552
commit f12a282ff4
4 changed files with 101 additions and 75 deletions
Download patch file Download diff file Expand all files Collapse all files

98
python/qemu/machine.py
View file

@ -23,11 +23,13 @@ import logging
import os
import shutil
import signal
import socket
import subprocess
import tempfile
from types import TracebackType
from typing import (
Any,
BinaryIO,
Dict,
List,
Optional,
@ -37,7 +39,7 @@ from typing import (
)
from . import console_socket, qmp
from .qmp import QMPMessage, SocketAddrT
from .qmp import QMPMessage, QMPReturnValue, SocketAddrT
LOG = logging.getLogger(__name__)
@ -67,7 +69,7 @@ class AbnormalShutdown(QEMUMachineError):
class QEMUMachine:
"""
A QEMU VM
A QEMU VM.
Use this object as a context manager to ensure
the QEMU process terminates::
@ -84,8 +86,10 @@ class QEMUMachine:
name: Optional[str] = None,
test_dir: str = "/var/tmp",
monitor_address: Optional[SocketAddrT] = None,
socket_scm_helper=None, sock_dir=None,
drain_console=False, console_log=None):
socket_scm_helper: Optional[str] = None,
sock_dir: Optional[str] = None,
drain_console: bool = False,
console_log: Optional[str] = None):
'''
Initialize a QEMUMachine
@ -129,28 +133,28 @@ class QEMUMachine:
self._drain_console = drain_console
# Runstate
self._qemu_log_path = None
self._qemu_log_file = None
self._qemu_log_path: Optional[str] = None
self._qemu_log_file: Optional[BinaryIO] = None
self._popen: Optional['subprocess.Popen[bytes]'] = None
self._events = []
self._iolog = None
self._events: List[QMPMessage] = []
self._iolog: Optional[str] = None
self._qmp_set = True # Enable QMP monitor by default.
self._qmp_connection: Optional[qmp.QEMUMonitorProtocol] = None
self._qemu_full_args: Tuple[str, ...] = ()
self._temp_dir = None
self._temp_dir: Optional[str] = None
self._launched = False
self._machine = None
self._machine: Optional[str] = None
self._console_index = 0
self._console_set = False
self._console_device_type = None
self._console_device_type: Optional[str] = None
self._console_address = os.path.join(
self._sock_dir, f"{self._name}-console.sock"
)
self._console_socket = None
self._remove_files = []
self._console_socket: Optional[socket.socket] = None
self._remove_files: List[str] = []
self._user_killed = False
def __enter__(self):
def __enter__(self) -> 'QEMUMachine':
return self
def __exit__(self,
@ -159,14 +163,15 @@ class QEMUMachine:
exc_tb: Optional[TracebackType]) -> None:
self.shutdown()
def add_monitor_null(self):
def add_monitor_null(self) -> None:
"""
This can be used to add an unused monitor instance.
"""
self._args.append('-monitor')
self._args.append('null')
def add_fd(self, fd, fdset, opaque, opts=''):
def add_fd(self, fd: int, fdset: int,
opaque: str, opts: str = '') -> 'QEMUMachine':
"""
Pass a file descriptor to the VM
"""
@ -185,7 +190,8 @@ class QEMUMachine:
self._args.append(','.join(options))
return self
def send_fd_scm(self, fd=None, file_path=None):
def send_fd_scm(self, fd: Optional[int] = None,
file_path: Optional[str] = None) -> int:
"""
Send an fd or file_path to socket_scm_helper.
@ -229,7 +235,7 @@ class QEMUMachine:
return proc.returncode
@staticmethod
def _remove_if_exists(path):
def _remove_if_exists(path: str) -> None:
"""
Remove file object at path if it exists
"""
@ -240,7 +246,7 @@ class QEMUMachine:
return
raise
def is_running(self):
def is_running(self) -> bool:
"""Returns true if the VM is running."""
return self._popen is not None and self._popen.poll() is None
@ -250,19 +256,19 @@ class QEMUMachine:
raise QEMUMachineError('Subprocess pipe not present')
return self._popen
def exitcode(self):
def exitcode(self) -> Optional[int]:
"""Returns the exit code if possible, or None."""
if self._popen is None:
return None
return self._popen.poll()
def get_pid(self):
def get_pid(self) -> Optional[int]:
"""Returns the PID of the running process, or None."""
if not self.is_running():
return None
return self._subp.pid
def _load_io_log(self):
def _load_io_log(self) -> None:
if self._qemu_log_path is not None:
with open(self._qemu_log_path, "r") as iolog:
self._iolog = iolog.read()
@ -296,7 +302,7 @@ class QEMUMachine:
args.extend(['-device', device])
return args
def _pre_launch(self):
def _pre_launch(self) -> None:
self._temp_dir = tempfile.mkdtemp(dir=self._test_dir)
self._qemu_log_path = os.path.join(self._temp_dir, self._name + ".log")
self._qemu_log_file = open(self._qemu_log_path, 'wb')
@ -314,11 +320,11 @@ class QEMUMachine:
nickname=self._name
)
def _post_launch(self):
def _post_launch(self) -> None:
if self._qmp_connection:
self._qmp.accept()
def _post_shutdown(self):
def _post_shutdown(self) -> None:
"""
Called to cleanup the VM instance after the process has exited.
May also be called after a failed launch.
@ -358,7 +364,7 @@ class QEMUMachine:
self._user_killed = False
self._launched = False
def launch(self):
def launch(self) -> None:
"""
Launch the VM and make sure we cleanup and expose the
command line/output in case of exception
@ -382,7 +388,7 @@ class QEMUMachine:
LOG.debug('Output: %r', self._iolog)
raise
def _launch(self):
def _launch(self) -> None:
"""
Launch the VM and establish a QMP connection
"""
@ -501,7 +507,7 @@ class QEMUMachine:
finally:
self._post_shutdown()
def kill(self):
def kill(self) -> None:
"""
Terminate the VM forcefully, wait for it to exit, and perform cleanup.
"""
@ -516,7 +522,7 @@ class QEMUMachine:
"""
self.shutdown(has_quit=True, timeout=timeout)
def set_qmp_monitor(self, enabled=True):
def set_qmp_monitor(self, enabled: bool = True) -> None:
"""
Set the QMP monitor.
@ -552,7 +558,9 @@ class QEMUMachine:
qmp_args = self._qmp_args(conv_keys, **args)
return self._qmp.cmd(cmd, args=qmp_args)
def command(self, cmd, conv_keys=True, **args):
def command(self, cmd: str,
conv_keys: bool = True,
**args: Any) -> QMPReturnValue:
"""
Invoke a QMP command.
On success return the response dict.
@ -561,7 +569,7 @@ class QEMUMachine:
qmp_args = self._qmp_args(conv_keys, **args)
return self._qmp.command(cmd, **qmp_args)
def get_qmp_event(self, wait=False):
def get_qmp_event(self, wait: bool = False) -> Optional[QMPMessage]:
"""
Poll for one queued QMP events and return it
"""
@ -569,7 +577,7 @@ class QEMUMachine:
return self._events.pop(0)
return self._qmp.pull_event(wait=wait)
def get_qmp_events(self, wait=False):
def get_qmp_events(self, wait: bool = False) -> List[QMPMessage]:
"""
Poll for queued QMP events and return a list of dicts
"""
@ -580,7 +588,7 @@ class QEMUMachine:
return events
@staticmethod
def event_match(event, match=None):
def event_match(event: Any, match: Optional[Any]) -> bool:
"""
Check if an event matches optional match criteria.
@ -610,9 +618,11 @@ class QEMUMachine:
return True
except TypeError:
# either match or event wasn't iterable (not a dict)
return match == event
return bool(match == event)
def event_wait(self, name, timeout=60.0, match=None):
def event_wait(self, name: str,
timeout: float = 60.0,
match: Optional[QMPMessage] = None) -> Optional[QMPMessage]:
"""
event_wait waits for and returns a named event from QMP with a timeout.
@ -622,7 +632,9 @@ class QEMUMachine:
"""
return self.events_wait([(name, match)], timeout)
def events_wait(self, events, timeout=60.0):
def events_wait(self,
events: Sequence[Tuple[str, Any]],
timeout: float = 60.0) -> Optional[QMPMessage]:
"""
events_wait waits for and returns a single named event from QMP.
In the case of multiple qualifying events, this function returns the
@ -639,7 +651,7 @@ class QEMUMachine:
:return: A QMP event matching the filter criteria.
If timeout was 0 and no event matched, None.
"""
def _match(event):
def _match(event: QMPMessage) -> bool:
for name, match in events:
if event['event'] == name and self.event_match(event, match):
return True
@ -666,20 +678,20 @@ class QEMUMachine:
return None
def get_log(self):
def get_log(self) -> Optional[str]:
"""
After self.shutdown or failed qemu execution, this returns the output
of the qemu process.
"""
return self._iolog
def add_args(self, *args):
def add_args(self, *args: str) -> None:
"""
Adds to the list of extra arguments to be given to the QEMU binary
"""
self._args.extend(args)
def set_machine(self, machine_type):
def set_machine(self, machine_type: str) -> None:
"""
Sets the machine type
@ -688,7 +700,9 @@ class QEMUMachine:
"""
self._machine = machine_type
def set_console(self, device_type=None, console_index=0):
def set_console(self,
device_type: Optional[str] = None,
console_index: int = 0) -> None:
"""
Sets the device type for a console device
@ -719,7 +733,7 @@ class QEMUMachine:
self._console_index = console_index
@property
def console_socket(self):
def console_socket(self) -> socket.socket:
"""
Returns a socket connected to the console
"""