| 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 | |
| 14 | import logging |
| 15 | import os |
| 16 | from pathlib import Path |
| 17 | import shutil |
| 18 | from subprocess import run |
| 19 | import sys |
| 20 | import tempfile |
| 21 | import warnings |
| 22 | import unittest |
| 23 | import uuid |
| 24 | |
| 25 | import pycotap |
| 26 | |
| 27 | from qemu.machine import QEMUMachine |
| 28 | from qemu.utils import hvf_available, kvm_available, tcg_available |
| 29 | |
| 30 | from .archive import archive_extract |
| 31 | from .asset import Asset |
| 32 | from .config import BUILD_DIR, dso_suffix |
| 33 | from .uncompress import uncompress |
| 34 | |
| 35 | |
| 36 | class QemuBaseTest(unittest.TestCase): |
| 37 | |
| 38 | def uncompress(self, compressed, target=None, format=None): |
| 39 | ''' |
| 40 | @params compressed: filename, Asset, or file-like object to uncompress |
| 41 | @params format: optional compression format (gzip, lzma) |
| 42 | |
| 43 | Uncompresses @compressed into the scratch directory. |
| 44 | |
| 45 | If @format is None, heuristics will be applied to guess the |
| 46 | format from the filename or Asset URL. @format must be non-None |
| 47 | if @uncompressed is a file-like object. |
| 48 | |
| 49 | Returns the fully qualified path to the uncompressed file |
| 50 | ''' |
| 51 | self.log.debug(f"Uncompress {compressed} format={format}") |
| 52 | if isinstance(compressed, Asset): |
| 53 | compressed.fetch() |
| 54 | |
| 55 | if target is not None: |
| 56 | uncompressed = self.scratch_file(target) |
| 57 | else: |
| 58 | (name, _ext) = os.path.splitext(str(compressed)) |
| 59 | uncompressed = self.scratch_file(os.path.basename(name)) |
| 60 | |
| 61 | uncompress(compressed, uncompressed, format) |
| 62 | |
| 63 | return uncompressed |
| 64 | |
| 65 | def archive_extract(self, archive, format=None, sub_dir=None, member=None): |
| 66 | ''' |
| 67 | @params archive: filename, Asset, or file-like object to extract |
| 68 | @params format: optional archive format (tar, zip, deb, cpio) |
| 69 | @params sub_dir: optional sub-directory to extract into |
| 70 | @params member: optional member file to limit extraction to |
| 71 | |
| 72 | Extracts @archive into the scratch directory, or a directory beneath |
| 73 | named by @sub_dir. All files are extracted unless @member specifies |
| 74 | a limit. |
| 75 | |
| 76 | If @format is None, heuristics will be applied to guess the |
| 77 | format from the filename or Asset URL. @format must be non-None |
| 78 | if @archive is a file-like object. |
| 79 | |
| 80 | If @member is non-None, returns the fully qualified path to @member |
| 81 | ''' |
| 82 | self.log.debug(f"Extract {archive} format={format}" + |
| 83 | f"sub_dir={sub_dir} member={member}") |
| 84 | if isinstance(archive, Asset): |
| 85 | archive.fetch() |
| 86 | if sub_dir is None: |
| 87 | archive_extract(archive, self.scratch_file(), format, member) |
| 88 | else: |
| 89 | archive_extract(archive, self.scratch_file(sub_dir), |
| 90 | format, member) |
| 91 | |
| 92 | if member is not None: |
| 93 | return self.scratch_file(member) |
| 94 | return None |
| 95 | |
| 96 | def socket_dir(self): |
| 97 | ''' |
| 98 | Create a temporary directory suitable for storing UNIX |
| 99 | socket paths. |
| 100 | |
| 101 | Returns: a tempfile.TemporaryDirectory instance |
| 102 | ''' |
| 103 | if self.socketdir is None: |
| 104 | self.socketdir = tempfile.TemporaryDirectory( |
| 105 | prefix="qemu_func_test_sock_") |
| 106 | return self.socketdir |
| 107 | |
| 108 | def data_file(self, *args): |
| 109 | ''' |
| 110 | @params args list of zero or more subdirectories or file |
| 111 | |
| 112 | Construct a path for accessing a data file located |
| 113 | relative to the source directory that is the root for |
| 114 | functional tests. |
| 115 | |
| 116 | @args may be an empty list to reference the root dir |
| 117 | itself, may be a single element to reference a file in |
| 118 | the root directory, or may be multiple elements to |
| 119 | reference a file nested below. The path components |
| 120 | will be joined using the platform appropriate path |
| 121 | separator. |
| 122 | |
| 123 | Returns: string representing a file path |
| 124 | ''' |
| 125 | return str(Path(Path(__file__).parent.parent, *args)) |
| 126 | |
| 127 | def build_file(self, *args): |
| 128 | ''' |
| 129 | @params args list of zero or more subdirectories or file |
| 130 | |
| 131 | Construct a path for accessing a data file located |
| 132 | relative to the build directory root. |
| 133 | |
| 134 | @args may be an empty list to reference the build dir |
| 135 | itself, may be a single element to reference a file in |
| 136 | the build directory, or may be multiple elements to |
| 137 | reference a file nested below. The path components |
| 138 | will be joined using the platform appropriate path |
| 139 | separator. |
| 140 | |
| 141 | Returns: string representing a file path |
| 142 | ''' |
| 143 | return str(Path(BUILD_DIR, *args)) |
| 144 | |
| 145 | def scratch_file(self, *args): |
| 146 | ''' |
| 147 | @params args list of zero or more subdirectories or file |
| 148 | |
| 149 | Construct a path for accessing/creating a scratch file |
| 150 | located relative to a temporary directory dedicated to |
| 151 | this test case. The directory and its contents will be |
| 152 | purged upon completion of the test. |
| 153 | |
| 154 | @args may be an empty list to reference the scratch dir |
| 155 | itself, may be a single element to reference a file in |
| 156 | the scratch directory, or may be multiple elements to |
| 157 | reference a file nested below. The path components |
| 158 | will be joined using the platform appropriate path |
| 159 | separator. |
| 160 | |
| 161 | Returns: string representing a file path |
| 162 | ''' |
| 163 | return str(Path(self.workdir, *args)) |
| 164 | |
| 165 | def log_file(self, *args): |
| 166 | ''' |
| 167 | @params args list of zero or more subdirectories or file |
| 168 | |
| 169 | Construct a path for accessing/creating a log file |
| 170 | located relative to a temporary directory dedicated to |
| 171 | this test case. The directory and its log files will be |
| 172 | preserved upon completion of the test. |
| 173 | |
| 174 | @args may be an empty list to reference the log dir |
| 175 | itself, may be a single element to reference a file in |
| 176 | the log directory, or may be multiple elements to |
| 177 | reference a file nested below. The path components |
| 178 | will be joined using the platform appropriate path |
| 179 | separator. |
| 180 | |
| 181 | Returns: string representing a file path |
| 182 | ''' |
| 183 | return str(Path(self.outputdir, *args)) |
| 184 | |
| 185 | def plugin_file(self, plugin_name): |
| 186 | ''' |
| 187 | @params plugin name |
| 188 | |
| 189 | Return the full path to the plugin taking into account any host OS |
| 190 | specific suffixes. |
| 191 | ''' |
| 192 | sfx = dso_suffix() |
| 193 | return os.path.join('tests', 'tcg', 'plugins', f'{plugin_name}.{sfx}') |
| 194 | |
| 195 | def assets_available(self): |
| 196 | for name, asset in vars(self.__class__).items(): |
| 197 | if name.startswith("ASSET_") and isinstance(asset, Asset): |
| 198 | if not asset.available(): |
| 199 | self.log.debug(f"Asset {asset.url} not available") |
| 200 | return False |
| 201 | return True |
| 202 | |
| 203 | def setUp(self): |
| 204 | self.qemu_bin = os.getenv('QEMU_TEST_QEMU_BINARY') |
| 205 | self.assertIsNotNone(self.qemu_bin, 'QEMU_TEST_QEMU_BINARY must be set') |
| 206 | self.arch = self.qemu_bin.split('-')[-1] |
| 207 | self.socketdir = None |
| 208 | |
| 209 | self.outputdir = self.build_file('tests', 'functional', |
| 210 | self.arch, self.id()) |
| 211 | self.workdir = os.path.join(self.outputdir, 'scratch') |
| 212 | if os.path.exists(self.workdir): |
| 213 | # Purge as safety net in case of unclean termination of |
| 214 | # previous test, or use of QEMU_TEST_KEEP_SCRATCH |
| 215 | shutil.rmtree(self.workdir) |
| 216 | os.makedirs(self.workdir, exist_ok=True) |
| 217 | |
| 218 | self.log_filename = self.log_file('base.log') |
| 219 | self.log = logging.getLogger('qemu-test') |
| 220 | self.log.setLevel(logging.DEBUG) |
| 221 | self._log_fh = logging.FileHandler(self.log_filename, mode='w') |
| 222 | self._log_fh.setLevel(logging.DEBUG) |
| 223 | file_formatter = logging.Formatter( |
| 224 | '%(asctime)s - %(levelname)s: %(name)s.%(funcName)s %(message)s') |
| 225 | self._log_fh.setFormatter(file_formatter) |
| 226 | self.log.addHandler(self._log_fh) |
| 227 | |
| 228 | # Capture QEMUMachine logging |
| 229 | self.machinelog = logging.getLogger('qemu.machine') |
| 230 | self.machinelog.setLevel(logging.DEBUG) |
| 231 | self.machinelog.addHandler(self._log_fh) |
| 232 | self.qmplog = logging.getLogger('qemu.qmp') |
| 233 | self.qmplog.setLevel(logging.DEBUG) |
| 234 | self.qmplog.addHandler(self._log_fh) |
| 235 | |
| 236 | if not self.assets_available(): |
| 237 | self.skipTest('One or more assets is not available') |
| 238 | |
| 239 | def tearDown(self): |
| 240 | if "QEMU_TEST_KEEP_SCRATCH" not in os.environ: |
| 241 | shutil.rmtree(self.workdir) |
| 242 | if self.socketdir is not None: |
| 243 | self.socketdir.cleanup() |
| 244 | self.socketdir = None |
| 245 | self.qmplog.removeHandler(self._log_fh) |
| 246 | self.machinelog.removeHandler(self._log_fh) |
| 247 | self.log.removeHandler(self._log_fh) |
| 248 | self._log_fh.close() |
| 249 | |
| 250 | @staticmethod |
| 251 | def main(): |
| 252 | warnings.simplefilter("default") |
| 253 | os.environ["PYTHONWARNINGS"] = "default" |
| 254 | |
| 255 | test_module = os.path.basename(sys.argv[0])[:-3] |
| 256 | |
| 257 | cache = os.environ.get("QEMU_TEST_PRECACHE", None) |
| 258 | if cache is not None: |
| 259 | Asset.precache_suites(test_module, cache) |
| 260 | return |
| 261 | |
| 262 | tr = pycotap.TAPTestRunner(message_log = pycotap.LogMode.LogToError, |
| 263 | test_output_log = pycotap.LogMode.LogToError) |
| 264 | res = unittest.main(test_module, testRunner = tr, exit = False) |
| 265 | failed = {} |
| 266 | for (test, _message) in res.result.errors + res.result.failures: |
| 267 | if hasattr(test, "log_filename") and not test.id() in failed: |
| 268 | print('More information on ' + test.id() + ' could be found here:' |
| 269 | '\n %s' % test.log_filename, file=sys.stderr) |
| 270 | if hasattr(test, 'console_log_name'): |
| 271 | print(' %s' % test.console_log_name, file=sys.stderr) |
| 272 | failed[test.id()] = True |
| 273 | sys.exit(not res.result.wasSuccessful()) |
| 274 | |
| 275 | |
| 276 | class QemuUserTest(QemuBaseTest): |
| 277 | |
| 278 | def setUp(self): |
| 279 | super().setUp() |
| 280 | self._ldpath = [] |
| 281 | |
| 282 | def add_ldpath(self, ldpath): |
| 283 | self._ldpath.append(os.path.abspath(ldpath)) |
| 284 | |
| 285 | def run_cmd(self, bin_path, args=None): |
| 286 | if args is None: |
| 287 | args = [] |
| 288 | return run([self.qemu_bin] |
| 289 | + ["-L %s" % ldpath for ldpath in self._ldpath] |
| 290 | + [bin_path] |
| 291 | + args, |
| 292 | text=True, capture_output=True) |
| 293 | |
| 294 | class QemuSystemTest(QemuBaseTest): |
| 295 | """Facilitates system emulation tests.""" |
| 296 | |
| 297 | cpu = None |
| 298 | machine = None |
| 299 | _machinehelp = None |
| 300 | |
| 301 | def setUp(self): |
| 302 | self._vms = {} |
| 303 | |
| 304 | super().setUp() |
| 305 | |
| 306 | console_log = logging.getLogger('console') |
| 307 | console_log.setLevel(logging.DEBUG) |
| 308 | self.console_log_name = self.log_file('console.log') |
| 309 | self._console_log_fh = logging.FileHandler(self.console_log_name, |
| 310 | mode='w') |
| 311 | self._console_log_fh.setLevel(logging.DEBUG) |
| 312 | file_formatter = logging.Formatter('%(asctime)s: %(message)s') |
| 313 | self._console_log_fh.setFormatter(file_formatter) |
| 314 | console_log.addHandler(self._console_log_fh) |
| 315 | |
| 316 | def set_machine(self, machinename): |
| 317 | cls = type(self) |
| 318 | |
| 319 | if not hasattr(cls, "_machines"): |
| 320 | tmp_vm = QEMUMachine(self.qemu_bin) |
| 321 | tmp_vm.set_machine('none') |
| 322 | |
| 323 | try: |
| 324 | tmp_vm.launch() |
| 325 | resp = tmp_vm.qmp('query-machines') |
| 326 | |
| 327 | machines = resp.get('return', []) |
| 328 | cls._machines = [] |
| 329 | for m in machines: |
| 330 | if 'name' in m: |
| 331 | cls._machines.append(m['name']) |
| 332 | if 'alias' in m: |
| 333 | cls._machines.append(m['alias']) |
| 334 | |
| 335 | finally: |
| 336 | try: |
| 337 | tmp_vm.shutdown() |
| 338 | except Exception: |
| 339 | pass |
| 340 | |
| 341 | self._machines = cls._machines |
| 342 | |
| 343 | if machinename not in self._machines: |
| 344 | self.skipTest('no support for machine ' + machinename) |
| 345 | |
| 346 | self.machine = machinename |
| 347 | |
| 348 | def require_accelerator(self, accelerator): |
| 349 | """ |
| 350 | Requires an accelerator to be available for the test to continue |
| 351 | |
| 352 | It takes into account the currently set qemu binary. |
| 353 | |
| 354 | If the check fails, the test is canceled. If the check itself |
| 355 | for the given accelerator is not available, the test is also |
| 356 | canceled. |
| 357 | |
| 358 | :param accelerator: name of the accelerator, such as "kvm" or "tcg" |
| 359 | :type accelerator: str |
| 360 | """ |
| 361 | checker = {'tcg': tcg_available, |
| 362 | 'kvm': kvm_available, |
| 363 | 'hvf': hvf_available, |
| 364 | }.get(accelerator) |
| 365 | if checker is None: |
| 366 | self.skipTest("Don't know how to check for the presence " |
| 367 | "of accelerator %s" % accelerator) |
| 368 | if not checker(qemu_bin=self.qemu_bin): |
| 369 | self.skipTest("%s accelerator does not seem to be " |
| 370 | "available" % accelerator) |
| 371 | |
| 372 | def require_netdev(self, netdevname): |
| 373 | helptxt = run([self.qemu_bin, '-M', 'none', '-netdev', 'help'], |
| 374 | capture_output=True, check=True, encoding='utf8').stdout |
| 375 | if helptxt.find('\n' + netdevname + '\n') < 0: |
| 376 | self.skipTest('no support for ' + netdevname + ' networking') |
| 377 | |
| 378 | def require_device(self, devicename): |
| 379 | helptxt = run([self.qemu_bin, '-M', 'none', '-device', 'help'], |
| 380 | capture_output=True, check=True, encoding='utf8').stdout |
| 381 | if helptxt.find(devicename) < 0: |
| 382 | self.skipTest('no support for device ' + devicename) |
| 383 | |
| 384 | def _new_vm(self, name, monitor_address): |
| 385 | vm = QEMUMachine(self.qemu_bin, |
| 386 | name=name, |
| 387 | base_temp_dir=self.workdir, |
| 388 | log_dir=self.log_file(), |
| 389 | monitor_address=monitor_address) |
| 390 | self.log.debug('QEMUMachine "%s" created', name) |
| 391 | self.log.debug('QEMUMachine "%s" temp_dir: %s', name, vm.temp_dir) |
| 392 | |
| 393 | sockpath = os.environ.get("QEMU_TEST_QMP_BACKDOOR", None) |
| 394 | if sockpath is not None: |
| 395 | vm.add_args("-chardev", |
| 396 | f"socket,id=backdoor,path={sockpath},server=on,wait=off", |
| 397 | "-mon", "chardev=backdoor,mode=control") |
| 398 | return vm |
| 399 | |
| 400 | @property |
| 401 | def vm(self): |
| 402 | return self.get_vm(name='default') |
| 403 | |
| 404 | def get_vm(self, name=None, monitor_address=None): |
| 405 | if not name: |
| 406 | name = str(uuid.uuid4()) |
| 407 | if self._vms.get(name) is None: |
| 408 | self._vms[name] = self._new_vm(name, monitor_address) |
| 409 | if self.cpu is not None: |
| 410 | self._vms[name].add_args('-cpu', self.cpu) |
| 411 | if self.machine is not None: |
| 412 | self._vms[name].set_machine(self.machine) |
| 413 | return self._vms[name] |
| 414 | |
| 415 | def set_vm_arg(self, arg, value): |
| 416 | """ |
| 417 | Set an argument to list of extra arguments to be given to the QEMU |
| 418 | binary. If the argument already exists then its value is replaced. |
| 419 | |
| 420 | :param arg: the QEMU argument, such as "-cpu" in "-cpu host" |
| 421 | :type arg: str |
| 422 | :param value: the argument value, such as "host" in "-cpu host" |
| 423 | :type value: str |
| 424 | """ |
| 425 | if not arg or not value: |
| 426 | return |
| 427 | if arg not in self.vm.args: |
| 428 | self.vm.args.extend([arg, value]) |
| 429 | else: |
| 430 | idx = self.vm.args.index(arg) + 1 |
| 431 | if idx < len(self.vm.args): |
| 432 | self.vm.args[idx] = value |
| 433 | else: |
| 434 | self.vm.args.append(value) |
| 435 | |
| 436 | def tearDown(self): |
| 437 | for vm in self._vms.values(): |
| 438 | try: |
| 439 | vm.shutdown() |
| 440 | except Exception as ex: |
| 441 | self.log.error("Failed to teardown VM: %s", ex) |
| 442 | logging.getLogger('console').removeHandler(self._console_log_fh) |
| 443 | self._console_log_fh.close() |
| 444 | super().tearDown() |