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 pycotap
17import sys
18import unittest
19import uuid
20
21from qemu.machine import QEMUMachine
22from qemu.utils import kvm_available, tcg_available
23
24from .cmd import run_cmd
25from .config import BUILD_DIR
26
27
28class QemuBaseTest(unittest.TestCase):
29
30    qemu_bin = os.getenv('QEMU_TEST_QEMU_BINARY')
31    arch = None
32
33    workdir = None
34    log = logging.getLogger('qemu-test')
35
36    def setUp(self, bin_prefix):
37        self.assertIsNotNone(self.qemu_bin, 'QEMU_TEST_QEMU_BINARY must be set')
38        self.arch = self.qemu_bin.split('-')[-1]
39
40        self.workdir = os.path.join(BUILD_DIR, 'tests/functional', self.arch,
41                                    self.id())
42        os.makedirs(self.workdir, exist_ok=True)
43
44    def main():
45        path = os.path.basename(sys.argv[0])[:-3]
46        tr = pycotap.TAPTestRunner(message_log = pycotap.LogMode.LogToError,
47                                   test_output_log = pycotap.LogMode.LogToError)
48        unittest.main(module = None, testRunner = tr, argv=["__dummy__", path])
49
50
51class QemuSystemTest(QemuBaseTest):
52    """Facilitates system emulation tests."""
53
54    cpu = None
55    machine = None
56    _machinehelp = None
57
58    def setUp(self):
59        self._vms = {}
60
61        super().setUp('qemu-system-')
62
63    def set_machine(self, machinename):
64        # TODO: We should use QMP to get the list of available machines
65        if not self._machinehelp:
66            self._machinehelp = run_cmd([self.qemu_bin, '-M', 'help'])[0];
67        if self._machinehelp.find(machinename) < 0:
68            self.skipTest('no support for machine ' + machinename)
69        self.machine = machinename
70
71    def require_accelerator(self, accelerator):
72        """
73        Requires an accelerator to be available for the test to continue
74
75        It takes into account the currently set qemu binary.
76
77        If the check fails, the test is canceled.  If the check itself
78        for the given accelerator is not available, the test is also
79        canceled.
80
81        :param accelerator: name of the accelerator, such as "kvm" or "tcg"
82        :type accelerator: str
83        """
84        checker = {'tcg': tcg_available,
85                   'kvm': kvm_available}.get(accelerator)
86        if checker is None:
87            self.skipTest("Don't know how to check for the presence "
88                          "of accelerator %s" % accelerator)
89        if not checker(qemu_bin=self.qemu_bin):
90            self.skipTest("%s accelerator does not seem to be "
91                          "available" % accelerator)
92
93    def require_netdev(self, netdevname):
94        netdevhelp = run_cmd([self.qemu_bin,
95                             '-M', 'none', '-netdev', 'help'])[0];
96        if netdevhelp.find('\n' + netdevname + '\n') < 0:
97            self.skipTest('no support for " + netdevname + " networking')
98
99    def require_device(self, devicename):
100        devhelp = run_cmd([self.qemu_bin,
101                           '-M', 'none', '-device', 'help'])[0];
102        if devhelp.find(devicename) < 0:
103            self.skipTest('no support for device ' + devicename)
104
105    def _new_vm(self, name, *args):
106        vm = QEMUMachine(self.qemu_bin, base_temp_dir=self.workdir)
107        self.log.debug('QEMUMachine "%s" created', name)
108        self.log.debug('QEMUMachine "%s" temp_dir: %s', name, vm.temp_dir)
109        self.log.debug('QEMUMachine "%s" log_dir: %s', name, vm.log_dir)
110        if args:
111            vm.add_args(*args)
112        return vm
113
114    @property
115    def vm(self):
116        return self.get_vm(name='default')
117
118    def get_vm(self, *args, name=None):
119        if not name:
120            name = str(uuid.uuid4())
121        if self._vms.get(name) is None:
122            self._vms[name] = self._new_vm(name, *args)
123            if self.cpu is not None:
124                self._vms[name].add_args('-cpu', self.cpu)
125            if self.machine is not None:
126                self._vms[name].set_machine(self.machine)
127        return self._vms[name]
128
129    def set_vm_arg(self, arg, value):
130        """
131        Set an argument to list of extra arguments to be given to the QEMU
132        binary. If the argument already exists then its value is replaced.
133
134        :param arg: the QEMU argument, such as "-cpu" in "-cpu host"
135        :type arg: str
136        :param value: the argument value, such as "host" in "-cpu host"
137        :type value: str
138        """
139        if not arg or not value:
140            return
141        if arg not in self.vm.args:
142            self.vm.args.extend([arg, value])
143        else:
144            idx = self.vm.args.index(arg) + 1
145            if idx < len(self.vm.args):
146                self.vm.args[idx] = value
147            else:
148                self.vm.args.append(value)
149
150    def tearDown(self):
151        for vm in self._vms.values():
152            vm.shutdown()
153        super().tearDown()
154