aboutsummaryrefslogtreecommitdiff
path: root/tests/avocado
diff options
context:
space:
mode:
Diffstat (limited to 'tests/avocado')
-rw-r--r--tests/avocado/README.rst10
-rw-r--r--tests/avocado/avocado_qemu/__init__.py424
-rw-r--r--tests/avocado/avocado_qemu/linuxtest.py253
-rw-r--r--tests/avocado/boot_linux.py132
-rw-r--r--tests/avocado/boot_linux_console.py96
-rw-r--r--tests/avocado/linux_ssh_mips_malta.py205
-rw-r--r--tests/avocado/replay_kernel.py110
-rw-r--r--tests/avocado/replay_linux.py206
-rw-r--r--tests/avocado/reverse_debugging.py272
-rw-r--r--tests/avocado/smmu.py139
10 files changed, 0 insertions, 1847 deletions
diff --git a/tests/avocado/README.rst b/tests/avocado/README.rst
deleted file mode 100644
index 9448837..0000000
--- a/tests/avocado/README.rst
+++ /dev/null
@@ -1,10 +0,0 @@
-=============================================
-Integration tests using the Avocado Framework
-=============================================
-
-This directory contains integration tests. They're usually higher
-level, and may interact with external resources and with various
-guest operating systems.
-
-For more information, please refer to ``docs/devel/testing.rst``,
-section "Integration tests using the Avocado Framework".
diff --git a/tests/avocado/avocado_qemu/__init__.py b/tests/avocado/avocado_qemu/__init__.py
deleted file mode 100644
index 93c3460..0000000
--- a/tests/avocado/avocado_qemu/__init__.py
+++ /dev/null
@@ -1,424 +0,0 @@
-# Test class and utilities for functional tests
-#
-# Copyright (c) 2018 Red Hat, Inc.
-#
-# Author:
-# Cleber Rosa <crosa@redhat.com>
-#
-# This work is licensed under the terms of the GNU GPL, version 2 or
-# later. See the COPYING file in the top-level directory.
-
-import logging
-import os
-import subprocess
-import sys
-import tempfile
-import time
-import uuid
-
-import avocado
-from avocado.utils import ssh
-from avocado.utils.path import find_command
-
-from qemu.machine import QEMUMachine
-from qemu.utils import (get_info_usernet_hostfwd_port, kvm_available,
- tcg_available)
-
-
-#: The QEMU build root directory. It may also be the source directory
-#: if building from the source dir, but it's safer to use BUILD_DIR for
-#: that purpose. Be aware that if this code is moved outside of a source
-#: and build tree, it will not be accurate.
-BUILD_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
-
-
-def has_cmd(name, args=None):
- """
- This function is for use in a @avocado.skipUnless decorator, e.g.:
-
- @skipUnless(*has_cmd('sudo -n', ('sudo', '-n', 'true')))
- def test_something_that_needs_sudo(self):
- ...
- """
-
- if args is None:
- args = ('which', name)
-
- try:
- _, stderr, exitcode = run_cmd(args)
- except Exception as e:
- exitcode = -1
- stderr = str(e)
-
- if exitcode != 0:
- cmd_line = ' '.join(args)
- err = f'{name} required, but "{cmd_line}" failed: {stderr.strip()}'
- return (False, err)
- else:
- return (True, '')
-
-def has_cmds(*cmds):
- """
- This function is for use in a @avocado.skipUnless decorator and
- allows checking for the availability of multiple commands, e.g.:
-
- @skipUnless(*has_cmds(('cmd1', ('cmd1', '--some-parameter')),
- 'cmd2', 'cmd3'))
- def test_something_that_needs_cmd1_and_cmd2(self):
- ...
- """
-
- for cmd in cmds:
- if isinstance(cmd, str):
- cmd = (cmd,)
-
- ok, errstr = has_cmd(*cmd)
- if not ok:
- return (False, errstr)
-
- return (True, '')
-
-def run_cmd(args):
- subp = subprocess.Popen(args,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- universal_newlines=True)
- stdout, stderr = subp.communicate()
- ret = subp.returncode
-
- return (stdout, stderr, ret)
-
-def is_readable_executable_file(path):
- return os.path.isfile(path) and os.access(path, os.R_OK | os.X_OK)
-
-
-def pick_default_qemu_bin(bin_prefix='qemu-system-', arch=None):
- """
- Picks the path of a QEMU binary, starting either in the current working
- directory or in the source tree root directory.
-
- :param arch: the arch to use when looking for a QEMU binary (the target
- will match the arch given). If None (the default), arch
- will be the current host system arch (as given by
- :func:`os.uname`).
- :type arch: str
- :returns: the path to the default QEMU binary or None if one could not
- be found
- :rtype: str or None
- """
- if arch is None:
- arch = os.uname()[4]
- # qemu binary path does not match arch for powerpc, handle it
- if 'ppc64le' in arch:
- arch = 'ppc64'
- qemu_bin_name = bin_prefix + arch
- qemu_bin_paths = [
- os.path.join(".", qemu_bin_name),
- os.path.join(BUILD_DIR, qemu_bin_name),
- os.path.join(BUILD_DIR, "build", qemu_bin_name),
- ]
- for path in qemu_bin_paths:
- if is_readable_executable_file(path):
- return path
- return None
-
-
-def _console_interaction(test, success_message, failure_message,
- send_string, keep_sending=False, vm=None):
- assert not keep_sending or send_string
- if vm is None:
- vm = test.vm
- console = vm.console_file
- console_logger = logging.getLogger('console')
- while True:
- if send_string:
- vm.console_socket.sendall(send_string.encode())
- if not keep_sending:
- 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 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:
- 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,
- interrupt_string='\r'):
- """
- Keep sending a string to interrupt a console prompt, while logging the
- console output. Typical use case is to break a boot loader prompt, such:
-
- Press a key within 5 seconds to interrupt boot process.
- 5
- 4
- 3
- 2
- 1
- Booting default image...
-
- :param test: an Avocado test containing a VM that will have its console
- read and probed for a success or failure message
- :type test: :class:`avocado_qemu.QemuSystemTest`
- :param success_message: if this message appears, test succeeds
- :param failure_message: if this message appears, test fails
- :param interrupt_string: a string to send to the console before trying
- to read a new line
- """
- _console_interaction(test, success_message, failure_message,
- interrupt_string, True)
-
-def wait_for_console_pattern(test, success_message, failure_message=None,
- vm=None):
- """
- Waits for messages to appear on the console, while logging the content
-
- :param test: an Avocado test containing a VM that will have its console
- read and probed for a success or failure message
- :type test: :class:`avocado_qemu.QemuSystemTest`
- :param success_message: if this message appears, test succeeds
- :param failure_message: if this message appears, test fails
- """
- _console_interaction(test, success_message, failure_message, None, vm=vm)
-
-def exec_command(test, command):
- """
- Send a command to a console (appending CRLF characters), while logging
- the content.
-
- :param test: an Avocado test containing a VM.
- :type test: :class:`avocado_qemu.QemuSystemTest`
- :param command: the command to send
- :type command: str
- """
- _console_interaction(test, None, None, command + '\r')
-
-def exec_command_and_wait_for_pattern(test, command,
- success_message, failure_message=None):
- """
- Send a command to a console (appending CRLF characters), then wait
- for success_message to appear on the console, while logging the.
- content. Mark the test as failed if failure_message is found instead.
-
- :param test: an Avocado test containing a VM that will have its console
- read and probed for a success or failure message
- :type test: :class:`avocado_qemu.QemuSystemTest`
- :param command: the command to send
- :param success_message: if this message appears, test succeeds
- :param failure_message: if this message appears, test fails
- """
- _console_interaction(test, success_message, failure_message, command + '\r')
-
-class QemuBaseTest(avocado.Test):
-
- # default timeout for all tests, can be overridden
- timeout = 120
-
- def _get_unique_tag_val(self, tag_name):
- """
- Gets a tag value, if unique for a key
- """
- vals = self.tags.get(tag_name, [])
- if len(vals) == 1:
- return vals.pop()
- return None
-
- def setUp(self, bin_prefix):
- self.arch = self.params.get('arch',
- default=self._get_unique_tag_val('arch'))
-
- self.cpu = self.params.get('cpu',
- default=self._get_unique_tag_val('cpu'))
-
- default_qemu_bin = pick_default_qemu_bin(bin_prefix, arch=self.arch)
- self.qemu_bin = self.params.get('qemu_bin',
- default=default_qemu_bin)
- if self.qemu_bin is None:
- self.cancel("No QEMU binary defined or found in the build tree")
-
- def fetch_asset(self, name,
- asset_hash, algorithm=None,
- locations=None, expire=None,
- find_only=False, cancel_on_missing=True):
- return super().fetch_asset(name,
- asset_hash=asset_hash,
- algorithm=algorithm,
- locations=locations,
- expire=expire,
- find_only=find_only,
- cancel_on_missing=cancel_on_missing)
-
-
-class QemuSystemTest(QemuBaseTest):
- """Facilitates system emulation tests."""
-
- def setUp(self):
- self._vms = {}
-
- super().setUp('qemu-system-')
-
- accel_required = self._get_unique_tag_val('accel')
- if accel_required:
- self.require_accelerator(accel_required)
-
- self.machine = self.params.get('machine',
- default=self._get_unique_tag_val('machine'))
-
- def require_accelerator(self, accelerator):
- """
- Requires an accelerator to be available for the test to continue
-
- It takes into account the currently set qemu binary.
-
- If the check fails, the test is canceled. If the check itself
- for the given accelerator is not available, the test is also
- canceled.
-
- :param accelerator: name of the accelerator, such as "kvm" or "tcg"
- :type accelerator: str
- """
- checker = {'tcg': tcg_available,
- 'kvm': kvm_available}.get(accelerator)
- if checker is None:
- self.cancel("Don't know how to check for the presence "
- "of accelerator %s" % accelerator)
- if not checker(qemu_bin=self.qemu_bin):
- self.cancel("%s accelerator does not seem to be "
- "available" % accelerator)
-
- def require_netdev(self, netdevname):
- netdevhelp = run_cmd([self.qemu_bin,
- '-M', 'none', '-netdev', 'help'])[0];
- if netdevhelp.find('\n' + netdevname + '\n') < 0:
- self.cancel('no support for user networking')
-
- def _new_vm(self, name, *args):
- self._sd = tempfile.TemporaryDirectory(prefix="qemu_")
- vm = QEMUMachine(self.qemu_bin, 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)
- if args:
- vm.add_args(*args)
- return vm
-
- def get_qemu_img(self):
- self.log.debug('Looking for and selecting a qemu-img binary')
-
- # If qemu-img has been built, use it, otherwise the system wide one
- # will be used.
- qemu_img = os.path.join(BUILD_DIR, 'qemu-img')
- if not os.path.exists(qemu_img):
- qemu_img = find_command('qemu-img', False)
- if qemu_img is False:
- self.cancel('Could not find "qemu-img"')
-
- return qemu_img
-
- @property
- def vm(self):
- return self.get_vm(name='default')
-
- def get_vm(self, *args, name=None):
- if not name:
- name = str(uuid.uuid4())
- if self._vms.get(name) is None:
- self._vms[name] = self._new_vm(name, *args)
- if self.cpu is not None:
- self._vms[name].add_args('-cpu', self.cpu)
- if self.machine is not None:
- self._vms[name].set_machine(self.machine)
- return self._vms[name]
-
- def set_vm_arg(self, arg, value):
- """
- Set an argument to list of extra arguments to be given to the QEMU
- binary. If the argument already exists then its value is replaced.
-
- :param arg: the QEMU argument, such as "-cpu" in "-cpu host"
- :type arg: str
- :param value: the argument value, such as "host" in "-cpu host"
- :type value: str
- """
- if not arg or not value:
- return
- if arg not in self.vm.args:
- self.vm.args.extend([arg, value])
- else:
- idx = self.vm.args.index(arg) + 1
- if idx < len(self.vm.args):
- self.vm.args[idx] = value
- else:
- self.vm.args.append(value)
-
- def tearDown(self):
- for vm in self._vms.values():
- vm.shutdown()
- self._sd = None
- super().tearDown()
-
-
-class LinuxSSHMixIn:
- """Contains utility methods for interacting with a guest via SSH."""
-
- def ssh_connect(self, username, credential, credential_is_key=True):
- self.ssh_logger = logging.getLogger('ssh')
- res = self.vm.cmd('human-monitor-command',
- command_line='info usernet')
- port = get_info_usernet_hostfwd_port(res)
- self.assertIsNotNone(port)
- self.assertGreater(port, 0)
- self.log.debug('sshd listening on port: %d', port)
- if credential_is_key:
- self.ssh_session = ssh.Session('127.0.0.1', port=port,
- user=username, key=credential)
- else:
- self.ssh_session = ssh.Session('127.0.0.1', port=port,
- user=username, password=credential)
- for i in range(10):
- try:
- self.ssh_session.connect()
- return
- except:
- time.sleep(i)
- self.fail('ssh connection timeout')
-
- def ssh_command(self, command):
- self.ssh_logger.info(command)
- result = self.ssh_session.cmd(command)
- stdout_lines = [line.rstrip() for line
- in result.stdout_text.splitlines()]
- for line in stdout_lines:
- self.ssh_logger.info(line)
- stderr_lines = [line.rstrip() for line
- in result.stderr_text.splitlines()]
- for line in stderr_lines:
- self.ssh_logger.warning(line)
-
- self.assertEqual(result.exit_status, 0,
- f'Guest command failed: {command}')
- return stdout_lines, stderr_lines
-
- def ssh_command_output_contains(self, cmd, exp):
- stdout, _ = self.ssh_command(cmd)
- for line in stdout:
- if exp in line:
- break
- else:
- self.fail('"%s" output does not contain "%s"' % (cmd, exp))
diff --git a/tests/avocado/avocado_qemu/linuxtest.py b/tests/avocado/avocado_qemu/linuxtest.py
deleted file mode 100644
index 66fb9f1..0000000
--- a/tests/avocado/avocado_qemu/linuxtest.py
+++ /dev/null
@@ -1,253 +0,0 @@
-# Test class and utilities for functional Linux-based tests
-#
-# Copyright (c) 2018 Red Hat, Inc.
-#
-# Author:
-# Cleber Rosa <crosa@redhat.com>
-#
-# This work is licensed under the terms of the GNU GPL, version 2 or
-# later. See the COPYING file in the top-level directory.
-
-import os
-import shutil
-
-from avocado.utils import cloudinit, datadrainer, process, vmimage
-
-from avocado_qemu import LinuxSSHMixIn
-from avocado_qemu import QemuSystemTest
-
-if os.path.islink(os.path.dirname(os.path.dirname(__file__))):
- # The link to the avocado tests dir in the source code directory
- lnk = os.path.dirname(os.path.dirname(__file__))
- #: The QEMU root source directory
- SOURCE_DIR = os.path.dirname(os.path.dirname(os.readlink(lnk)))
-else:
- SOURCE_DIR = BUILD_DIR
-
-class LinuxDistro:
- """Represents a Linux distribution
-
- Holds information of known distros.
- """
- #: A collection of known distros and their respective image checksum
- KNOWN_DISTROS = {
- 'fedora': {
- '31': {
- 'x86_64':
- {'checksum': ('e3c1b309d9203604922d6e255c2c5d09'
- '8a309c2d46215d8fc026954f3c5c27a0'),
- 'pxeboot_url': ('https://archives.fedoraproject.org/'
- 'pub/archive/fedora/linux/releases/31/'
- 'Everything/x86_64/os/images/pxeboot/'),
- 'kernel_params': ('root=UUID=b1438b9b-2cab-4065-a99a-'
- '08a96687f73c ro no_timer_check '
- 'net.ifnames=0 console=tty1 '
- 'console=ttyS0,115200n8'),
- },
- 'aarch64':
- {'checksum': ('1e18d9c0cf734940c4b5d5ec592facae'
- 'd2af0ad0329383d5639c997fdf16fe49'),
- 'pxeboot_url': 'https://archives.fedoraproject.org/'
- 'pub/archive/fedora/linux/releases/31/'
- 'Everything/aarch64/os/images/pxeboot/',
- 'kernel_params': ('root=UUID=b6950a44-9f3c-4076-a9c2-'
- '355e8475b0a7 ro earlyprintk=pl011,0x9000000'
- ' ignore_loglevel no_timer_check'
- ' printk.time=1 rd_NO_PLYMOUTH'
- ' console=ttyAMA0'),
- },
- 'ppc64':
- {'checksum': ('7c3528b85a3df4b2306e892199a9e1e4'
- '3f991c506f2cc390dc4efa2026ad2f58')},
- 's390x':
- {'checksum': ('4caaab5a434fd4d1079149a072fdc789'
- '1e354f834d355069ca982fdcaf5a122d')},
- },
- '32': {
- 'aarch64':
- {'checksum': ('b367755c664a2d7a26955bbfff985855'
- 'adfa2ca15e908baf15b4b176d68d3967'),
- 'pxeboot_url': ('http://dl.fedoraproject.org/pub/fedora/linux/'
- 'releases/32/Server/aarch64/os/images/'
- 'pxeboot/'),
- 'kernel_params': ('root=UUID=3df75b65-be8d-4db4-8655-'
- '14d95c0e90c5 ro no_timer_check net.ifnames=0'
- ' console=tty1 console=ttyS0,115200n8'),
- },
- },
- '33': {
- 'aarch64':
- {'checksum': ('e7f75cdfd523fe5ac2ca9eeece68edc1'
- 'a81f386a17f969c1d1c7c87031008a6b'),
- 'pxeboot_url': ('http://dl.fedoraproject.org/pub/fedora/linux/'
- 'releases/33/Server/aarch64/os/images/'
- 'pxeboot/'),
- 'kernel_params': ('root=UUID=d20b3ffa-6397-4a63-a734-'
- '1126a0208f8a ro no_timer_check net.ifnames=0'
- ' console=tty1 console=ttyS0,115200n8'
- ' console=tty0'),
- },
- },
- }
- }
-
- def __init__(self, name, version, arch):
- self.name = name
- self.version = version
- self.arch = arch
- try:
- info = self.KNOWN_DISTROS.get(name).get(version).get(arch)
- except AttributeError:
- # Unknown distro
- info = None
- self._info = info or {}
-
- @property
- def checksum(self):
- """Gets the cloud-image file checksum"""
- return self._info.get('checksum', None)
-
- @checksum.setter
- def checksum(self, value):
- self._info['checksum'] = value
-
- @property
- def pxeboot_url(self):
- """Gets the repository url where pxeboot files can be found"""
- return self._info.get('pxeboot_url', None)
-
- @property
- def default_kernel_params(self):
- """Gets the default kernel parameters"""
- return self._info.get('kernel_params', None)
-
-
-class LinuxTest(LinuxSSHMixIn, QemuSystemTest):
- """Facilitates having a cloud-image Linux based available.
-
- For tests that intend to interact with guests, this is a better choice
- to start with than the more vanilla `QemuSystemTest` class.
- """
-
- distro = None
- username = 'root'
- password = 'password'
- smp = '2'
- memory = '1024'
-
- def _set_distro(self):
- distro_name = self.params.get(
- 'distro',
- default=self._get_unique_tag_val('distro'))
- if not distro_name:
- distro_name = 'fedora'
-
- distro_version = self.params.get(
- 'distro_version',
- default=self._get_unique_tag_val('distro_version'))
- if not distro_version:
- distro_version = '31'
-
- self.distro = LinuxDistro(distro_name, distro_version, self.arch)
-
- # The distro checksum behaves differently than distro name and
- # version. First, it does not respect a tag with the same
- # name, given that it's not expected to be used for filtering
- # (distro name versions are the natural choice). Second, the
- # order of precedence is: parameter, attribute and then value
- # from KNOWN_DISTROS.
- distro_checksum = self.params.get('distro_checksum',
- default=None)
- if distro_checksum:
- self.distro.checksum = distro_checksum
-
- def setUp(self, ssh_pubkey=None, network_device_type='virtio-net'):
- super().setUp()
- self.require_netdev('user')
- self._set_distro()
- self.vm.add_args('-smp', self.smp)
- self.vm.add_args('-m', self.memory)
- # The following network device allows for SSH connections
- self.vm.add_args('-netdev', 'user,id=vnet,hostfwd=:127.0.0.1:0-:22',
- '-device', '%s,netdev=vnet' % network_device_type)
- self.set_up_boot()
- if ssh_pubkey is None:
- ssh_pubkey, self.ssh_key = self.set_up_existing_ssh_keys()
- self.set_up_cloudinit(ssh_pubkey)
-
- def set_up_existing_ssh_keys(self):
- ssh_public_key = os.path.join(SOURCE_DIR, 'tests', 'keys', 'id_rsa.pub')
- source_private_key = os.path.join(SOURCE_DIR, 'tests', 'keys', 'id_rsa')
- ssh_dir = os.path.join(self.workdir, '.ssh')
- os.mkdir(ssh_dir, mode=0o700)
- ssh_private_key = os.path.join(ssh_dir,
- os.path.basename(source_private_key))
- shutil.copyfile(source_private_key, ssh_private_key)
- os.chmod(ssh_private_key, 0o600)
- return (ssh_public_key, ssh_private_key)
-
- def download_boot(self):
- # Set the qemu-img binary.
- # If none is available, the test will cancel.
- vmimage.QEMU_IMG = super().get_qemu_img()
-
- self.log.info('Downloading/preparing boot image')
- # Fedora 31 only provides ppc64le images
- image_arch = self.arch
- if self.distro.name == 'fedora':
- if image_arch == 'ppc64':
- image_arch = 'ppc64le'
-
- try:
- boot = vmimage.get(
- self.distro.name, arch=image_arch, version=self.distro.version,
- checksum=self.distro.checksum,
- algorithm='sha256',
- cache_dir=self.cache_dirs[0],
- snapshot_dir=self.workdir)
- except:
- self.cancel('Failed to download/prepare boot image')
- return boot.path
-
- def prepare_cloudinit(self, ssh_pubkey=None):
- self.log.info('Preparing cloudinit image')
- try:
- cloudinit_iso = os.path.join(self.workdir, 'cloudinit.iso')
- pubkey_content = None
- if ssh_pubkey:
- with open(ssh_pubkey) as pubkey:
- pubkey_content = pubkey.read()
- cloudinit.iso(cloudinit_iso, self.name,
- username=self.username,
- password=self.password,
- # QEMU's hard coded usermode router address
- phone_home_host='10.0.2.2',
- phone_home_port=self.phone_server.server_port,
- authorized_key=pubkey_content)
- except Exception:
- self.cancel('Failed to prepare the cloudinit image')
- return cloudinit_iso
-
- def set_up_boot(self):
- path = self.download_boot()
- self.vm.add_args('-drive', 'file=%s' % path)
-
- def set_up_cloudinit(self, ssh_pubkey=None):
- self.phone_server = cloudinit.PhoneHomeServer(('0.0.0.0', 0),
- self.name)
- cloudinit_iso = self.prepare_cloudinit(ssh_pubkey)
- self.vm.add_args('-drive', 'file=%s,format=raw' % cloudinit_iso)
-
- def launch_and_wait(self, set_up_ssh_connection=True):
- self.vm.set_console()
- self.vm.launch()
- console_drainer = datadrainer.LineLogger(self.vm.console_socket.fileno(),
- logger=self.log.getChild('console'))
- console_drainer.start()
- self.log.info('VM launched, waiting for boot confirmation from guest')
- while not self.phone_server.instance_phoned_back:
- self.phone_server.handle_request()
-
- if set_up_ssh_connection:
- self.log.info('Setting up the SSH connection')
- self.ssh_connect(self.username, self.ssh_key)
diff --git a/tests/avocado/boot_linux.py b/tests/avocado/boot_linux.py
deleted file mode 100644
index a029ef4..0000000
--- a/tests/avocado/boot_linux.py
+++ /dev/null
@@ -1,132 +0,0 @@
-# Functional test that boots a complete Linux system via a cloud image
-#
-# Copyright (c) 2018-2020 Red Hat, Inc.
-#
-# Author:
-# Cleber Rosa <crosa@redhat.com>
-#
-# This work is licensed under the terms of the GNU GPL, version 2 or
-# later. See the COPYING file in the top-level directory.
-
-import os
-
-from avocado_qemu.linuxtest import LinuxTest
-from avocado_qemu import BUILD_DIR
-
-from avocado import skipUnless
-
-
-class BootLinuxX8664(LinuxTest):
- """
- :avocado: tags=arch:x86_64
- """
- timeout = 480
-
- def test_pc_i440fx_tcg(self):
- """
- :avocado: tags=machine:pc
- :avocado: tags=accel:tcg
- """
- self.require_accelerator("tcg")
- self.vm.add_args("-accel", "tcg")
- self.launch_and_wait(set_up_ssh_connection=False)
-
- def test_pc_i440fx_kvm(self):
- """
- :avocado: tags=machine:pc
- :avocado: tags=accel:kvm
- """
- self.require_accelerator("kvm")
- self.vm.add_args("-accel", "kvm")
- self.launch_and_wait(set_up_ssh_connection=False)
-
- def test_pc_q35_tcg(self):
- """
- :avocado: tags=machine:q35
- :avocado: tags=accel:tcg
- """
- self.require_accelerator("tcg")
- self.vm.add_args("-accel", "tcg")
- self.launch_and_wait(set_up_ssh_connection=False)
-
- def test_pc_q35_kvm(self):
- """
- :avocado: tags=machine:q35
- :avocado: tags=accel:kvm
- """
- self.require_accelerator("kvm")
- self.vm.add_args("-accel", "kvm")
- self.launch_and_wait(set_up_ssh_connection=False)
-
-
-# For Aarch64 we only boot KVM tests in CI as booting the current
-# Fedora OS in TCG tests is very heavyweight. There are lighter weight
-# distros which we use in the machine_aarch64_virt.py tests.
-class BootLinuxAarch64(LinuxTest):
- """
- :avocado: tags=arch:aarch64
- :avocado: tags=machine:virt
- """
- timeout = 720
-
- def test_virt_kvm(self):
- """
- :avocado: tags=accel:kvm
- :avocado: tags=cpu:host
- """
- self.require_accelerator("kvm")
- self.vm.add_args("-accel", "kvm")
- self.vm.add_args("-machine", "virt,gic-version=host")
- self.vm.add_args('-bios',
- os.path.join(BUILD_DIR, 'pc-bios',
- 'edk2-aarch64-code.fd'))
- self.vm.add_args('-device', 'virtio-rng-pci,rng=rng0')
- self.vm.add_args('-object', 'rng-random,id=rng0,filename=/dev/urandom')
- self.launch_and_wait(set_up_ssh_connection=False)
-
-
-# See the tux_baseline.py tests for almost the same coverage in a lot
-# less time.
-class BootLinuxPPC64(LinuxTest):
- """
- :avocado: tags=arch:ppc64
- """
-
- timeout = 360
-
- @skipUnless(os.getenv('SPEED') == 'slow', 'runtime limited')
- def test_pseries_tcg(self):
- """
- :avocado: tags=machine:pseries
- :avocado: tags=accel:tcg
- """
- self.require_accelerator("tcg")
- self.vm.add_args("-accel", "tcg")
- self.launch_and_wait(set_up_ssh_connection=False)
-
- def test_pseries_kvm(self):
- """
- :avocado: tags=machine:pseries
- :avocado: tags=accel:kvm
- """
- self.require_accelerator("kvm")
- self.vm.add_args("-accel", "kvm")
- self.vm.add_args("-machine", "cap-ccf-assist=off")
- self.launch_and_wait(set_up_ssh_connection=False)
-
-class BootLinuxS390X(LinuxTest):
- """
- :avocado: tags=arch:s390x
- """
-
- timeout = 240
-
- @skipUnless(os.getenv('SPEED') == 'slow', 'runtime limited')
- def test_s390_ccw_virtio_tcg(self):
- """
- :avocado: tags=machine:s390-ccw-virtio
- :avocado: tags=accel:tcg
- """
- self.require_accelerator("tcg")
- self.vm.add_args("-accel", "tcg")
- self.launch_and_wait(set_up_ssh_connection=False)
diff --git a/tests/avocado/boot_linux_console.py b/tests/avocado/boot_linux_console.py
deleted file mode 100644
index c15f39a..0000000
--- a/tests/avocado/boot_linux_console.py
+++ /dev/null
@@ -1,96 +0,0 @@
-# Functional test that boots a Linux kernel and checks the console
-#
-# Copyright (c) 2018 Red Hat, Inc.
-#
-# Author:
-# Cleber Rosa <crosa@redhat.com>
-#
-# This work is licensed under the terms of the GNU GPL, version 2 or
-# later. See the COPYING file in the top-level directory.
-
-import os
-import lzma
-import gzip
-import shutil
-
-from avocado import skip
-from avocado import skipUnless
-from avocado import skipUnless
-from avocado_qemu import QemuSystemTest
-from avocado_qemu import exec_command
-from avocado_qemu import exec_command_and_wait_for_pattern
-from avocado_qemu import interrupt_interactive_console_until_pattern
-from avocado_qemu import wait_for_console_pattern
-from avocado.utils import process
-from avocado.utils import archive
-
-class LinuxKernelTest(QemuSystemTest):
- KERNEL_COMMON_COMMAND_LINE = 'printk.time=0 '
-
- def wait_for_console_pattern(self, success_message, vm=None):
- wait_for_console_pattern(self, success_message,
- failure_message='Kernel panic - not syncing',
- vm=vm)
-
- def extract_from_deb(self, deb, path):
- """
- Extracts a file from a deb package into the test workdir
-
- :param deb: path to the deb archive
- :param path: path within the deb archive of the file to be extracted
- :returns: path of the extracted file
- """
- cwd = os.getcwd()
- os.chdir(self.workdir)
- file_path = process.run("ar t %s" % deb).stdout_text.split()[2]
- process.run("ar x %s %s" % (deb, file_path))
- archive.extract(file_path, self.workdir)
- os.chdir(cwd)
- # Return complete path to extracted file. Because callers to
- # extract_from_deb() specify 'path' with a leading slash, it is
- # necessary to use os.path.relpath() as otherwise os.path.join()
- # interprets it as an absolute path and drops the self.workdir part.
- return os.path.normpath(os.path.join(self.workdir,
- os.path.relpath(path, '/')))
-
- def extract_from_rpm(self, rpm, path):
- """
- Extracts a file from an RPM package into the test workdir.
-
- :param rpm: path to the rpm archive
- :param path: path within the rpm archive of the file to be extracted
- needs to be a relative path (starting with './') because
- cpio(1), which is used to extract the file, expects that.
- :returns: path of the extracted file
- """
- cwd = os.getcwd()
- os.chdir(self.workdir)
- process.run("rpm2cpio %s | cpio -id %s" % (rpm, path), shell=True)
- os.chdir(cwd)
- return os.path.normpath(os.path.join(self.workdir, path))
-
-class BootLinuxConsole(LinuxKernelTest):
- """
- Boots a Linux kernel and checks that the console is operational and the
- kernel command line is properly passed from QEMU to the kernel
- """
- timeout = 90
-
- def test_x86_64_pc(self):
- """
- :avocado: tags=arch:x86_64
- :avocado: tags=machine:pc
- """
- kernel_url = ('https://archives.fedoraproject.org/pub/archive/fedora'
- '/linux/releases/29/Everything/x86_64/os/images/pxeboot'
- '/vmlinuz')
- kernel_hash = '23bebd2680757891cf7adedb033532163a792495'
- kernel_path = self.fetch_asset(kernel_url, asset_hash=kernel_hash)
-
- self.vm.set_console()
- kernel_command_line = self.KERNEL_COMMON_COMMAND_LINE + 'console=ttyS0'
- self.vm.add_args('-kernel', kernel_path,
- '-append', kernel_command_line)
- self.vm.launch()
- console_pattern = 'Kernel command line: %s' % kernel_command_line
- self.wait_for_console_pattern(console_pattern)
diff --git a/tests/avocado/linux_ssh_mips_malta.py b/tests/avocado/linux_ssh_mips_malta.py
deleted file mode 100644
index d9bb525..0000000
--- a/tests/avocado/linux_ssh_mips_malta.py
+++ /dev/null
@@ -1,205 +0,0 @@
-# Functional test that boots a VM and run commands via a SSH session
-#
-# Copyright (c) Philippe Mathieu-Daudé <f4bug@amsat.org>
-#
-# This work is licensed under the terms of the GNU GPL, version 2 or
-# later. See the COPYING file in the top-level directory.
-
-import os
-import re
-import base64
-import logging
-import time
-
-from avocado import skipUnless
-from avocado_qemu import LinuxSSHMixIn
-from avocado_qemu import QemuSystemTest
-from avocado_qemu import wait_for_console_pattern
-from avocado.utils import process
-from avocado.utils import archive
-from avocado.utils import ssh
-
-
-@skipUnless(os.getenv('AVOCADO_TIMEOUT_EXPECTED'), 'Test might timeout')
-@skipUnless(ssh.SSH_CLIENT_BINARY, 'No SSH client available')
-class LinuxSSH(QemuSystemTest, LinuxSSHMixIn):
- """
- :avocado: tags=accel:tcg
- """
-
- timeout = 150 # Not for 'configure --enable-debug --enable-debug-tcg'
-
- KERNEL_COMMON_COMMAND_LINE = 'printk.time=0 '
- VM_IP = '127.0.0.1'
-
- BASE_URL = 'https://people.debian.org/~aurel32/qemu/'
- IMAGE_INFO = {
- 'be': {'base_url': 'mips',
- 'image_name': 'debian_wheezy_mips_standard.qcow2',
- 'image_hash': '8987a63270df67345b2135a6b7a4885a35e392d5',
- 'kernel_hash': {
- 32: '592e384a4edc16dade52a6cd5c785c637bcbc9ad',
- 64: 'db6eea7de35d36c77d8c165b6bcb222e16eb91db'}
- },
- 'le': {'base_url': 'mipsel',
- 'image_name': 'debian_wheezy_mipsel_standard.qcow2',
- 'image_hash': '7866764d9de3ef536ffca24c9fb9f04ffdb45802',
- 'kernel_hash': {
- 32: 'a66bea5a8adaa2cb3d36a1d4e0ccdb01be8f6c2a',
- 64: '6a7f77245acf231415a0e8b725d91ed2f3487794'}
- }
- }
- CPU_INFO = {
- 32: {'cpu': 'MIPS 24Kc', 'kernel_release': '3.2.0-4-4kc-malta'},
- 64: {'cpu': 'MIPS 20Kc', 'kernel_release': '3.2.0-4-5kc-malta'}
- }
-
- def get_url(self, endianess, path=''):
- qkey = {'le': 'el', 'be': ''}
- return '%s/mips%s/%s' % (self.BASE_URL, qkey[endianess], path)
-
- def get_image_info(self, endianess):
- dinfo = self.IMAGE_INFO[endianess]
- image_url = self.get_url(endianess, dinfo['image_name'])
- image_hash = dinfo['image_hash']
- return (image_url, image_hash)
-
- def get_kernel_info(self, endianess, wordsize):
- minfo = self.CPU_INFO[wordsize]
- kernel_url = self.get_url(endianess,
- 'vmlinux-%s' % minfo['kernel_release'])
- kernel_hash = self.IMAGE_INFO[endianess]['kernel_hash'][wordsize]
- return kernel_url, kernel_hash
-
- def ssh_disconnect_vm(self):
- self.ssh_session.quit()
-
- def boot_debian_wheezy_image_and_ssh_login(self, endianess, kernel_path):
- image_url, image_hash = self.get_image_info(endianess)
- image_path = self.fetch_asset(image_url, asset_hash=image_hash)
-
- self.vm.set_console()
- kernel_command_line = (self.KERNEL_COMMON_COMMAND_LINE
- + 'console=ttyS0 root=/dev/sda1')
- self.vm.add_args('-no-reboot',
- '-kernel', kernel_path,
- '-append', kernel_command_line,
- '-drive', 'file=%s,snapshot=on' % image_path,
- '-netdev', 'user,id=vnet,hostfwd=:127.0.0.1:0-:22',
- '-device', 'pcnet,netdev=vnet')
- self.vm.launch()
-
- self.log.info('VM launched, waiting for sshd')
- console_pattern = 'Starting OpenBSD Secure Shell server: sshd'
- wait_for_console_pattern(self, console_pattern, 'Oops')
- self.log.info('sshd ready')
-
- self.ssh_connect('root', 'root', False)
-
- def shutdown_via_ssh(self):
- self.ssh_command('poweroff')
- self.ssh_disconnect_vm()
- wait_for_console_pattern(self, 'Power down', 'Oops')
-
- def run_common_commands(self, wordsize):
- self.ssh_command_output_contains(
- 'cat /proc/cpuinfo',
- self.CPU_INFO[wordsize]['cpu'])
- self.ssh_command_output_contains(
- 'uname -m',
- 'mips')
- self.ssh_command_output_contains(
- 'uname -r',
- self.CPU_INFO[wordsize]['kernel_release'])
- self.ssh_command_output_contains(
- 'cat /proc/interrupts',
- 'XT-PIC timer')
- self.ssh_command_output_contains(
- 'cat /proc/interrupts',
- 'XT-PIC i8042')
- self.ssh_command_output_contains(
- 'cat /proc/interrupts',
- 'XT-PIC serial')
- self.ssh_command_output_contains(
- 'cat /proc/interrupts',
- 'XT-PIC ata_piix')
- self.ssh_command_output_contains(
- 'cat /proc/interrupts',
- 'XT-PIC eth0')
- self.ssh_command_output_contains(
- 'cat /proc/devices',
- 'input')
- self.ssh_command_output_contains(
- 'cat /proc/devices',
- 'usb')
- self.ssh_command_output_contains(
- 'cat /proc/devices',
- 'fb')
- self.ssh_command_output_contains(
- 'cat /proc/ioports',
- ' : serial')
- self.ssh_command_output_contains(
- 'cat /proc/ioports',
- ' : ata_piix')
- self.ssh_command_output_contains(
- 'cat /proc/ioports',
- ' : piix4_smbus')
- self.ssh_command_output_contains(
- 'lspci -d 11ab:4620',
- 'GT-64120')
- self.ssh_command_output_contains(
- 'cat /sys/bus/i2c/devices/i2c-0/name',
- 'SMBus PIIX4 adapter')
- self.ssh_command_output_contains(
- 'cat /proc/mtd',
- 'YAMON')
- # Empty 'Board Config' (64KB)
- self.ssh_command_output_contains(
- 'md5sum /dev/mtd2ro',
- '0dfbe8aa4c20b52e1b8bf3cb6cbdf193')
-
- def check_mips_malta(self, uname_m, endianess):
- wordsize = 64 if '64' in uname_m else 32
- kernel_url, kernel_hash = self.get_kernel_info(endianess, wordsize)
- kernel_path = self.fetch_asset(kernel_url, asset_hash=kernel_hash)
- self.boot_debian_wheezy_image_and_ssh_login(endianess, kernel_path)
-
- stdout, _ = self.ssh_command('uname -a')
- self.assertIn(True, [uname_m + " GNU/Linux" in line for line in stdout])
-
- self.run_common_commands(wordsize)
- self.shutdown_via_ssh()
- # Wait for VM to shut down gracefully
- self.vm.wait()
-
- def test_mips_malta32eb_kernel3_2_0(self):
- """
- :avocado: tags=arch:mips
- :avocado: tags=endian:big
- :avocado: tags=device:pcnet32
- """
- self.check_mips_malta('mips', 'be')
-
- def test_mips_malta32el_kernel3_2_0(self):
- """
- :avocado: tags=arch:mipsel
- :avocado: tags=endian:little
- :avocado: tags=device:pcnet32
- """
- self.check_mips_malta('mips', 'le')
-
- def test_mips_malta64eb_kernel3_2_0(self):
- """
- :avocado: tags=arch:mips64
- :avocado: tags=endian:big
- :avocado: tags=device:pcnet32
- """
- self.check_mips_malta('mips64', 'be')
-
- def test_mips_malta64el_kernel3_2_0(self):
- """
- :avocado: tags=arch:mips64el
- :avocado: tags=endian:little
- :avocado: tags=device:pcnet32
- """
- self.check_mips_malta('mips64', 'le')
diff --git a/tests/avocado/replay_kernel.py b/tests/avocado/replay_kernel.py
deleted file mode 100644
index 3551532..0000000
--- a/tests/avocado/replay_kernel.py
+++ /dev/null
@@ -1,110 +0,0 @@
-# Record/replay test that boots a Linux kernel
-#
-# Copyright (c) 2020 ISP RAS
-#
-# Author:
-# Pavel Dovgalyuk <Pavel.Dovgaluk@ispras.ru>
-#
-# This work is licensed under the terms of the GNU GPL, version 2 or
-# later. See the COPYING file in the top-level directory.
-
-import os
-import lzma
-import shutil
-import logging
-import time
-import subprocess
-
-from avocado import skip
-from avocado import skipUnless
-from avocado import skipUnless
-from avocado_qemu import wait_for_console_pattern
-from avocado.utils import archive
-from avocado.utils import process
-from boot_linux_console import LinuxKernelTest
-
-class ReplayKernelBase(LinuxKernelTest):
- """
- Boots a Linux kernel in record mode and checks that the console
- is operational and the kernel command line is properly passed
- from QEMU to the kernel.
- Then replays the same scenario and verifies, that QEMU correctly
- terminates.
- """
-
- timeout = 180
- KERNEL_COMMON_COMMAND_LINE = 'printk.time=1 panic=-1 '
-
- def run_vm(self, kernel_path, kernel_command_line, console_pattern,
- record, shift, args, replay_path):
- # icount requires TCG to be available
- self.require_accelerator('tcg')
-
- logger = logging.getLogger('replay')
- start_time = time.time()
- vm = self.get_vm()
- vm.set_console()
- if record:
- logger.info('recording the execution...')
- mode = 'record'
- else:
- logger.info('replaying the execution...')
- mode = 'replay'
- vm.add_args('-icount', 'shift=%s,rr=%s,rrfile=%s' %
- (shift, mode, replay_path),
- '-kernel', kernel_path,
- '-append', kernel_command_line,
- '-net', 'none',
- '-no-reboot')
- if args:
- vm.add_args(*args)
- vm.launch()
- self.wait_for_console_pattern(console_pattern, vm)
- if record:
- vm.shutdown()
- logger.info('finished the recording with log size %s bytes'
- % os.path.getsize(replay_path))
- self.run_replay_dump(replay_path)
- logger.info('successfully tested replay-dump.py')
- else:
- vm.wait()
- logger.info('successfully finished the replay')
- elapsed = time.time() - start_time
- logger.info('elapsed time %.2f sec' % elapsed)
- return elapsed
-
- def run_replay_dump(self, replay_path):
- try:
- subprocess.check_call(["./scripts/replay-dump.py",
- "-f", replay_path],
- stdout=subprocess.DEVNULL)
- except subprocess.CalledProcessError:
- self.fail('replay-dump.py failed')
-
- def run_rr(self, kernel_path, kernel_command_line, console_pattern,
- shift=7, args=None):
- replay_path = os.path.join(self.workdir, 'replay.bin')
- t1 = self.run_vm(kernel_path, kernel_command_line, console_pattern,
- True, shift, args, replay_path)
- t2 = self.run_vm(kernel_path, kernel_command_line, console_pattern,
- False, shift, args, replay_path)
- logger = logging.getLogger('replay')
- logger.info('replay overhead {:.2%}'.format(t2 / t1 - 1))
-
-class ReplayKernelNormal(ReplayKernelBase):
-
- def test_i386_pc(self):
- """
- :avocado: tags=arch:i386
- :avocado: tags=machine:pc
- """
- kernel_url = ('https://storage.tuxboot.com/20230331/i386/bzImage')
- kernel_hash = 'a3e5b32a354729e65910f5a1ffcda7c14a6c12a55e8213fb86e277f1b76ed956'
- kernel_path = self.fetch_asset(kernel_url,
- asset_hash=kernel_hash,
- algorithm = "sha256")
-
- kernel_command_line = self.KERNEL_COMMON_COMMAND_LINE + 'console=ttyS0'
- console_pattern = 'VFS: Cannot open root device'
-
- self.run_rr(kernel_path, kernel_command_line, console_pattern, shift=5)
diff --git a/tests/avocado/replay_linux.py b/tests/avocado/replay_linux.py
deleted file mode 100644
index 5916922..0000000
--- a/tests/avocado/replay_linux.py
+++ /dev/null
@@ -1,206 +0,0 @@
-# Record/replay test that boots a complete Linux system via a cloud image
-#
-# Copyright (c) 2020 ISP RAS
-#
-# Author:
-# Pavel Dovgalyuk <Pavel.Dovgaluk@ispras.ru>
-#
-# This work is licensed under the terms of the GNU GPL, version 2 or
-# later. See the COPYING file in the top-level directory.
-
-import os
-import logging
-import time
-
-from avocado import skipUnless
-from avocado_qemu import BUILD_DIR
-from avocado.utils import cloudinit
-from avocado.utils import network
-from avocado.utils import vmimage
-from avocado.utils import datadrainer
-from avocado.utils.path import find_command
-from avocado_qemu.linuxtest import LinuxTest
-
-class ReplayLinux(LinuxTest):
- """
- Boots a Linux system, checking for a successful initialization
- """
-
- timeout = 1800
- chksum = None
- hdd = 'ide-hd'
- cd = 'ide-cd'
- bus = 'ide'
-
- def setUp(self):
- # LinuxTest does many replay-incompatible things, but includes
- # useful methods. Do not setup LinuxTest here and just
- # call some functions.
- super(LinuxTest, self).setUp()
- self._set_distro()
- self.boot_path = self.download_boot()
- self.phone_server = cloudinit.PhoneHomeServer(('0.0.0.0', 0),
- self.name)
- ssh_pubkey, self.ssh_key = self.set_up_existing_ssh_keys()
- self.cloudinit_path = self.prepare_cloudinit(ssh_pubkey)
-
- def vm_add_disk(self, vm, path, id, device):
- bus_string = ''
- if self.bus:
- bus_string = ',bus=%s.%d' % (self.bus, id,)
- vm.add_args('-drive', 'file=%s,snapshot=on,id=disk%s,if=none' % (path, id))
- vm.add_args('-drive',
- 'driver=blkreplay,id=disk%s-rr,if=none,image=disk%s' % (id, id))
- vm.add_args('-device',
- '%s,drive=disk%s-rr%s' % (device, id, bus_string))
-
- def vm_add_cdrom(self, vm, path, id, device):
- vm.add_args('-drive', 'file=%s,id=disk%s,if=none,media=cdrom' % (path, id))
-
- def launch_and_wait(self, record, args, shift):
- self.require_netdev('user')
- vm = self.get_vm()
- vm.add_args('-smp', '1')
- vm.add_args('-m', '1024')
- vm.add_args('-netdev', 'user,id=vnet,hostfwd=:127.0.0.1:0-:22',
- '-device', 'virtio-net,netdev=vnet')
- vm.add_args('-object', 'filter-replay,id=replay,netdev=vnet')
- if args:
- vm.add_args(*args)
- self.vm_add_disk(vm, self.boot_path, 0, self.hdd)
- self.vm_add_cdrom(vm, self.cloudinit_path, 1, self.cd)
- logger = logging.getLogger('replay')
- if record:
- logger.info('recording the execution...')
- mode = 'record'
- else:
- logger.info('replaying the execution...')
- mode = 'replay'
- replay_path = os.path.join(self.workdir, 'replay.bin')
- vm.add_args('-icount', 'shift=%s,rr=%s,rrfile=%s' %
- (shift, mode, replay_path))
-
- start_time = time.time()
-
- vm.set_console()
- vm.launch()
- console_drainer = datadrainer.LineLogger(vm.console_socket.fileno(),
- logger=self.log.getChild('console'),
- stop_check=(lambda : not vm.is_running()))
- console_drainer.start()
- if record:
- while not self.phone_server.instance_phoned_back:
- self.phone_server.handle_request()
- vm.shutdown()
- logger.info('finished the recording with log size %s bytes'
- % os.path.getsize(replay_path))
- self.run_replay_dump(replay_path)
- logger.info('successfully tested replay-dump.py')
- else:
- vm.event_wait('SHUTDOWN', self.timeout)
- vm.wait()
- logger.info('successfully finished the replay')
- elapsed = time.time() - start_time
- logger.info('elapsed time %.2f sec' % elapsed)
- return elapsed
-
- def run_rr(self, args=None, shift=7):
- t1 = self.launch_and_wait(True, args, shift)
- t2 = self.launch_and_wait(False, args, shift)
- logger = logging.getLogger('replay')
- logger.info('replay overhead {:.2%}'.format(t2 / t1 - 1))
-
- def run_replay_dump(self, replay_path):
- try:
- subprocess.check_call(["./scripts/replay-dump.py",
- "-f", replay_path],
- stdout=subprocess.DEVNULL)
- except subprocess.CalledProcessError:
- self.fail('replay-dump.py failed')
-
-@skipUnless(os.getenv('AVOCADO_TIMEOUT_EXPECTED'), 'Test might timeout')
-class ReplayLinuxX8664(ReplayLinux):
- """
- :avocado: tags=arch:x86_64
- :avocado: tags=accel:tcg
- """
-
- chksum = 'e3c1b309d9203604922d6e255c2c5d098a309c2d46215d8fc026954f3c5c27a0'
-
- def test_pc_i440fx(self):
- """
- :avocado: tags=machine:pc
- """
- self.run_rr(shift=1)
-
- def test_pc_q35(self):
- """
- :avocado: tags=machine:q35
- """
- self.run_rr(shift=3)
-
-@skipUnless(os.getenv('AVOCADO_TIMEOUT_EXPECTED'), 'Test might timeout')
-class ReplayLinuxX8664Virtio(ReplayLinux):
- """
- :avocado: tags=arch:x86_64
- :avocado: tags=virtio
- :avocado: tags=accel:tcg
- """
-
- hdd = 'virtio-blk-pci'
- cd = 'virtio-blk-pci'
- bus = None
-
- chksum = 'e3c1b309d9203604922d6e255c2c5d098a309c2d46215d8fc026954f3c5c27a0'
-
- def test_pc_i440fx(self):
- """
- :avocado: tags=machine:pc
- """
- self.run_rr(shift=1)
-
- def test_pc_q35(self):
- """
- :avocado: tags=machine:q35
- """
- self.run_rr(shift=3)
-
-@skipUnless(os.getenv('AVOCADO_TIMEOUT_EXPECTED'), 'Test might timeout')
-class ReplayLinuxAarch64(ReplayLinux):
- """
- :avocado: tags=accel:tcg
- :avocado: tags=arch:aarch64
- :avocado: tags=machine:virt
- :avocado: tags=cpu:max
- """
-
- chksum = '1e18d9c0cf734940c4b5d5ec592facaed2af0ad0329383d5639c997fdf16fe49'
-
- hdd = 'virtio-blk-device'
- cd = 'virtio-blk-device'
- bus = None
-
- def get_common_args(self):
- return ('-bios',
- os.path.join(BUILD_DIR, 'pc-bios', 'edk2-aarch64-code.fd'),
- "-cpu", "max,lpa2=off",
- '-device', 'virtio-rng-pci,rng=rng0',
- '-object', 'rng-builtin,id=rng0')
-
- def test_virt_gicv2(self):
- """
- :avocado: tags=machine:gic-version=2
- """
-
- self.run_rr(shift=3,
- args=(*self.get_common_args(),
- "-machine", "virt,gic-version=2"))
-
- def test_virt_gicv3(self):
- """
- :avocado: tags=machine:gic-version=3
- """
-
- self.run_rr(shift=3,
- args=(*self.get_common_args(),
- "-machine", "virt,gic-version=3"))
diff --git a/tests/avocado/reverse_debugging.py b/tests/avocado/reverse_debugging.py
deleted file mode 100644
index f24287c..0000000
--- a/tests/avocado/reverse_debugging.py
+++ /dev/null
@@ -1,272 +0,0 @@
-# Reverse debugging test
-#
-# Copyright (c) 2020 ISP RAS
-#
-# Author:
-# Pavel Dovgalyuk <Pavel.Dovgalyuk@ispras.ru>
-#
-# This work is licensed under the terms of the GNU GPL, version 2 or
-# later. See the COPYING file in the top-level directory.
-import os
-import logging
-
-from avocado import skipUnless
-from avocado_qemu import BUILD_DIR
-from avocado.utils import datadrainer
-from avocado.utils import gdb
-from avocado.utils import process
-from avocado.utils.network.ports import find_free_port
-from avocado.utils.path import find_command
-from boot_linux_console import LinuxKernelTest
-
-class ReverseDebugging(LinuxKernelTest):
- """
- Test GDB reverse debugging commands: reverse step and reverse continue.
- Recording saves the execution of some instructions and makes an initial
- VM snapshot to allow reverse execution.
- Replay saves the order of the first instructions and then checks that they
- are executed backwards in the correct order.
- After that the execution is replayed to the end, and reverse continue
- command is checked by setting several breakpoints, and asserting
- that the execution is stopped at the last of them.
- """
-
- timeout = 10
- STEPS = 10
- endian_is_le = True
-
- def run_vm(self, record, shift, args, replay_path, image_path, port):
- logger = logging.getLogger('replay')
- vm = self.get_vm()
- vm.set_console()
- if record:
- logger.info('recording the execution...')
- mode = 'record'
- else:
- logger.info('replaying the execution...')
- mode = 'replay'
- vm.add_args('-gdb', 'tcp::%d' % port, '-S')
- vm.add_args('-icount', 'shift=%s,rr=%s,rrfile=%s,rrsnapshot=init' %
- (shift, mode, replay_path),
- '-net', 'none')
- vm.add_args('-drive', 'file=%s,if=none' % image_path)
- if args:
- vm.add_args(*args)
- vm.launch()
- console_drainer = datadrainer.LineLogger(vm.console_socket.fileno(),
- logger=self.log.getChild('console'),
- stop_check=(lambda : not vm.is_running()))
- console_drainer.start()
- return vm
-
- @staticmethod
- def get_reg_le(g, reg):
- res = g.cmd(b'p%x' % reg)
- num = 0
- for i in range(len(res))[-2::-2]:
- num = 0x100 * num + int(res[i:i + 2], 16)
- return num
-
- @staticmethod
- def get_reg_be(g, reg):
- res = g.cmd(b'p%x' % reg)
- return int(res, 16)
-
- def get_reg(self, g, reg):
- # value may be encoded in BE or LE order
- if self.endian_is_le:
- return self.get_reg_le(g, reg)
- else:
- return self.get_reg_be(g, reg)
-
- def get_pc(self, g):
- return self.get_reg(g, self.REG_PC)
-
- def check_pc(self, g, addr):
- pc = self.get_pc(g)
- if pc != addr:
- self.fail('Invalid PC (read %x instead of %x)' % (pc, addr))
-
- @staticmethod
- def gdb_step(g):
- g.cmd(b's', b'T05thread:01;')
-
- @staticmethod
- def gdb_bstep(g):
- g.cmd(b'bs', b'T05thread:01;')
-
- @staticmethod
- def vm_get_icount(vm):
- return vm.qmp('query-replay')['return']['icount']
-
- def reverse_debugging(self, shift=7, args=None):
- logger = logging.getLogger('replay')
-
- # create qcow2 for snapshots
- logger.info('creating qcow2 image for VM snapshots')
- image_path = os.path.join(self.workdir, 'disk.qcow2')
- qemu_img = os.path.join(BUILD_DIR, 'qemu-img')
- if not os.path.exists(qemu_img):
- qemu_img = find_command('qemu-img', False)
- if qemu_img is False:
- self.cancel('Could not find "qemu-img", which is required to '
- 'create the temporary qcow2 image')
- cmd = '%s create -f qcow2 %s 128M' % (qemu_img, image_path)
- process.run(cmd)
-
- replay_path = os.path.join(self.workdir, 'replay.bin')
- port = find_free_port()
-
- # record the log
- vm = self.run_vm(True, shift, args, replay_path, image_path, port)
- while self.vm_get_icount(vm) <= self.STEPS:
- pass
- last_icount = self.vm_get_icount(vm)
- vm.shutdown()
-
- logger.info("recorded log with %s+ steps" % last_icount)
-
- # replay and run debug commands
- vm = self.run_vm(False, shift, args, replay_path, image_path, port)
- logger.info('connecting to gdbstub')
- g = gdb.GDBRemote('127.0.0.1', port, False, False)
- g.connect()
- r = g.cmd(b'qSupported')
- if b'qXfer:features:read+' in r:
- g.cmd(b'qXfer:features:read:target.xml:0,ffb')
- if b'ReverseStep+' not in r:
- self.fail('Reverse step is not supported by QEMU')
- if b'ReverseContinue+' not in r:
- self.fail('Reverse continue is not supported by QEMU')
-
- logger.info('stepping forward')
- steps = []
- # record first instruction addresses
- for _ in range(self.STEPS):
- pc = self.get_pc(g)
- logger.info('saving position %x' % pc)
- steps.append(pc)
- self.gdb_step(g)
-
- # visit the recorded instruction in reverse order
- logger.info('stepping backward')
- for addr in steps[::-1]:
- self.gdb_bstep(g)
- self.check_pc(g, addr)
- logger.info('found position %x' % addr)
-
- # visit the recorded instruction in forward order
- logger.info('stepping forward')
- for addr in steps:
- self.check_pc(g, addr)
- self.gdb_step(g)
- logger.info('found position %x' % addr)
-
- # set breakpoints for the instructions just stepped over
- logger.info('setting breakpoints')
- for addr in steps:
- # hardware breakpoint at addr with len=1
- g.cmd(b'Z1,%x,1' % addr, b'OK')
-
- # this may hit a breakpoint if first instructions are executed
- # again
- logger.info('continuing execution')
- vm.qmp('replay-break', icount=last_icount - 1)
- # continue - will return after pausing
- # This could stop at the end and get a T02 return, or by
- # re-executing one of the breakpoints and get a T05 return.
- g.cmd(b'c')
- if self.vm_get_icount(vm) == last_icount - 1:
- logger.info('reached the end (icount %s)' % (last_icount - 1))
- else:
- logger.info('hit a breakpoint again at %x (icount %s)' %
- (self.get_pc(g), self.vm_get_icount(vm)))
-
- logger.info('running reverse continue to reach %x' % steps[-1])
- # reverse continue - will return after stopping at the breakpoint
- g.cmd(b'bc', b'T05thread:01;')
-
- # assume that none of the first instructions is executed again
- # breaking the order of the breakpoints
- self.check_pc(g, steps[-1])
- logger.info('successfully reached %x' % steps[-1])
-
- logger.info('exiting gdb and qemu')
- vm.shutdown()
-
-class ReverseDebugging_X86_64(ReverseDebugging):
- """
- :avocado: tags=accel:tcg
- """
-
- REG_PC = 0x10
- REG_CS = 0x12
- def get_pc(self, g):
- return self.get_reg_le(g, self.REG_PC) \
- + self.get_reg_le(g, self.REG_CS) * 0x10
-
- # unidentified gitlab timeout problem
- @skipUnless(os.getenv('QEMU_TEST_FLAKY_TESTS'), 'Test is unstable on GitLab')
- def test_x86_64_pc(self):
- """
- :avocado: tags=arch:x86_64
- :avocado: tags=machine:pc
- """
- # start with BIOS only
- self.reverse_debugging()
-
-class ReverseDebugging_AArch64(ReverseDebugging):
- """
- :avocado: tags=accel:tcg
- """
-
- REG_PC = 32
-
- # unidentified gitlab timeout problem
- @skipUnless(os.getenv('QEMU_TEST_FLAKY_TESTS'), 'Test is unstable on GitLab')
- def test_aarch64_virt(self):
- """
- :avocado: tags=arch:aarch64
- :avocado: tags=machine:virt
- :avocado: tags=cpu:cortex-a53
- """
- kernel_url = ('https://archives.fedoraproject.org/pub/archive/fedora'
- '/linux/releases/29/Everything/aarch64/os/images/pxeboot'
- '/vmlinuz')
- kernel_hash = '8c73e469fc6ea06a58dc83a628fc695b693b8493'
- kernel_path = self.fetch_asset(kernel_url, asset_hash=kernel_hash)
-
- self.reverse_debugging(
- args=('-kernel', kernel_path))
-
-class ReverseDebugging_ppc64(ReverseDebugging):
- """
- :avocado: tags=accel:tcg
- """
-
- REG_PC = 0x40
-
- # unidentified gitlab timeout problem
- @skipUnless(os.getenv('QEMU_TEST_FLAKY_TESTS'), 'Test is unstable on GitLab')
- def test_ppc64_pseries(self):
- """
- :avocado: tags=arch:ppc64
- :avocado: tags=machine:pseries
- :avocado: tags=flaky
- """
- # SLOF branches back to its entry point, which causes this test
- # to take the 'hit a breakpoint again' path. That's not a problem,
- # just slightly different than the other machines.
- self.endian_is_le = False
- self.reverse_debugging()
-
- # See https://gitlab.com/qemu-project/qemu/-/issues/1992
- @skipUnless(os.getenv('QEMU_TEST_FLAKY_TESTS'), 'Test is unstable on GitLab')
- def test_ppc64_powernv(self):
- """
- :avocado: tags=arch:ppc64
- :avocado: tags=machine:powernv
- :avocado: tags=flaky
- """
- self.endian_is_le = False
- self.reverse_debugging()
diff --git a/tests/avocado/smmu.py b/tests/avocado/smmu.py
deleted file mode 100644
index 83fd79e..0000000
--- a/tests/avocado/smmu.py
+++ /dev/null
@@ -1,139 +0,0 @@
-# SMMUv3 Functional tests
-#
-# Copyright (c) 2021 Red Hat, Inc.
-#
-# Author:
-# Eric Auger <eric.auger@redhat.com>
-#
-# This work is licensed under the terms of the GNU GPL, version 2 or
-# later. See the COPYING file in the top-level directory.
-import os
-
-from avocado import skipUnless
-from avocado_qemu import BUILD_DIR
-from avocado_qemu.linuxtest import LinuxTest
-
-@skipUnless(os.getenv('QEMU_TEST_FLAKY_TESTS'), 'Test is unstable on GitLab')
-class SMMU(LinuxTest):
- """
- :avocado: tags=accel:kvm
- :avocado: tags=cpu:host
- :avocado: tags=arch:aarch64
- :avocado: tags=machine:virt
- :avocado: tags=distro:fedora
- :avocado: tags=smmu
- :avocado: tags=flaky
- """
-
- IOMMU_ADDON = ',iommu_platform=on,disable-modern=off,disable-legacy=on'
- kernel_path = None
- initrd_path = None
- kernel_params = None
-
- def set_up_boot(self):
- path = self.download_boot()
- self.vm.add_args('-device', 'virtio-blk-pci,bus=pcie.0,' +
- 'drive=drv0,id=virtio-disk0,bootindex=1,'
- 'werror=stop,rerror=stop' + self.IOMMU_ADDON)
- self.vm.add_args('-drive',
- 'file=%s,if=none,cache=writethrough,id=drv0' % path)
-
- def setUp(self):
- super(SMMU, self).setUp(None, 'virtio-net-pci' + self.IOMMU_ADDON)
-
- def common_vm_setup(self, custom_kernel=False):
- self.require_accelerator("kvm")
- self.vm.add_args("-accel", "kvm")
- self.vm.add_args("-cpu", "host")
- self.vm.add_args("-machine", "iommu=smmuv3")
- self.vm.add_args("-d", "guest_errors")
- self.vm.add_args('-bios', os.path.join(BUILD_DIR, 'pc-bios',
- 'edk2-aarch64-code.fd'))
- self.vm.add_args('-device', 'virtio-rng-pci,rng=rng0')
- self.vm.add_args('-object',
- 'rng-random,id=rng0,filename=/dev/urandom')
-
- if custom_kernel is False:
- return
-
- kernel_url = self.distro.pxeboot_url + 'vmlinuz'
- initrd_url = self.distro.pxeboot_url + 'initrd.img'
- self.kernel_path = self.fetch_asset(kernel_url)
- self.initrd_path = self.fetch_asset(initrd_url)
-
- def run_and_check(self):
- if self.kernel_path:
- self.vm.add_args('-kernel', self.kernel_path,
- '-append', self.kernel_params,
- '-initrd', self.initrd_path)
- self.launch_and_wait()
- self.ssh_command('cat /proc/cmdline')
- self.ssh_command('dnf -y install numactl-devel')
-
-
- # 5.3 kernel without RIL #
-
- def test_smmu_noril(self):
- """
- :avocado: tags=smmu_noril
- :avocado: tags=smmu_noril_tests
- :avocado: tags=distro_version:31
- """
- self.common_vm_setup()
- self.run_and_check()
-
- def test_smmu_noril_passthrough(self):
- """
- :avocado: tags=smmu_noril_passthrough
- :avocado: tags=smmu_noril_tests
- :avocado: tags=distro_version:31
- """
- self.common_vm_setup(True)
- self.kernel_params = (self.distro.default_kernel_params +
- ' iommu.passthrough=on')
- self.run_and_check()
-
- def test_smmu_noril_nostrict(self):
- """
- :avocado: tags=smmu_noril_nostrict
- :avocado: tags=smmu_noril_tests
- :avocado: tags=distro_version:31
- """
- self.common_vm_setup(True)
- self.kernel_params = (self.distro.default_kernel_params +
- ' iommu.strict=0')
- self.run_and_check()
-
- # 5.8 kernel featuring range invalidation
- # >= v5.7 kernel
-
- def test_smmu_ril(self):
- """
- :avocado: tags=smmu_ril
- :avocado: tags=smmu_ril_tests
- :avocado: tags=distro_version:33
- """
- self.common_vm_setup()
- self.run_and_check()
-
- def test_smmu_ril_passthrough(self):
- """
- :avocado: tags=smmu_ril_passthrough
- :avocado: tags=smmu_ril_tests
- :avocado: tags=distro_version:33
- """
- self.common_vm_setup(True)
- self.kernel_params = (self.distro.default_kernel_params +
- ' iommu.passthrough=on')
- self.run_and_check()
-
- def test_smmu_ril_nostrict(self):
- """
- :avocado: tags=smmu_ril_nostrict
- :avocado: tags=smmu_ril_tests
- :avocado: tags=distro_version:33
- """
- self.common_vm_setup(True)
- self.kernel_params = (self.distro.default_kernel_params +
- ' iommu.strict=0')
- self.run_and_check()
generated by cgit v1.1 at 2025-04-24 19:48:23 +0000