xref: /qemu/tests/functional/qemu_test/cmd.py (revision 9132fff802431438a2805389e74402321fb9afed) 
1# Test class and utilities for functional tests
2#
3# Copyright 2018, 2024 Red Hat, Inc.
4#
5# Original Author (Avocado-based tests):
6#  Cleber Rosa <crosa@redhat.com>
7#
8# Adaption for standalone version:
9#  Thomas Huth <thuth@redhat.com>
10#
11# This work is licensed under the terms of the GNU GPL, version 2 or
12# later.  See the COPYING file in the top-level directory.
13
14import logging
15import os
16import os.path
17import subprocess
18
19from .config import BUILD_DIR
20
21def which(tool):
22    """ looks up the full path for @tool, returns None if not found
23        or if @tool does not have executable permissions.
24    """
25    paths=os.getenv('PATH')
26    for p in paths.split(os.path.pathsep):
27        p = os.path.join(p, tool)
28        if os.access(p, os.X_OK):
29            return p
30    return None
31
32def run_cmd(args):
33    subp = subprocess.Popen(args,
34                            stdout=subprocess.PIPE,
35                            stderr=subprocess.PIPE,
36                            universal_newlines=True)
37    stdout, stderr = subp.communicate()
38    ret = subp.returncode
39
40    return (stdout, stderr, ret)
41
42def is_readable_executable_file(path):
43    return os.path.isfile(path) and os.access(path, os.R_OK | os.X_OK)
44
45# @test: functional test to fail if @failure is seen
46# @vm: the VM whose console to process
47# @success: a non-None string to look for
48# @failure: a string to look for that triggers test failure, or None
49#
50# Read up to 1 line of text from @vm, looking for @success
51# and optionally @failure.
52#
53# If @success or @failure are seen, immediately return True,
54# even if end of line is not yet seen. ie remainder of the
55# line is left unread.
56#
57# If end of line is seen, with neither @success or @failure
58# return False
59#
60# If @failure is seen, then mark @test as failed
61def _console_read_line_until_match(test, vm, success, failure):
62    msg = bytes([])
63    done = False
64    while True:
65        c = vm.console_socket.recv(1)
66        if c is None:
67            done = True
68            test.fail(
69                f"EOF in console, expected '{success}'")
70            break
71        msg += c
72
73        if success in msg:
74            done = True
75            break
76        if failure and failure in msg:
77            done = True
78            vm.console_socket.close()
79            test.fail(
80                f"'{failure}' found in console, expected '{success}'")
81
82        if c == b'\n':
83            break
84
85    console_logger = logging.getLogger('console')
86    try:
87        console_logger.debug(msg.decode().strip())
88    except:
89        console_logger.debug(msg)
90
91    return done
92
93def _console_interaction(test, success_message, failure_message,
94                         send_string, keep_sending=False, vm=None):
95    assert not keep_sending or send_string
96    assert success_message or send_string
97
98    if vm is None:
99        vm = test.vm
100
101    test.log.debug(
102        f"Console interaction: success_msg='{success_message}' " +
103        f"failure_msg='{failure_message}' send_string='{send_string}'")
104
105    # We'll process console in bytes, to avoid having to
106    # deal with unicode decode errors from receiving
107    # partial utf8 byte sequences
108    success_message_b = None
109    if success_message is not None:
110        success_message_b = success_message.encode()
111
112    failure_message_b = None
113    if failure_message is not None:
114        failure_message_b = failure_message.encode()
115
116    while True:
117        if send_string:
118            vm.console_socket.sendall(send_string.encode())
119            if not keep_sending:
120                send_string = None # send only once
121
122        # Only consume console output if waiting for something
123        if success_message is None:
124            if send_string is None:
125                break
126            continue
127
128        if _console_read_line_until_match(test, vm,
129                                          success_message_b,
130                                          failure_message_b):
131            break
132
133def interrupt_interactive_console_until_pattern(test, success_message,
134                                                failure_message=None,
135                                                interrupt_string='\r'):
136    """
137    Keep sending a string to interrupt a console prompt, while logging the
138    console output. Typical use case is to break a boot loader prompt, such:
139
140        Press a key within 5 seconds to interrupt boot process.
141        5
142        4
143        3
144        2
145        1
146        Booting default image...
147
148    :param test: a  test containing a VM that will have its console
149                 read and probed for a success or failure message
150    :type test: :class:`qemu_test.QemuSystemTest`
151    :param success_message: if this message appears, test succeeds
152    :param failure_message: if this message appears, test fails
153    :param interrupt_string: a string to send to the console before trying
154                             to read a new line
155    """
156    assert success_message
157    _console_interaction(test, success_message, failure_message,
158                         interrupt_string, True)
159
160def wait_for_console_pattern(test, success_message, failure_message=None,
161                             vm=None):
162    """
163    Waits for messages to appear on the console, while logging the content
164
165    :param test: a test containing a VM that will have its console
166                 read and probed for a success or failure message
167    :type test: :class:`qemu_test.QemuSystemTest`
168    :param success_message: if this message appears, test succeeds
169    :param failure_message: if this message appears, test fails
170    """
171    assert success_message
172    _console_interaction(test, success_message, failure_message, None, vm=vm)
173
174def exec_command(test, command):
175    """
176    Send a command to a console (appending CRLF characters), while logging
177    the content.
178
179    :param test: a test containing a VM.
180    :type test: :class:`qemu_test.QemuSystemTest`
181    :param command: the command to send
182    :type command: str
183    """
184    _console_interaction(test, None, None, command + '\r')
185
186def exec_command_and_wait_for_pattern(test, command,
187                                      success_message, failure_message=None):
188    """
189    Send a command to a console (appending CRLF characters), then wait
190    for success_message to appear on the console, while logging the.
191    content. Mark the test as failed if failure_message is found instead.
192
193    :param test: a test containing a VM that will have its console
194                 read and probed for a success or failure message
195    :type test: :class:`qemu_test.QemuSystemTest`
196    :param command: the command to send
197    :param success_message: if this message appears, test succeeds
198    :param failure_message: if this message appears, test fails
199    """
200    assert success_message
201    _console_interaction(test, success_message, failure_message, command + '\r')
202
203def get_qemu_img(test):
204    test.log.debug('Looking for and selecting a qemu-img binary')
205
206    # If qemu-img has been built, use it, otherwise the system wide one
207    # will be used.
208    qemu_img = os.path.join(BUILD_DIR, 'qemu-img')
209    if os.path.exists(qemu_img):
210        return qemu_img
211    qemu_img = which('qemu-img')
212    if qemu_img is not None:
213        return qemu_img
214    test.skipTest(f"qemu-img not found in {BUILD_DIR} or '$PATH'")
215