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