Re: [Qemu-devel] [PATCH v3 2/5] qtest: Add scripts/qtest/qtest.py

["Benoît Canet" <benoit.canet@irqsave.net>]

The Tuesday 11 Feb 2014 à 14:40:59 (+0800), Fam Zheng wrote :
> This removes the dummy scripts/qtest and adds scripts/qtest/qtest.py as
> a python library for qtest protocol.
> 
> This is a skeleton with a basic "cmd" method to execute a command,
> reading and parsing of qtest output will be added later on demand.
> 
> Signed-off-by: Fam Zheng <famz@redhat.com>
> ---
>  scripts/qtest          |  5 ----
>  scripts/qtest/qtest.py | 74 ++++++++++++++++++++++++++++++++++++++++++++++++++
>  2 files changed, 74 insertions(+), 5 deletions(-)
>  delete mode 100755 scripts/qtest
>  create mode 100644 scripts/qtest/qtest.py
> 
> diff --git a/scripts/qtest b/scripts/qtest
> deleted file mode 100755
> index 4ef6c1c..0000000
> --- a/scripts/qtest
> +++ /dev/null
> @@ -1,5 +0,0 @@
> -#!/bin/sh
> -
> -export QTEST_QEMU_BINARY=$1
> -shift
> -"$@"
> diff --git a/scripts/qtest/qtest.py b/scripts/qtest/qtest.py
> new file mode 100644
> index 0000000..16c6713
> --- /dev/null
> +++ b/scripts/qtest/qtest.py
> @@ -0,0 +1,74 @@
> +# QEMU qtest library
> +#
> +# Copyright (C) 2014 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 errno
> +import socket
> +
> +class QEMUQtestProtocol:
> +    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()
> +        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")
> +
> +    def close(self):
> +        self.__sockfile.close()
> +        self.__sock.close()
> +
> +    def settimeout(self, timeout):
> +        self.__sock.settimeout(timeout)
> -- 
> 1.8.5.4
> 
Reviewed-by: Benoit Canet <benoit@irqsave.net>