python/qemu: make 'args' style arguments immutable

These arguments don't need to be mutable and aren't really used as
such. Clarify their types as immutable and adjust code to match where
necessary.

In general, It's probably best not to accept a user-defined mutable
object and store it as internal object state unless there's a strong
justification for doing so. Instead, try to use generic types as input
with empty tuples as the default, and coerce to list where necessary.

Signed-off-by: John Snow <jsnow@redhat.com>
Reviewed-by: Kevin Wolf <kwolf@redhat.com>
Reviewed-by: Philippe Mathieu-Daudé <philmd@redhat.com>
Message-id: 20201006235817.3280413-10-jsnow@redhat.com
Signed-off-by: John Snow <jsnow@redhat.com>
This commit is contained in:
John Snow 2020-10-06 19:58:06 -04:00
parent 9223fda464
commit aad3f3bb6c
2 changed files with 34 additions and 18 deletions

View File

@ -18,6 +18,7 @@
# #
import errno import errno
from itertools import chain
import logging import logging
import os import os
import shutil import shutil
@ -30,6 +31,8 @@
Dict, Dict,
List, List,
Optional, Optional,
Sequence,
Tuple,
Type, Type,
) )
@ -74,8 +77,12 @@ class QEMUMachine:
# vm is guaranteed to be shut down here # vm is guaranteed to be shut down here
""" """
def __init__(self, binary, args=None, wrapper=None, name=None, def __init__(self,
test_dir="/var/tmp", binary: str,
args: Sequence[str] = (),
wrapper: Sequence[str] = (),
name: Optional[str] = None,
test_dir: str = "/var/tmp",
monitor_address: Optional[SocketAddrT] = None, monitor_address: Optional[SocketAddrT] = None,
socket_scm_helper=None, sock_dir=None, socket_scm_helper=None, sock_dir=None,
drain_console=False, console_log=None): drain_console=False, console_log=None):
@ -97,14 +104,7 @@ def __init__(self, binary, args=None, wrapper=None, name=None,
# Direct user configuration # Direct user configuration
self._binary = binary self._binary = binary
if args is None:
args = []
# Copy mutable input: we will be modifying our copy
self._args = list(args) self._args = list(args)
if wrapper is None:
wrapper = []
self._wrapper = wrapper self._wrapper = wrapper
self._name = name or "qemu-%d" % os.getpid() self._name = name or "qemu-%d" % os.getpid()
@ -136,7 +136,7 @@ def __init__(self, binary, args=None, wrapper=None, name=None,
self._iolog = None self._iolog = None
self._qmp_set = True # Enable QMP monitor by default. self._qmp_set = True # Enable QMP monitor by default.
self._qmp_connection: Optional[qmp.QEMUMonitorProtocol] = None self._qmp_connection: Optional[qmp.QEMUMonitorProtocol] = None
self._qemu_full_args = None self._qemu_full_args: Tuple[str, ...] = ()
self._temp_dir = None self._temp_dir = None
self._launched = False self._launched = False
self._machine = None self._machine = None
@ -368,7 +368,7 @@ def launch(self):
raise QEMUMachineError('VM already launched') raise QEMUMachineError('VM already launched')
self._iolog = None self._iolog = None
self._qemu_full_args = None self._qemu_full_args = ()
try: try:
self._launch() self._launch()
self._launched = True self._launched = True
@ -388,8 +388,12 @@ def _launch(self):
""" """
devnull = open(os.path.devnull, 'rb') devnull = open(os.path.devnull, 'rb')
self._pre_launch() self._pre_launch()
self._qemu_full_args = (self._wrapper + [self._binary] + self._qemu_full_args = tuple(
self._base_args + self._args) chain(self._wrapper,
[self._binary],
self._base_args,
self._args)
)
LOG.debug('VM launch command: %r', ' '.join(self._qemu_full_args)) LOG.debug('VM launch command: %r', ' '.join(self._qemu_full_args))
self._popen = subprocess.Popen(self._qemu_full_args, self._popen = subprocess.Popen(self._qemu_full_args,
stdin=devnull, stdin=devnull,

View File

@ -19,7 +19,12 @@
import os import os
import socket import socket
from typing import List, Optional, TextIO from typing import (
List,
Optional,
Sequence,
TextIO,
)
from .machine import QEMUMachine from .machine import QEMUMachine
@ -99,8 +104,13 @@ class QEMUQtestMachine(QEMUMachine):
A QEMU VM, with a qtest socket available. A QEMU VM, with a qtest socket available.
""" """
def __init__(self, binary, args=None, name=None, test_dir="/var/tmp", def __init__(self,
socket_scm_helper=None, sock_dir=None): binary: str,
args: Sequence[str] = (),
name: Optional[str] = None,
test_dir: str = "/var/tmp",
socket_scm_helper: Optional[str] = None,
sock_dir: Optional[str] = None):
if name is None: if name is None:
name = "qemu-%d" % os.getpid() name = "qemu-%d" % os.getpid()
if sock_dir is None: if sock_dir is None:
@ -114,8 +124,10 @@ def __init__(self, binary, args=None, name=None, test_dir="/var/tmp",
@property @property
def _base_args(self) -> List[str]: def _base_args(self) -> List[str]:
args = super()._base_args args = super()._base_args
args.extend(['-qtest', 'unix:path=' + self._qtest_path, args.extend([
'-accel', 'qtest']) '-qtest', f"unix:path={self._qtest_path}",
'-accel', 'qtest'
])
return args return args
def _pre_launch(self): def _pre_launch(self):