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 21 22def has_cmd(name, args=None): 23 """ 24 This function is for use in a @skipUnless decorator, e.g.: 25 26 @skipUnless(*has_cmd('sudo -n', ('sudo', '-n', 'true'))) 27 def test_something_that_needs_sudo(self): 28 ... 29 """ 30 31 if args is None: 32 args = ('which', name) 33 34 try: 35 _, stderr, exitcode = run_cmd(args) 36 except Exception as e: 37 exitcode = -1 38 stderr = str(e) 39 40 if exitcode != 0: 41 cmd_line = ' '.join(args) 42 err = f'{name} required, but "{cmd_line}" failed: {stderr.strip()}' 43 return (False, err) 44 else: 45 return (True, '') 46 47def has_cmds(*cmds): 48 """ 49 This function is for use in a @skipUnless decorator and 50 allows checking for the availability of multiple commands, e.g.: 51 52 @skipUnless(*has_cmds(('cmd1', ('cmd1', '--some-parameter')), 53 'cmd2', 'cmd3')) 54 def test_something_that_needs_cmd1_and_cmd2(self): 55 ... 56 """ 57 58 for cmd in cmds: 59 if isinstance(cmd, str): 60 cmd = (cmd,) 61 62 ok, errstr = has_cmd(*cmd) 63 if not ok: 64 return (False, errstr) 65 66 return (True, '') 67 68def run_cmd(args): 69 subp = subprocess.Popen(args, 70 stdout=subprocess.PIPE, 71 stderr=subprocess.PIPE, 72 universal_newlines=True) 73 stdout, stderr = subp.communicate() 74 ret = subp.returncode 75 76 return (stdout, stderr, ret) 77 78def is_readable_executable_file(path): 79 return os.path.isfile(path) and os.access(path, os.R_OK | os.X_OK) 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 if vm is None: 85 vm = test.vm 86 console = vm.console_file 87 console_logger = logging.getLogger('console') 88 test.log.debug( 89 f"Console interaction: success_msg='{success_message}' " + 90 f"failure_msg='{failure_message}' send_string='{send_string}'") 91 while True: 92 if send_string: 93 vm.console_socket.sendall(send_string.encode()) 94 if not keep_sending: 95 send_string = None # send only once 96 97 # Only consume console output if waiting for something 98 if success_message is None and failure_message is None: 99 if send_string is None: 100 break 101 continue 102 103 try: 104 msg = console.readline().decode().strip() 105 except UnicodeDecodeError: 106 msg = None 107 if not msg: 108 continue 109 console_logger.debug(msg) 110 if success_message is None or success_message in msg: 111 break 112 if failure_message and failure_message in msg: 113 console.close() 114 fail = 'Failure message found in console: "%s". Expected: "%s"' % \ 115 (failure_message, success_message) 116 test.fail(fail) 117 118def interrupt_interactive_console_until_pattern(test, success_message, 119 failure_message=None, 120 interrupt_string='\r'): 121 """ 122 Keep sending a string to interrupt a console prompt, while logging the 123 console output. Typical use case is to break a boot loader prompt, such: 124 125 Press a key within 5 seconds to interrupt boot process. 126 5 127 4 128 3 129 2 130 1 131 Booting default image... 132 133 :param test: a test containing a VM that will have its console 134 read and probed for a success or failure message 135 :type test: :class:`qemu_test.QemuSystemTest` 136 :param success_message: if this message appears, test succeeds 137 :param failure_message: if this message appears, test fails 138 :param interrupt_string: a string to send to the console before trying 139 to read a new line 140 """ 141 _console_interaction(test, success_message, failure_message, 142 interrupt_string, True) 143 144def wait_for_console_pattern(test, success_message, failure_message=None, 145 vm=None): 146 """ 147 Waits for messages to appear on the console, while logging the content 148 149 :param test: a test containing a VM that will have its console 150 read and probed for a success or failure message 151 :type test: :class:`qemu_test.QemuSystemTest` 152 :param success_message: if this message appears, test succeeds 153 :param failure_message: if this message appears, test fails 154 """ 155 _console_interaction(test, success_message, failure_message, None, vm=vm) 156 157def exec_command(test, command): 158 """ 159 Send a command to a console (appending CRLF characters), while logging 160 the content. 161 162 :param test: a test containing a VM. 163 :type test: :class:`qemu_test.QemuSystemTest` 164 :param command: the command to send 165 :type command: str 166 """ 167 _console_interaction(test, None, None, command + '\r') 168 169def exec_command_and_wait_for_pattern(test, command, 170 success_message, failure_message=None): 171 """ 172 Send a command to a console (appending CRLF characters), then wait 173 for success_message to appear on the console, while logging the. 174 content. Mark the test as failed if failure_message is found instead. 175 176 :param test: a test containing a VM that will have its console 177 read and probed for a success or failure message 178 :type test: :class:`qemu_test.QemuSystemTest` 179 :param command: the command to send 180 :param success_message: if this message appears, test succeeds 181 :param failure_message: if this message appears, test fails 182 """ 183 _console_interaction(test, success_message, failure_message, command + '\r') 184 185def get_qemu_img(test): 186 test.log.debug('Looking for and selecting a qemu-img binary') 187 188 # If qemu-img has been built, use it, otherwise the system wide one 189 # will be used. 190 qemu_img = os.path.join(BUILD_DIR, 'qemu-img') 191 if os.path.exists(qemu_img): 192 return qemu_img 193 (has_system_qemu_img, errmsg) = has_cmd('qemu-img') 194 if has_system_qemu_img: 195 return 'qemu-img' 196 test.skipTest(errmsg) 197