LXR qemu/scripts/qtest.py

   1# QEMU qtest library
   2#
   3# Copyright (C) 2015 Red Hat Inc.
   4#
   5# Authors:
   6#  Fam Zheng <famz@redhat.com>
   7#
   8# This work is licensed under the terms of the GNU GPL, version 2.  See
   9# the COPYING file in the top-level directory.
  10#
  11# Based on qmp.py.
  12#
  13
  14import socket
  15import os
  16import qemu
  17
  18
  19class QEMUQtestProtocol(object):
  20    def __init__(self, address, server=False):
  21        """
  22        Create a QEMUQtestProtocol object.
  23
  24        @param address: QEMU address, can be either a unix socket path (string)
  25                        or a tuple in the form ( address, port ) for a TCP
  26                        connection
  27        @param server: server mode, listens on the socket (bool)
  28        @raise socket.error on socket connection errors
  29        @note No connection is established, this is done by the connect() or
  30              accept() methods
  31        """
  32        self._address = address
  33        self._sock = self._get_sock()
  34        if server:
  35            self._sock.bind(self._address)
  36            self._sock.listen(1)
  37
  38    def _get_sock(self):
  39        if isinstance(self._address, tuple):
  40            family = socket.AF_INET
  41        else:
  42            family = socket.AF_UNIX
  43        return socket.socket(family, socket.SOCK_STREAM)
  44
  45    def connect(self):
  46        """
  47        Connect to the qtest socket.
  48
  49        @raise socket.error on socket connection errors
  50        """
  51        self._sock.connect(self._address)
  52
  53    def accept(self):
  54        """
  55        Await connection from QEMU.
  56
  57        @raise socket.error on socket connection errors
  58        """
  59        self._sock, _ = self._sock.accept()
  60
  61    def cmd(self, qtest_cmd):
  62        """
  63        Send a qtest command on the wire.
  64
  65        @param qtest_cmd: qtest command text to be sent
  66        """
  67        self._sock.sendall(qtest_cmd + "\n")
  68
  69    def close(self):
  70        self._sock.close()
  71
  72    def settimeout(self, timeout):
  73        self._sock.settimeout(timeout)
  74
  75
  76class QEMUQtestMachine(qemu.QEMUMachine):
  77    '''A QEMU VM'''
  78
  79    def __init__(self, binary, args=None, name=None, test_dir="/var/tmp",
  80                 socket_scm_helper=None):
  81        if name is None:
  82            name = "qemu-%d" % os.getpid()
  83        super(QEMUQtestMachine,
  84              self).__init__(binary, args, name=name, test_dir=test_dir,
  85                             socket_scm_helper=socket_scm_helper)
  86        self._qtest = None
  87        self._qtest_path = os.path.join(test_dir, name + "-qtest.sock")
  88
  89    def _base_args(self):
  90        args = super(QEMUQtestMachine, self)._base_args()
  91        args.extend(['-qtest', 'unix:path=' + self._qtest_path,
  92                     '-machine', 'accel=qtest'])
  93        return args
  94
  95    def _pre_launch(self):
  96        super(QEMUQtestMachine, self)._pre_launch()
  97        self._qtest = QEMUQtestProtocol(self._qtest_path, server=True)
  98
  99    def _post_launch(self):
 100        super(QEMUQtestMachine, self)._post_launch()
 101        self._qtest.accept()
 102
 103    def _post_shutdown(self):
 104        super(QEMUQtestMachine, self)._post_shutdown()
 105        self._remove_if_exists(self._qtest_path)
 106
 107    def qtest(self, cmd):
 108        '''Send a qtest command to guest'''
 109        return self._qtest.cmd(cmd)
 110