diff options
Diffstat (limited to 'tests/functional/qemu_test')
| -rw-r--r-- | tests/functional/qemu_test/cmd.py | 89 | ||||
| -rw-r--r-- | tests/functional/qemu_test/testcase.py | 43 | ||||
| -rw-r--r-- | tests/functional/qemu_test/tuxruntest.py | 17 |
3 files changed, 111 insertions, 38 deletions
diff --git a/tests/functional/qemu_test/cmd.py b/tests/functional/qemu_test/cmd.py index cbabb1ceed..11c8334a7c 100644 --- a/tests/functional/qemu_test/cmd.py +++ b/tests/functional/qemu_test/cmd.py @@ -78,13 +78,77 @@ def run_cmd(args): def is_readable_executable_file(path): return os.path.isfile(path) and os.access(path, os.R_OK | os.X_OK) +# @test: functional test to fail if @failure is seen +# @vm: the VM whose console to process +# @success: a non-None string to look for +# @failure: a string to look for that triggers test failure, or None +# +# Read up to 1 line of text from @vm, looking for @success +# and optionally @failure. +# +# If @success or @failure are seen, immediately return True, +# even if end of line is not yet seen. ie remainder of the +# line is left unread. +# +# If end of line is seen, with neither @success or @failure +# return False +# +# If @failure is seen, then mark @test as failed +def _console_read_line_until_match(test, vm, success, failure): + msg = bytes([]) + done = False + while True: + c = vm.console_socket.recv(1) + if c is None: + done = True + test.fail( + f"EOF in console, expected '{success}'") + break + msg += c + + if success in msg: + done = True + break + if failure and failure in msg: + done = True + vm.console_socket.close() + test.fail( + f"'{failure}' found in console, expected '{success}'") + + if c == b'\n': + break + + console_logger = logging.getLogger('console') + try: + console_logger.debug(msg.decode().strip()) + except: + console_logger.debug(msg) + + return done + def _console_interaction(test, success_message, failure_message, send_string, keep_sending=False, vm=None): assert not keep_sending or send_string + assert success_message or send_string + if vm is None: vm = test.vm - console = vm.console_file - console_logger = logging.getLogger('console') + + test.log.debug( + f"Console interaction: success_msg='{success_message}' " + + f"failure_msg='{failure_message}' send_string='{send_string}'") + + # We'll process console in bytes, to avoid having to + # deal with unicode decode errors from receiving + # partial utf8 byte sequences + success_message_b = None + if success_message is not None: + success_message_b = success_message.encode() + + failure_message_b = None + if failure_message is not None: + failure_message_b = failure_message.encode() + while True: if send_string: vm.console_socket.sendall(send_string.encode()) @@ -92,25 +156,15 @@ def _console_interaction(test, success_message, failure_message, send_string = None # send only once # Only consume console output if waiting for something - if success_message is None and failure_message is None: + if success_message is None: if send_string is None: break continue - try: - msg = console.readline().decode().strip() - except UnicodeDecodeError: - msg = None - if not msg: - continue - console_logger.debug(msg) - if success_message is None or success_message in msg: + if _console_read_line_until_match(test, vm, + success_message_b, + failure_message_b): break - if failure_message and failure_message in msg: - console.close() - fail = 'Failure message found in console: "%s". Expected: "%s"' % \ - (failure_message, success_message) - test.fail(fail) def interrupt_interactive_console_until_pattern(test, success_message, failure_message=None, @@ -135,6 +189,7 @@ def interrupt_interactive_console_until_pattern(test, success_message, :param interrupt_string: a string to send to the console before trying to read a new line """ + assert success_message _console_interaction(test, success_message, failure_message, interrupt_string, True) @@ -149,6 +204,7 @@ def wait_for_console_pattern(test, success_message, failure_message=None, :param success_message: if this message appears, test succeeds :param failure_message: if this message appears, test fails """ + assert success_message _console_interaction(test, success_message, failure_message, None, vm=vm) def exec_command(test, command): @@ -177,6 +233,7 @@ def exec_command_and_wait_for_pattern(test, command, :param success_message: if this message appears, test succeeds :param failure_message: if this message appears, test fails """ + assert success_message _console_interaction(test, success_message, failure_message, command + '\r') def get_qemu_img(test): diff --git a/tests/functional/qemu_test/testcase.py b/tests/functional/qemu_test/testcase.py index 411978b5ef..90ae59eb54 100644 --- a/tests/functional/qemu_test/testcase.py +++ b/tests/functional/qemu_test/testcase.py @@ -13,8 +13,9 @@ import logging import os -import subprocess import pycotap +import shutil +import subprocess import sys import unittest import uuid @@ -40,11 +41,12 @@ class QemuBaseTest(unittest.TestCase): self.assertIsNotNone(self.qemu_bin, 'QEMU_TEST_QEMU_BINARY must be set') self.arch = self.qemu_bin.split('-')[-1] - self.workdir = os.path.join(BUILD_DIR, 'tests/functional', self.arch, - self.id()) + self.outputdir = os.path.join(BUILD_DIR, 'tests', 'functional', + self.arch, self.id()) + self.workdir = os.path.join(self.outputdir, 'scratch') os.makedirs(self.workdir, exist_ok=True) - self.logdir = self.workdir + self.logdir = self.outputdir self.log_filename = os.path.join(self.logdir, 'base.log') self.log = logging.getLogger('qemu-test') self.log.setLevel(logging.DEBUG) @@ -55,7 +57,15 @@ class QemuBaseTest(unittest.TestCase): self._log_fh.setFormatter(fileFormatter) self.log.addHandler(self._log_fh) + # Capture QEMUMachine logging + self.machinelog = logging.getLogger('qemu.machine') + self.machinelog.setLevel(logging.DEBUG) + self.machinelog.addHandler(self._log_fh) + def tearDown(self): + if "QEMU_TEST_KEEP_SCRATCH" not in os.environ: + shutil.rmtree(self.workdir) + self.machinelog.removeHandler(self._log_fh) self.log.removeHandler(self._log_fh) def main(): @@ -71,10 +81,12 @@ class QemuBaseTest(unittest.TestCase): res = unittest.main(module = None, testRunner = tr, exit = False, argv=["__dummy__", path]) for (test, message) in res.result.errors + res.result.failures: - print('More information on ' + test.id() + ' could be found here:' - '\n %s' % test.log_filename, file=sys.stderr) - if hasattr(test, 'console_log_name'): - print(' %s' % test.console_log_name, file=sys.stderr) + + if hasattr(test, "log_filename"): + print('More information on ' + test.id() + ' could be found here:' + '\n %s' % test.log_filename, file=sys.stderr) + if hasattr(test, 'console_log_name'): + print(' %s' % test.console_log_name, file=sys.stderr) sys.exit(not res.result.wasSuccessful()) @@ -108,7 +120,7 @@ class QemuSystemTest(QemuBaseTest): console_log = logging.getLogger('console') console_log.setLevel(logging.DEBUG) - self.console_log_name = os.path.join(self.workdir, 'console.log') + self.console_log_name = os.path.join(self.logdir, 'console.log') self._console_log_fh = logging.FileHandler(self.console_log_name, mode='w') self._console_log_fh.setLevel(logging.DEBUG) @@ -159,10 +171,19 @@ class QemuSystemTest(QemuBaseTest): self.skipTest('no support for device ' + devicename) def _new_vm(self, name, *args): - vm = QEMUMachine(self.qemu_bin, base_temp_dir=self.workdir) + vm = QEMUMachine(self.qemu_bin, + name=name, + base_temp_dir=self.workdir, + log_dir=self.logdir) self.log.debug('QEMUMachine "%s" created', name) self.log.debug('QEMUMachine "%s" temp_dir: %s', name, vm.temp_dir) - self.log.debug('QEMUMachine "%s" log_dir: %s', name, vm.log_dir) + + sockpath = os.environ.get("QEMU_TEST_QMP_BACKDOOR", None) + if sockpath is not None: + vm.add_args("-chardev", + f"socket,id=backdoor,path={sockpath},server=on,wait=off", + "-mon", "chardev=backdoor,mode=control") + if args: vm.add_args(*args) return vm diff --git a/tests/functional/qemu_test/tuxruntest.py b/tests/functional/qemu_test/tuxruntest.py index f05aa96ad7..ab3b27da43 100644 --- a/tests/functional/qemu_test/tuxruntest.py +++ b/tests/functional/qemu_test/tuxruntest.py @@ -39,7 +39,6 @@ class TuxRunBaselineTest(QemuSystemTest): super().setUp() # We need zstd for all the tuxrun tests - # See https://github.com/avocado-framework/avocado/issues/5609 (has_zstd, msg) = has_cmd('zstd') if has_zstd is False: self.skipTest(msg) @@ -125,16 +124,12 @@ class TuxRunBaselineTest(QemuSystemTest): then do a few things on the console. Trigger a shutdown and wait to exit cleanly. """ - self.wait_for_console_pattern("Welcome to TuxTest") - time.sleep(0.2) - exec_command(self, 'root') - time.sleep(0.2) - exec_command(self, 'cat /proc/interrupts') - time.sleep(0.1) - exec_command(self, 'cat /proc/self/maps') - time.sleep(0.1) - exec_command(self, 'uname -a') - time.sleep(0.1) + ps1='root@tuxtest:~#' + self.wait_for_console_pattern('tuxtest login:') + exec_command_and_wait_for_pattern(self, 'root', ps1) + exec_command_and_wait_for_pattern(self, 'cat /proc/interrupts', ps1) + exec_command_and_wait_for_pattern(self, 'cat /proc/self/maps', ps1) + exec_command_and_wait_for_pattern(self, 'uname -a', ps1) exec_command_and_wait_for_pattern(self, 'halt', haltmsg) # Wait for VM to shut down gracefully if it can |