From 8f8fd9edba4bd6768da2c8e2bea49ad5c16ced1a Mon Sep 17 00:00:00 2001 From: Cleber Rosa Date: Wed, 6 Feb 2019 11:29:01 -0500 Subject: Introduce a Python module structure This is a simple move of Python code that wraps common QEMU functionality, and are used by a number of different tests and scripts. By treating that code as a real Python module, we can more easily: * reuse code * have a proper place for the module's own unittests * apply a more consistent style * generate documentation Signed-off-by: Cleber Rosa Reviewed-by: Caio Carrara Reviewed-by: Stefan Hajnoczi Message-Id: <20190206162901.19082-2-crosa@redhat.com> Signed-off-by: Cleber Rosa --- python/qemu/qtest.py | 116 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 116 insertions(+) create mode 100644 python/qemu/qtest.py (limited to 'python/qemu/qtest.py') 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 +# +# 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) -- cgit v1.1