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# If @failure is seen, then mark @test as failed 49def _console_read_line_until_match(test, vm, success, failure): 50 msg = bytes([]) 51 done = False 52 while True: 53 c = vm.console_socket.recv(1) 54 if c is None: 55 done = True 56 test.fail( 57 f"EOF in console, expected '{success}'") 58 break 59 msg += c 60 61 if success in msg: 62 done = True 63 break 64 if failure and failure in msg: 65 done = True 66 vm.console_socket.close() 67 test.fail( 68 f"'{failure}' found in console, expected '{success}'") 69 70 if c == b'\n': 71 break 72 73 console_logger = logging.getLogger('console') 74 try: 75 console_logger.debug(msg.decode().strip()) 76 except: 77 console_logger.debug(msg) 78 79 return done 80 81def _console_interaction(test, success_message, failure_message, 82 send_string, keep_sending=False, vm=None): 83 assert not keep_sending or send_string 84 assert success_message or send_string 85 86 if vm is None: 87 vm = test.vm 88 89 test.log.debug( 90 f"Console interaction: success_msg='{success_message}' " + 91 f"failure_msg='{failure_message}' send_string='{send_string}'") 92 93 # We'll process console in bytes, to avoid having to 94 # deal with unicode decode errors from receiving 95 # partial utf8 byte sequences 96 success_message_b = None 97 if success_message is not None: 98 success_message_b = success_message.encode() 99 100 failure_message_b = None 101 if failure_message is not None: 102 failure_message_b = failure_message.encode() 103 104 while True: 105 if send_string: 106 vm.console_socket.sendall(send_string.encode()) 107 if not keep_sending: 108 send_string = None # send only once 109 110 # Only consume console output if waiting for something 111 if success_message is None: 112 if send_string is None: 113 break 114 continue 115 116 if _console_read_line_until_match(test, vm, 117 success_message_b, 118 failure_message_b): 119 break 120 121def interrupt_interactive_console_until_pattern(test, success_message, 122 failure_message=None, 123 interrupt_string='\r'): 124 """ 125 Keep sending a string to interrupt a console prompt, while logging the 126 console output. Typical use case is to break a boot loader prompt, such: 127 128 Press a key within 5 seconds to interrupt boot process. 129 5 130 4 131 3 132 2 133 1 134 Booting default image... 135 136 :param test: a test containing a VM that will have its console 137 read and probed for a success or failure message 138 :type test: :class:`qemu_test.QemuSystemTest` 139 :param success_message: if this message appears, test succeeds 140 :param failure_message: if this message appears, test fails 141 :param interrupt_string: a string to send to the console before trying 142 to read a new line 143 """ 144 assert success_message 145 _console_interaction(test, success_message, failure_message, 146 interrupt_string, True) 147 148def wait_for_console_pattern(test, success_message, failure_message=None, 149 vm=None): 150 """ 151 Waits for messages to appear on the console, while logging the content 152 153 :param test: a test containing a VM that will have its console 154 read and probed for a success or failure message 155 :type test: :class:`qemu_test.QemuSystemTest` 156 :param success_message: if this message appears, test succeeds 157 :param failure_message: if this message appears, test fails 158 """ 159 assert success_message 160 _console_interaction(test, success_message, failure_message, None, vm=vm) 161 162def exec_command(test, command): 163 """ 164 Send a command to a console (appending CRLF characters), while logging 165 the content. 166 167 :param test: a test containing a VM. 168 :type test: :class:`qemu_test.QemuSystemTest` 169 :param command: the command to send 170 :type command: str 171 """ 172 _console_interaction(test, None, None, command + '\r') 173 174def exec_command_and_wait_for_pattern(test, command, 175 success_message, failure_message=None): 176 """ 177 Send a command to a console (appending CRLF characters), then wait 178 for success_message to appear on the console, while logging the. 179 content. Mark the test as failed if failure_message is found instead. 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 command: the command to send 185 :param success_message: if this message appears, test succeeds 186 :param failure_message: if this message appears, test fails 187 """ 188 assert success_message 189 _console_interaction(test, success_message, failure_message, command + '\r') 190 191def get_qemu_img(test): 192 test.log.debug('Looking for and selecting a qemu-img binary') 193 194 # If qemu-img has been built, use it, otherwise the system wide one 195 # will be used. 196 qemu_img = test.build_file('qemu-img') 197 if os.path.exists(qemu_img): 198 return qemu_img 199 qemu_img = which('qemu-img') 200 if qemu_img is not None: 201 return qemu_img 202 test.skipTest(f"qemu-img not found in build dir or '$PATH'") 203