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