Merge remote-tracking branch 'remotes/cleber/tags/python-next-pull-request' into staging

Python queue, 2019-02-22

Python:
* introduce "python" directory with module namespace
* log QEMU launch command line on qemu.QEMUMachine

Acceptance Tests:
* initrd 4GiB+ test
* migration test
* multi vm support in test class
* bump Avocado version and drop ":avocado: enable"

# gpg: Signature made Fri 22 Feb 2019 19:37:07 GMT
# gpg:                using RSA key 657E8D33A5F209F3
# gpg: Good signature from "Cleber Rosa <crosa@redhat.com>" [marginal]
# gpg: WARNING: This key is not certified with sufficiently trusted signatures!
# gpg:          It is not certain that the signature belongs to the owner.
# Primary key fingerprint: 7ABB 96EB 8B46 B94D 5E0F  E9BB 657E 8D33 A5F2 09F3

* remotes/cleber/tags/python-next-pull-request:
  Acceptance tests: expect boot to extract 2GiB+ initrd with linux-v4.16
  Acceptance tests: use linux-3.6 and set vm memory to 4GiB
  tests.acceptance: adds simple migration test
  tests.acceptance: adds multi vm capability for acceptance tests
  scripts/qemu.py: log QEMU launch command line
  Introduce a Python module structure
  Acceptance tests: drop usage of ":avocado: enable"

Signed-off-by: Peter Maydell <peter.maydell@linaro.org>

3 files changed, 890 insertions, 0 deletions

