openkylin
/
qemu
mirror of https://gitee.com/openkylin/qemu.git
9
0
Fork 0
Code Issues Projects Releases Wiki Activity

python/machine.py: Fix monitor address typing 

Prior to this, it's difficult for mypy to intuit what the concrete type
of the monitor address is; it has difficulty inferring the type across
two variables.

Create _monitor_address as a property that always returns a valid
address to simplify static type analysis.

To preserve our ability to clean up, use a simple boolean to indicate
whether or not we should try to clean up the sock file after execution.

Signed-off-by: John Snow <jsnow@redhat.com>
Reviewed-by: Kevin Wolf <kwolf@redhat.com>
Message-id: 20201006235817.3280413-3-jsnow@redhat.com
Signed-off-by: John Snow <jsnow@redhat.com>
This commit is contained in:
John Snow 2020-10-06 19:57:59 -04:00
parent 932ca4bbde
commit c4e6023f05
1 changed files with 25 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
python/qemu/machine.py
View File

@ -28,6 +28,7 @@
from typing import Optional, Type
from . import console_socket, qmp
from .qmp import SocketAddrT
LOG = logging.getLogger(__name__)
@ -68,7 +69,8 @@ class QEMUMachine:
"""
def __init__(self, binary, args=None, wrapper=None, name=None,
test_dir="/var/tmp", monitor_address=None,
test_dir="/var/tmp",
monitor_address: Optional[SocketAddrT] = None,
socket_scm_helper=None, sock_dir=None,
drain_console=False, console_log=None):
'''
@ -95,8 +97,14 @@ def __init__(self, binary, args=None, wrapper=None, name=None,
if sock_dir is None:
sock_dir = test_dir
self._name = name
self._monitor_address = monitor_address
self._vm_monitor = None
if monitor_address is not None:
self._monitor_address = monitor_address
self._remove_monitor_sockfile = False
else:
self._monitor_address = os.path.join(
sock_dir, f"{name}-monitor.sock"
)
self._remove_monitor_sockfile = True
self._qemu_log_path = None
self._qemu_log_file = None
self._popen = None
@ -241,15 +249,17 @@ def _load_io_log(self):
def _base_args(self):
args = ['-display', 'none', '-vga', 'none']
if self._qmp_set:
if isinstance(self._monitor_address, tuple):
moncdev = "socket,id=mon,host=%s,port=%s" % (
self._monitor_address[0],
self._monitor_address[1])
moncdev = "socket,id=mon,host={},port={}".format(
*self._monitor_address
)
else:
moncdev = 'socket,id=mon,path=%s' % self._vm_monitor
moncdev = f"socket,id=mon,path={self._monitor_address}"
args.extend(['-chardev', moncdev, '-mon',
'chardev=mon,mode=control'])
if self._machine is not None:
args.extend(['-machine', self._machine])
for _ in range(self._console_index):
@ -274,14 +284,14 @@ def _pre_launch(self):
self._qemu_log_file = open(self._qemu_log_path, 'wb')
if self._qmp_set:
if self._monitor_address is not None:
self._vm_monitor = self._monitor_address
else:
self._vm_monitor = os.path.join(self._sock_dir,
self._name + "-monitor.sock")
self._remove_files.append(self._vm_monitor)
self._qmp = qmp.QEMUMonitorProtocol(self._vm_monitor, server=True,
nickname=self._name)
if self._remove_monitor_sockfile:
assert isinstance(self._monitor_address, str)
self._remove_files.append(self._monitor_address)
self._qmp = qmp.QEMUMonitorProtocol(
self._monitor_address,
server=True,
nickname=self._name
)
def _post_launch(self):
if self._qmp: