xref: /openbmc/qemu/python/qemu/machine/qtest.py (revision 40f23e4e) 
1"""
2QEMU qtest library
3
4qtest offers the QEMUQtestProtocol and QEMUQTestMachine classes, which
5offer a connection to QEMU's qtest protocol socket, and a qtest-enabled
6subclass of QEMUMachine, respectively.
7"""
8
9# Copyright (C) 2015 Red Hat Inc.
10#
11# Authors:
12#  Fam Zheng <famz@redhat.com>
13#
14# This work is licensed under the terms of the GNU GPL, version 2.  See
15# the COPYING file in the top-level directory.
16#
17# Based on qmp.py.
18#
19
20import os
21import socket
22from typing import (
23    List,
24    Optional,
25    Sequence,
26    TextIO,
27)
28
29from qemu.qmp import SocketAddrT  # pylint: disable=import-error
30
31from .machine import QEMUMachine
32
33
34class QEMUQtestProtocol:
35    """
36    QEMUQtestProtocol implements a connection to a qtest socket.
37
38    :param address: QEMU address, can be either a unix socket path (string)
39                    or a tuple in the form ( address, port ) for a TCP
40                    connection
41    :param server: server mode, listens on the socket (bool)
42    :raise socket.error: on socket connection errors
43
44    .. note::
45       No conection is estabalished by __init__(), this is done
46       by the connect() or accept() methods.
47    """
48    def __init__(self, address: SocketAddrT,
49                 server: bool = False):
50        self._address = address
51        self._sock = self._get_sock()
52        self._sockfile: Optional[TextIO] = None
53        if server:
54            self._sock.bind(self._address)
55            self._sock.listen(1)
56
57    def _get_sock(self) -> socket.socket:
58        if isinstance(self._address, tuple):
59            family = socket.AF_INET
60        else:
61            family = socket.AF_UNIX
62        return socket.socket(family, socket.SOCK_STREAM)
63
64    def connect(self) -> None:
65        """
66        Connect to the qtest socket.
67
68        @raise socket.error on socket connection errors
69        """
70        self._sock.connect(self._address)
71        self._sockfile = self._sock.makefile(mode='r')
72
73    def accept(self) -> None:
74        """
75        Await connection from QEMU.
76
77        @raise socket.error on socket connection errors
78        """
79        self._sock, _ = self._sock.accept()
80        self._sockfile = self._sock.makefile(mode='r')
81
82    def cmd(self, qtest_cmd: str) -> str:
83        """
84        Send a qtest command on the wire.
85
86        @param qtest_cmd: qtest command text to be sent
87        """
88        assert self._sockfile is not None
89        self._sock.sendall((qtest_cmd + "\n").encode('utf-8'))
90        resp = self._sockfile.readline()
91        return resp
92
93    def close(self) -> None:
94        """
95        Close this socket.
96        """
97        self._sock.close()
98        if self._sockfile:
99            self._sockfile.close()
100            self._sockfile = None
101
102    def settimeout(self, timeout: Optional[float]) -> None:
103        """Set a timeout, in seconds."""
104        self._sock.settimeout(timeout)
105
106
107class QEMUQtestMachine(QEMUMachine):
108    """
109    A QEMU VM, with a qtest socket available.
110    """
111
112    def __init__(self,
113                 binary: str,
114                 args: Sequence[str] = (),
115                 name: Optional[str] = None,
116                 base_temp_dir: str = "/var/tmp",
117                 socket_scm_helper: Optional[str] = None,
118                 sock_dir: Optional[str] = None):
119        # pylint: disable=too-many-arguments
120
121        if name is None:
122            name = "qemu-%d" % os.getpid()
123        if sock_dir is None:
124            sock_dir = base_temp_dir
125        super().__init__(binary, args, name=name, base_temp_dir=base_temp_dir,
126                         socket_scm_helper=socket_scm_helper,
127                         sock_dir=sock_dir)
128        self._qtest: Optional[QEMUQtestProtocol] = None
129        self._qtest_path = os.path.join(sock_dir, name + "-qtest.sock")
130
131    @property
132    def _base_args(self) -> List[str]:
133        args = super()._base_args
134        args.extend([
135            '-qtest', f"unix:path={self._qtest_path}",
136            '-accel', 'qtest'
137        ])
138        return args
139
140    def _pre_launch(self) -> None:
141        super()._pre_launch()
142        self._qtest = QEMUQtestProtocol(self._qtest_path, server=True)
143
144    def _post_launch(self) -> None:
145        assert self._qtest is not None
146        super()._post_launch()
147        self._qtest.accept()
148
149    def _post_shutdown(self) -> None:
150        super()._post_shutdown()
151        self._remove_if_exists(self._qtest_path)
152
153    def qtest(self, cmd: str) -> str:
154        """
155        Send a qtest command to the guest.
156
157        :param cmd: qtest command to send
158        :return: qtest server response
159        """
160        if self._qtest is None:
161            raise RuntimeError("qtest socket not available")
162        return self._qtest.cmd(cmd)
163