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 errno
  15import socket
  16
  17class QEMUQtestProtocol(object):
  18    def __init__(self, address, server=False):
  19        """
  20        Create a QEMUQtestProtocol object.
  21
  22        @param address: QEMU address, can be either a unix socket path (string)
  23                        or a tuple in the form ( address, port ) for a TCP
  24                        connection
  25        @param server: server mode, listens on the socket (bool)
  26        @raise socket.error on socket connection errors
  27        @note No connection is established, this is done by the connect() or
  28              accept() methods
  29        """
  30        self._address = address
  31        self._sock = self._get_sock()
  32        if server:
  33            self._sock.bind(self._address)
  34            self._sock.listen(1)
  35
  36    def _get_sock(self):
  37        if isinstance(self._address, tuple):
  38            family = socket.AF_INET
  39        else:
  40            family = socket.AF_UNIX
  41        return socket.socket(family, socket.SOCK_STREAM)
  42
  43    def connect(self):
  44        """
  45        Connect to the qtest socket.
  46
  47        @raise socket.error on socket connection errors
  48        """
  49        self._sock.connect(self._address)
  50
  51    def accept(self):
  52        """
  53        Await connection from QEMU.
  54
  55        @raise socket.error on socket connection errors
  56        """
  57        self._sock, _ = self._sock.accept()
  58
  59    def cmd(self, qtest_cmd):
  60        """
  61        Send a qtest command on the wire.
  62
  63        @param qtest_cmd: qtest command text to be sent
  64        """
  65        self._sock.sendall(qtest_cmd + "\n")
  66
  67    def close(self):
  68        self._sock.close()
  69
  70    def settimeout(self, timeout):
  71        self._sock.settimeout(timeout)
  72