diff --git a/python/qemu/__init__.py b/python/qemu/__init__.py
new file mode 100644
index 0000000..fd144c0
--- /dev/null
+++ b/python/qemu/__init__.py
@@ -0,0 +1,518 @@
+# QEMU library
+#
+# Copyright (C) 2015-2016 Red Hat Inc.
+# Copyright (C) 2012 IBM Corp.
+#
+# Authors:
+#  Fam Zheng <famz@redhat.com>
+#
+# This work is licensed under the terms of the GNU GPL, version 2.  See
+# the COPYING file in the top-level directory.
+#
+# Based on qmp.py.
+#
+
+import errno
+import logging
+import os
+import subprocess
+import re
+import shutil
+import socket
+import tempfile
+
+from . import qmp
+
+
+LOG = logging.getLogger(__name__)
+
+# Mapping host architecture to any additional architectures it can
+# support which often includes its 32 bit cousin.
+ADDITIONAL_ARCHES = {
+    "x86_64" : "i386",
+    "aarch64" : "armhf"
+}
+
+def kvm_available(target_arch=None):
+    host_arch = os.uname()[4]
+    if target_arch and target_arch != host_arch:
+        if target_arch != ADDITIONAL_ARCHES.get(host_arch):
+            return False
+    return os.access("/dev/kvm", os.R_OK | os.W_OK)
+
+
+#: Maps machine types to the preferred console device types
+CONSOLE_DEV_TYPES = {
+    r'^clipper$': 'isa-serial',
+    r'^malta': 'isa-serial',
+    r'^(pc.*|q35.*|isapc)$': 'isa-serial',
+    r'^(40p|powernv|prep)$': 'isa-serial',
+    r'^pseries.*': 'spapr-vty',
+    r'^s390-ccw-virtio.*': 'sclpconsole',
+    }
+
+
+class QEMUMachineError(Exception):
+    """
+    Exception called when an error in QEMUMachine happens.
+    """
+
+
+class QEMUMachineAddDeviceError(QEMUMachineError):
+    """
+    Exception raised when a request to add a device can not be fulfilled
+
+    The failures are caused by limitations, lack of information or conflicting
+    requests on the QEMUMachine methods.  This exception does not represent
+    failures reported by the QEMU binary itself.
+    """
+
+class MonitorResponseError(qmp.QMPError):
+    """
+    Represents erroneous QMP monitor reply
+    """
+    def __init__(self, reply):
+        try:
+            desc = reply["error"]["desc"]
+        except KeyError:
+            desc = reply
+        super(MonitorResponseError, self).__init__(desc)
+        self.reply = reply
+
+
+class QEMUMachine(object):
+    """
+    A QEMU VM
+
+    Use this object as a context manager to ensure the QEMU process terminates::
+
+        with VM(binary) as vm:
+            ...
+        # vm is guaranteed to be shut down here
+    """
+
+    def __init__(self, binary, args=None, wrapper=None, name=None,
+                 test_dir="/var/tmp", monitor_address=None,
+                 socket_scm_helper=None):
+        '''
+        Initialize a QEMUMachine
+
+        @param binary: path to the qemu binary
+        @param args: list of extra arguments
+        @param wrapper: list of arguments used as prefix to qemu binary
+        @param name: prefix for socket and log file names (default: qemu-PID)
+        @param test_dir: where to create socket and log file
+        @param monitor_address: address for QMP monitor
+        @param socket_scm_helper: helper program, required for send_fd_scm()
+        @note: Qemu process is not started until launch() is used.
+        '''
+        if args is None:
+            args = []
+        if wrapper is None:
+            wrapper = []
+        if name is None:
+            name = "qemu-%d" % os.getpid()
+        self._name = name
+        self._monitor_address = monitor_address
+        self._vm_monitor = None
+        self._qemu_log_path = None
+        self._qemu_log_file = None
+        self._popen = None
+        self._binary = binary
+        self._args = list(args)     # Force copy args in case we modify them
+        self._wrapper = wrapper
+        self._events = []
+        self._iolog = None
+        self._socket_scm_helper = socket_scm_helper
+        self._qmp = None
+        self._qemu_full_args = None
+        self._test_dir = test_dir
+        self._temp_dir = None
+        self._launched = False
+        self._machine = None
+        self._console_device_type = None
+        self._console_address = None
+        self._console_socket = None
+
+        # just in case logging wasn't configured by the main script:
+        logging.basicConfig()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.shutdown()
+        return False
+
+    # This can be used to add an unused monitor instance.
+    def add_monitor_null(self):
+        self._args.append('-monitor')
+        self._args.append('null')
+
+    def add_fd(self, fd, fdset, opaque, opts=''):
+        """
+        Pass a file descriptor to the VM
+        """
+        options = ['fd=%d' % fd,
+                   'set=%d' % fdset,
+                   'opaque=%s' % opaque]
+        if opts:
+            options.append(opts)
+
+        # This did not exist before 3.4, but since then it is
+        # mandatory for our purpose
+        if hasattr(os, 'set_inheritable'):
+            os.set_inheritable(fd, True)
+
+        self._args.append('-add-fd')
+        self._args.append(','.join(options))
+        return self
+
+    # Exactly one of fd and file_path must be given.
+    # (If it is file_path, the helper will open that file and pass its
+    # own fd)
+    def send_fd_scm(self, fd=None, file_path=None):
+        # In iotest.py, the qmp should always use unix socket.
+        assert self._qmp.is_scm_available()
+        if self._socket_scm_helper is None:
+            raise QEMUMachineError("No path to socket_scm_helper set")
+        if not os.path.exists(self._socket_scm_helper):
+            raise QEMUMachineError("%s does not exist" %
+                                   self._socket_scm_helper)
+
+        # This did not exist before 3.4, but since then it is
+        # mandatory for our purpose
+        if hasattr(os, 'set_inheritable'):
+            os.set_inheritable(self._qmp.get_sock_fd(), True)
+            if fd is not None:
+                os.set_inheritable(fd, True)
+
+        fd_param = ["%s" % self._socket_scm_helper,
+                    "%d" % self._qmp.get_sock_fd()]
+
+        if file_path is not None:
+            assert fd is None
+            fd_param.append(file_path)
+        else:
+            assert fd is not None
+            fd_param.append(str(fd))
+
+        devnull = open(os.path.devnull, 'rb')
+        proc = subprocess.Popen(fd_param, stdin=devnull, stdout=subprocess.PIPE,
+                                stderr=subprocess.STDOUT, close_fds=False)
+        output = proc.communicate()[0]
+        if output:
+            LOG.debug(output)
+
+        return proc.returncode
+
+    @staticmethod
+    def _remove_if_exists(path):
+        """
+        Remove file object at path if it exists
+        """
+        try:
+            os.remove(path)
+        except OSError as exception:
+            if exception.errno == errno.ENOENT:
+                return
+            raise
+
+    def is_running(self):
+        return self._popen is not None and self._popen.poll() is None
+
+    def exitcode(self):
+        if self._popen is None:
+            return None
+        return self._popen.poll()
+
+    def get_pid(self):
+        if not self.is_running():
+            return None
+        return self._popen.pid
+
+    def _load_io_log(self):
+        if self._qemu_log_path is not None:
+            with open(self._qemu_log_path, "r") as iolog:
+                self._iolog = iolog.read()
+
+    def _base_args(self):
+        if isinstance(self._monitor_address, tuple):
+            moncdev = "socket,id=mon,host=%s,port=%s" % (
+                self._monitor_address[0],
+                self._monitor_address[1])
+        else:
+            moncdev = 'socket,id=mon,path=%s' % self._vm_monitor
+        args = ['-chardev', moncdev,
+                '-mon', 'chardev=mon,mode=control',
+                '-display', 'none', '-vga', 'none']
+        if self._machine is not None:
+            args.extend(['-machine', self._machine])
+        if self._console_device_type is not None:
+            self._console_address = os.path.join(self._temp_dir,
+                                                 self._name + "-console.sock")
+            chardev = ('socket,id=console,path=%s,server,nowait' %
+                       self._console_address)
+            device = '%s,chardev=console' % self._console_device_type
+            args.extend(['-chardev', chardev, '-device', device])
+        return args
+
+    def _pre_launch(self):
+        self._temp_dir = tempfile.mkdtemp(dir=self._test_dir)
+        if self._monitor_address is not None:
+            self._vm_monitor = self._monitor_address
+        else:
+            self._vm_monitor = os.path.join(self._temp_dir,
+                                            self._name + "-monitor.sock")
+        self._qemu_log_path = os.path.join(self._temp_dir, self._name + ".log")
+        self._qemu_log_file = open(self._qemu_log_path, 'wb')
+
+        self._qmp = qmp.QEMUMonitorProtocol(self._vm_monitor,
+                                            server=True)
+
+    def _post_launch(self):
+        self._qmp.accept()
+
+    def _post_shutdown(self):
+        if self._qemu_log_file is not None:
+            self._qemu_log_file.close()
+            self._qemu_log_file = None
+
+        self._qemu_log_path = None
+
+        if self._console_socket is not None:
+            self._console_socket.close()
+            self._console_socket = None
+
+        if self._temp_dir is not None:
+            shutil.rmtree(self._temp_dir)
+            self._temp_dir = None
+
+    def launch(self):
+        """
+        Launch the VM and make sure we cleanup and expose the
+        command line/output in case of exception
+        """
+
+        if self._launched:
+            raise QEMUMachineError('VM already launched')
+
+        self._iolog = None
+        self._qemu_full_args = None
+        try:
+            self._launch()
+            self._launched = True
+        except:
+            self.shutdown()
+
+            LOG.debug('Error launching VM')
+            if self._qemu_full_args:
+                LOG.debug('Command: %r', ' '.join(self._qemu_full_args))
+            if self._iolog:
+                LOG.debug('Output: %r', self._iolog)
+            raise
+
+    def _launch(self):
+        """
+        Launch the VM and establish a QMP connection
+        """
+        devnull = open(os.path.devnull, 'rb')
+        self._pre_launch()
+        self._qemu_full_args = (self._wrapper + [self._binary] +
+                                self._base_args() + self._args)
+        LOG.debug('VM launch command: %r', ' '.join(self._qemu_full_args))
+        self._popen = subprocess.Popen(self._qemu_full_args,
+                                       stdin=devnull,
+                                       stdout=self._qemu_log_file,
+                                       stderr=subprocess.STDOUT,
+                                       shell=False,
+                                       close_fds=False)
+        self._post_launch()
+
+    def wait(self):
+        """
+        Wait for the VM to power off
+        """
+        self._popen.wait()
+        self._qmp.close()
+        self._load_io_log()
+        self._post_shutdown()
+
+    def shutdown(self):
+        """
+        Terminate the VM and clean up
+        """
+        if self.is_running():
+            try:
+                self._qmp.cmd('quit')
+                self._qmp.close()
+            except:
+                self._popen.kill()
+            self._popen.wait()
+
+        self._load_io_log()
+        self._post_shutdown()
+
+        exitcode = self.exitcode()
+        if exitcode is not None and exitcode < 0:
+            msg = 'qemu received signal %i: %s'
+            if self._qemu_full_args:
+                command = ' '.join(self._qemu_full_args)
+            else:
+                command = ''
+            LOG.warn(msg, -exitcode, command)
+
+        self._launched = False
+
+    def qmp(self, cmd, conv_keys=True, **args):
+        """
+        Invoke a QMP command and return the response dict
+        """
+        qmp_args = dict()
+        for key, value in args.items():
+            if conv_keys:
+                qmp_args[key.replace('_', '-')] = value
+            else:
+                qmp_args[key] = value
+
+        return self._qmp.cmd(cmd, args=qmp_args)
+
+    def command(self, cmd, conv_keys=True, **args):
+        """
+        Invoke a QMP command.
+        On success return the response dict.
+        On failure raise an exception.
+        """
+        reply = self.qmp(cmd, conv_keys, **args)
+        if reply is None:
+            raise qmp.QMPError("Monitor is closed")
+        if "error" in reply:
+            raise MonitorResponseError(reply)
+        return reply["return"]
+
+    def get_qmp_event(self, wait=False):
+        """
+        Poll for one queued QMP events and return it
+        """
+        if len(self._events) > 0:
+            return self._events.pop(0)
+        return self._qmp.pull_event(wait=wait)
+
+    def get_qmp_events(self, wait=False):
+        """
+        Poll for queued QMP events and return a list of dicts
+        """
+        events = self._qmp.get_events(wait=wait)
+        events.extend(self._events)
+        del self._events[:]
+        self._qmp.clear_events()
+        return events
+
+    def event_wait(self, name, timeout=60.0, match=None):
+        """
+        Wait for specified timeout on named event in QMP; optionally filter
+        results by match.
+
+        The 'match' is checked to be a recursive subset of the 'event'; skips
+        branch processing on match's value None
+           {"foo": {"bar": 1}} matches {"foo": None}
+           {"foo": {"bar": 1}} does not matches {"foo": {"baz": None}}
+        """
+        def event_match(event, match=None):
+            if match is None:
+                return True
+
+            for key in match:
+                if key in event:
+                    if isinstance(event[key], dict):
+                        if not event_match(event[key], match[key]):
+                            return False
+                    elif event[key] != match[key]:
+                        return False
+                else:
+                    return False
+
+            return True
+
+        # Search cached events
+        for event in self._events:
+            if (event['event'] == name) and event_match(event, match):
+                self._events.remove(event)
+                return event
+
+        # Poll for new events
+        while True:
+            event = self._qmp.pull_event(wait=timeout)
+            if (event['event'] == name) and event_match(event, match):
+                return event
+            self._events.append(event)
+
+        return None
+
+    def get_log(self):
+        """
+        After self.shutdown or failed qemu execution, this returns the output
+        of the qemu process.
+        """
+        return self._iolog
+
+    def add_args(self, *args):
+        """
+        Adds to the list of extra arguments to be given to the QEMU binary
+        """
+        self._args.extend(args)
+
+    def set_machine(self, machine_type):
+        """
+        Sets the machine type
+
+        If set, the machine type will be added to the base arguments
+        of the resulting QEMU command line.
+        """
+        self._machine = machine_type
+
+    def set_console(self, device_type=None):
+        """
+        Sets the device type for a console device
+
+        If set, the console device and a backing character device will
+        be added to the base arguments of the resulting QEMU command
+        line.
+
+        This is a convenience method that will either use the provided
+        device type, of if not given, it will used the device type set
+        on CONSOLE_DEV_TYPES.
+
+        The actual setting of command line arguments will be be done at
+        machine launch time, as it depends on the temporary directory
+        to be created.
+
+        @param device_type: the device type, such as "isa-serial"
+        @raises: QEMUMachineAddDeviceError if the device type is not given
+                 and can not be determined.
+        """
+        if device_type is None:
+            if self._machine is None:
+                raise QEMUMachineAddDeviceError("Can not add a console device:"
+                                                " QEMU instance without a "
+                                                "defined machine type")
+            for regex, device in CONSOLE_DEV_TYPES.items():
+                if re.match(regex, self._machine):
+                    device_type = device
+                    break
+            if device_type is None:
+                raise QEMUMachineAddDeviceError("Can not add a console device:"
+                                                " no matching console device "
+                                                "type definition")
+        self._console_device_type = device_type
+
+    @property
+    def console_socket(self):
+        """
+        Returns a socket connected to the console
+        """
+        if self._console_socket is None:
+            self._console_socket = socket.socket(socket.AF_UNIX,
+                                                 socket.SOCK_STREAM)
+            self._console_socket.connect(self._console_address)
+        return self._console_socket
diff --git a/python/qemu/qmp.py b/python/qemu/qmp.py
new file mode 100644
index 0000000..5c8cf6a
--- /dev/null
+++ b/python/qemu/qmp.py
@@ -0,0 +1,256 @@
+# QEMU Monitor Protocol Python class
+#
+# Copyright (C) 2009, 2010 Red Hat Inc.
+#
+# Authors:
+#  Luiz Capitulino <lcapitulino@redhat.com>
+#
+# This work is licensed under the terms of the GNU GPL, version 2.  See
+# the COPYING file in the top-level directory.
+
+import json
+import errno
+import socket
+import logging
+
+
+class QMPError(Exception):
+    pass
+
+
+class QMPConnectError(QMPError):
+    pass
+
+
+class QMPCapabilitiesError(QMPError):
+    pass
+
+
+class QMPTimeoutError(QMPError):
+    pass
+
+
+class QEMUMonitorProtocol(object):
+
+    #: Logger object for debugging messages
+    logger = logging.getLogger('QMP')
+    #: Socket's error class
+    error = socket.error
+    #: Socket's timeout
+    timeout = socket.timeout
+
+    def __init__(self, address, server=False):
+        """
+        Create a QEMUMonitorProtocol class.
+
+        @param address: QEMU address, can be either a unix socket path (string)
+                        or a tuple in the form ( address, port ) for a TCP
+                        connection
+        @param server: server mode listens on the socket (bool)
+        @raise socket.error on socket connection errors
+        @note No connection is established, this is done by the connect() or
+              accept() methods
+        """
+        self.__events = []
+        self.__address = address
+        self.__sock = self.__get_sock()
+        self.__sockfile = None
+        if server:
+            self.__sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            self.__sock.bind(self.__address)
+            self.__sock.listen(1)
+
+    def __get_sock(self):
+        if isinstance(self.__address, tuple):
+            family = socket.AF_INET
+        else:
+            family = socket.AF_UNIX
+        return socket.socket(family, socket.SOCK_STREAM)
+
+    def __negotiate_capabilities(self):
+        greeting = self.__json_read()
+        if greeting is None or "QMP" not in greeting:
+            raise QMPConnectError
+        # Greeting seems ok, negotiate capabilities
+        resp = self.cmd('qmp_capabilities')
+        if "return" in resp:
+            return greeting
+        raise QMPCapabilitiesError
+
+    def __json_read(self, only_event=False):
+        while True:
+            data = self.__sockfile.readline()
+            if not data:
+                return
+            resp = json.loads(data)
+            if 'event' in resp:
+                self.logger.debug("<<< %s", resp)
+                self.__events.append(resp)
+                if not only_event:
+                    continue
+            return resp
+
+    def __get_events(self, wait=False):
+        """
+        Check for new events in the stream and cache them in __events.
+
+        @param wait (bool): block until an event is available.
+        @param wait (float): If wait is a float, treat it as a timeout value.
+
+        @raise QMPTimeoutError: If a timeout float is provided and the timeout
+                                period elapses.
+        @raise QMPConnectError: If wait is True but no events could be
+                                retrieved or if some other error occurred.
+        """
+
+        # Check for new events regardless and pull them into the cache:
+        self.__sock.setblocking(0)
+        try:
+            self.__json_read()
+        except socket.error as err:
+            if err[0] == errno.EAGAIN:
+                # No data available
+                pass
+        self.__sock.setblocking(1)
+
+        # Wait for new events, if needed.
+        # if wait is 0.0, this means "no wait" and is also implicitly false.
+        if not self.__events and wait:
+            if isinstance(wait, float):
+                self.__sock.settimeout(wait)
+            try:
+                ret = self.__json_read(only_event=True)
+            except socket.timeout:
+                raise QMPTimeoutError("Timeout waiting for event")
+            except:
+                raise QMPConnectError("Error while reading from socket")
+            if ret is None:
+                raise QMPConnectError("Error while reading from socket")
+            self.__sock.settimeout(None)
+
+    def connect(self, negotiate=True):
+        """
+        Connect to the QMP Monitor and perform capabilities negotiation.
+
+        @return QMP greeting dict
+        @raise socket.error on socket connection errors
+        @raise QMPConnectError if the greeting is not received
+        @raise QMPCapabilitiesError if fails to negotiate capabilities
+        """
+        self.__sock.connect(self.__address)
+        self.__sockfile = self.__sock.makefile()
+        if negotiate:
+            return self.__negotiate_capabilities()
+
+    def accept(self):
+        """
+        Await connection from QMP Monitor and perform capabilities negotiation.
+
+        @return QMP greeting dict
+        @raise socket.error on socket connection errors
+        @raise QMPConnectError if the greeting is not received
+        @raise QMPCapabilitiesError if fails to negotiate capabilities
+        """
+        self.__sock.settimeout(15)
+        self.__sock, _ = self.__sock.accept()
+        self.__sockfile = self.__sock.makefile()
+        return self.__negotiate_capabilities()
+
+    def cmd_obj(self, qmp_cmd):
+        """
+        Send a QMP command to the QMP Monitor.
+
+        @param qmp_cmd: QMP command to be sent as a Python dict
+        @return QMP response as a Python dict or None if the connection has
+                been closed
+        """
+        self.logger.debug(">>> %s", qmp_cmd)
+        try:
+            self.__sock.sendall(json.dumps(qmp_cmd).encode('utf-8'))
+        except socket.error as err:
+            if err[0] == errno.EPIPE:
+                return
+            raise socket.error(err)
+        resp = self.__json_read()
+        self.logger.debug("<<< %s", resp)
+        return resp
+
+    def cmd(self, name, args=None, cmd_id=None):
+        """
+        Build a QMP command and send it to the QMP Monitor.
+
+        @param name: command name (string)
+        @param args: command arguments (dict)
+        @param cmd_id: command id (dict, list, string or int)
+        """
+        qmp_cmd = {'execute': name}
+        if args:
+            qmp_cmd['arguments'] = args
+        if cmd_id:
+            qmp_cmd['id'] = cmd_id
+        return self.cmd_obj(qmp_cmd)
+
+    def command(self, cmd, **kwds):
+        """
+        Build and send a QMP command to the monitor, report errors if any
+        """
+        ret = self.cmd(cmd, kwds)
+        if "error" in ret:
+            raise Exception(ret['error']['desc'])
+        return ret['return']
+
+    def pull_event(self, wait=False):
+        """
+        Pulls a single event.
+
+        @param wait (bool): block until an event is available.
+        @param wait (float): If wait is a float, treat it as a timeout value.
+
+        @raise QMPTimeoutError: If a timeout float is provided and the timeout
+                                period elapses.
+        @raise QMPConnectError: If wait is True but no events could be
+                                retrieved or if some other error occurred.
+
+        @return The first available QMP event, or None.
+        """
+        self.__get_events(wait)
+
+        if self.__events:
+            return self.__events.pop(0)
+        return None
+
+    def get_events(self, wait=False):
+        """
+        Get a list of available QMP events.
+
+        @param wait (bool): block until an event is available.
+        @param wait (float): If wait is a float, treat it as a timeout value.
+
+        @raise QMPTimeoutError: If a timeout float is provided and the timeout
+                                period elapses.
+        @raise QMPConnectError: If wait is True but no events could be
+                                retrieved or if some other error occurred.
+
+        @return The list of available QMP events.
+        """
+        self.__get_events(wait)
+        return self.__events
+
+    def clear_events(self):
+        """
+        Clear current list of pending events.
+        """
+        self.__events = []
+
+    def close(self):
+        self.__sock.close()
+        self.__sockfile.close()
+
+    def settimeout(self, timeout):
+        self.__sock.settimeout(timeout)
+
+    def get_sock_fd(self):
+        return self.__sock.fileno()
+
+    def is_scm_available(self):
+        return self.__sock.family == socket.AF_UNIX
diff --git a/python/qemu/qtest.py b/python/qemu/qtest.py
new file mode 100644
index 0000000..eb45824
--- /dev/null
+++ b/python/qemu/qtest.py
@@ -0,0 +1,116 @@
+# QEMU qtest library
+#
+# Copyright (C) 2015 Red Hat Inc.
+#
+# Authors:
+#  Fam Zheng <famz@redhat.com>
+#
+# This work is licensed under the terms of the GNU GPL, version 2.  See
+# the COPYING file in the top-level directory.
+#
+# Based on qmp.py.
+#
+
+import socket
+import os
+
+from . import QEMUMachine
+
+
+class QEMUQtestProtocol(object):
+    def __init__(self, address, server=False):
+        """
+        Create a QEMUQtestProtocol object.
+
+        @param address: QEMU address, can be either a unix socket path (string)
+                        or a tuple in the form ( address, port ) for a TCP
+                        connection
+        @param server: server mode, listens on the socket (bool)
+        @raise socket.error on socket connection errors
+        @note No connection is established, this is done by the connect() or
+              accept() methods
+        """
+        self._address = address
+        self._sock = self._get_sock()
+        self._sockfile = None
+        if server:
+            self._sock.bind(self._address)
+            self._sock.listen(1)
+
+    def _get_sock(self):
+        if isinstance(self._address, tuple):
+            family = socket.AF_INET
+        else:
+            family = socket.AF_UNIX
+        return socket.socket(family, socket.SOCK_STREAM)
+
+    def connect(self):
+        """
+        Connect to the qtest socket.
+
+        @raise socket.error on socket connection errors
+        """
+        self._sock.connect(self._address)
+        self._sockfile = self._sock.makefile()
+
+    def accept(self):
+        """
+        Await connection from QEMU.
+
+        @raise socket.error on socket connection errors
+        """
+        self._sock, _ = self._sock.accept()
+        self._sockfile = self._sock.makefile()
+
+    def cmd(self, qtest_cmd):
+        """
+        Send a qtest command on the wire.
+
+        @param qtest_cmd: qtest command text to be sent
+        """
+        self._sock.sendall((qtest_cmd + "\n").encode('utf-8'))
+        resp = self._sockfile.readline()
+        return resp
+
+    def close(self):
+        self._sock.close()
+        self._sockfile.close()
+
+    def settimeout(self, timeout):
+        self._sock.settimeout(timeout)
+
+
+class QEMUQtestMachine(QEMUMachine):
+    '''A QEMU VM'''
+
+    def __init__(self, binary, args=None, name=None, test_dir="/var/tmp",
+                 socket_scm_helper=None):
+        if name is None:
+            name = "qemu-%d" % os.getpid()
+        super(QEMUQtestMachine,
+              self).__init__(binary, args, name=name, test_dir=test_dir,
+                             socket_scm_helper=socket_scm_helper)
+        self._qtest = None
+        self._qtest_path = os.path.join(test_dir, name + "-qtest.sock")
+
+    def _base_args(self):
+        args = super(QEMUQtestMachine, self)._base_args()
+        args.extend(['-qtest', 'unix:path=' + self._qtest_path,
+                     '-machine', 'accel=qtest'])
+        return args
+
+    def _pre_launch(self):
+        super(QEMUQtestMachine, self)._pre_launch()
+        self._qtest = QEMUQtestProtocol(self._qtest_path, server=True)
+
+    def _post_launch(self):
+        super(QEMUQtestMachine, self)._post_launch()
+        self._qtest.accept()
+
+    def _post_shutdown(self):
+        super(QEMUQtestMachine, self)._post_shutdown()
+        self._remove_if_exists(self._qtest_path)
+
+    def qtest(self, cmd):
+        '''Send a qtest command to guest'''
+        return self._qtest.cmd(cmd)