tests/qemu-iotests/testenv.py

   1 # TestEnv class to manage test environment variables.
   2 #
   3 # Copyright (c) 2020-2021 Virtuozzo International GmbH
   4 #
   5 # This program is free software; you can redistribute it and/or modify
   6 # it under the terms of the GNU General Public License as published by
   7 # the Free Software Foundation; either version 2 of the License, or
   8 # (at your option) any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU General Public License
  16 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 #
  18
  19 import os
  20 import sys
  21 import tempfile
  22 from pathlib import Path
  23 import shutil
  24 import collections
  25 import random
  26 import subprocess
  27 import glob
  28 from typing import List, Dict, Any, Optional, ContextManager
  29
  30
  31 def isxfile(path: str) -> bool:
  32     return os.path.isfile(path) and os.access(path, os.X_OK)
  33
  34
  35 def get_default_machine(qemu_prog: str) -> str:
  36     outp = subprocess.run([qemu_prog, '-machine', 'help'], check=True,
  37                           universal_newlines=True,
  38                           stdout=subprocess.PIPE).stdout
  39
  40     machines = outp.split('\n')
  41     try:
  42         default_machine = next(m for m in machines if m.endswith(' (default)'))
  43     except StopIteration:
  44         return ''
  45     default_machine = default_machine.split(' ', 1)[0]
  46
  47     alias_suf = ' (alias of {})'.format(default_machine)
  48     alias = next((m for m in machines if m.endswith(alias_suf)), None)
  49     if alias is not None:
  50         default_machine = alias.split(' ', 1)[0]
  51
  52     return default_machine
  53
  54
  55 class TestEnv(ContextManager['TestEnv']):
  56     """
  57     Manage system environment for running tests
  58
  59     The following variables are supported/provided. They are represented by
  60     lower-cased TestEnv attributes.
  61     """
  62
  63     # We store environment variables as instance attributes, and there are a
  64     # lot of them. Silence pylint:
  65     # pylint: disable=too-many-instance-attributes
  66
  67     env_variables = ['PYTHONPATH', 'TEST_DIR', 'SOCK_DIR', 'SAMPLE_IMG_DIR',
  68                      'OUTPUT_DIR', 'PYTHON', 'QEMU_PROG', 'QEMU_IMG_PROG',
  69                      'QEMU_IO_PROG', 'QEMU_NBD_PROG', 'QSD_PROG',
  70                      'SOCKET_SCM_HELPER', 'QEMU_OPTIONS', 'QEMU_IMG_OPTIONS',
  71                      'QEMU_IO_OPTIONS', 'QEMU_IO_OPTIONS_NO_FMT',
  72                      'QEMU_NBD_OPTIONS', 'IMGOPTS', 'IMGFMT', 'IMGPROTO',
  73                      'AIOMODE', 'CACHEMODE', 'VALGRIND_QEMU',
  74                      'CACHEMODE_IS_DEFAULT', 'IMGFMT_GENERIC', 'IMGOPTSSYNTAX',
  75                      'IMGKEYSECRET', 'QEMU_DEFAULT_MACHINE', 'MALLOC_PERTURB_']
  76
  77     def prepare_subprocess(self, args: List[str]) -> Dict[str, str]:
  78         if self.debug:
  79             args.append('-d')
  80
  81         with open(args[0], encoding="utf-8") as f:
  82             try:
  83                 if f.readline().rstrip() == '#!/usr/bin/env python3':
  84                     args.insert(0, self.python)
  85             except UnicodeDecodeError:  # binary test? for future.
  86                 pass
  87
  88         os_env = os.environ.copy()
  89         os_env.update(self.get_env())
  90         return os_env
  91
  92     def get_env(self) -> Dict[str, str]:
  93         env = {}
  94         for v in self.env_variables:
  95             val = getattr(self, v.lower(), None)
  96             if val is not None:
  97                 env[v] = val
  98
  99         return env
 100
 101     def init_directories(self) -> None:
 102         """Init directory variables:
 103              PYTHONPATH
 104              TEST_DIR
 105              SOCK_DIR
 106              SAMPLE_IMG_DIR
 107              OUTPUT_DIR
 108         """
 109         self.pythonpath = os.getenv('PYTHONPATH')
 110         if self.pythonpath:
 111             self.pythonpath = self.source_iotests + os.pathsep + \
 112                 self.pythonpath
 113         else:
 114             self.pythonpath = self.source_iotests
 115
 116         self.test_dir = os.getenv('TEST_DIR',
 117                                   os.path.join(os.getcwd(), 'scratch'))
 118         Path(self.test_dir).mkdir(parents=True, exist_ok=True)
 119
 120         try:
 121             self.sock_dir = os.environ['SOCK_DIR']
 122             self.tmp_sock_dir = False
 123             Path(self.sock_dir).mkdir(parents=True, exist_ok=True)
 124         except KeyError:
 125             self.sock_dir = tempfile.mkdtemp()
 126             self.tmp_sock_dir = True
 127
 128         self.sample_img_dir = os.getenv('SAMPLE_IMG_DIR',
 129                                         os.path.join(self.source_iotests,
 130                                                      'sample_images'))
 131
 132         self.output_dir = os.getcwd()  # OUTPUT_DIR
 133
 134     def init_binaries(self) -> None:
 135         """Init binary path variables:
 136              PYTHON (for bash tests)
 137              QEMU_PROG, QEMU_IMG_PROG, QEMU_IO_PROG, QEMU_NBD_PROG, QSD_PROG
 138              SOCKET_SCM_HELPER
 139         """
 140         self.python = sys.executable
 141
 142         def root(*names: str) -> str:
 143             return os.path.join(self.build_root, *names)
 144
 145         arch = os.uname().machine
 146         if 'ppc64' in arch:
 147             arch = 'ppc64'
 148
 149         self.qemu_prog = os.getenv('QEMU_PROG', root(f'qemu-system-{arch}'))
 150         if not os.path.exists(self.qemu_prog):
 151             pattern = root('qemu-system-*')
 152             try:
 153                 progs = sorted(glob.iglob(pattern))
 154                 self.qemu_prog = next(p for p in progs if isxfile(p))
 155             except StopIteration:
 156                 sys.exit("Not found any Qemu executable binary by pattern "
 157                          f"'{pattern}'")
 158
 159         self.qemu_img_prog = os.getenv('QEMU_IMG_PROG', root('qemu-img'))
 160         self.qemu_io_prog = os.getenv('QEMU_IO_PROG', root('qemu-io'))
 161         self.qemu_nbd_prog = os.getenv('QEMU_NBD_PROG', root('qemu-nbd'))
 162         self.qsd_prog = os.getenv('QSD_PROG', root('storage-daemon',
 163                                                    'qemu-storage-daemon'))
 164
 165         for b in [self.qemu_img_prog, self.qemu_io_prog, self.qemu_nbd_prog,
 166                   self.qemu_prog, self.qsd_prog]:
 167             if not os.path.exists(b):
 168                 sys.exit('No such file: ' + b)
 169             if not isxfile(b):
 170                 sys.exit('Not executable: ' + b)
 171
 172         helper_path = os.path.join(self.build_iotests, 'socket_scm_helper')
 173         if isxfile(helper_path):
 174             self.socket_scm_helper = helper_path  # SOCKET_SCM_HELPER
 175
 176     def __init__(self, imgfmt: str, imgproto: str, aiomode: str,
 177                  cachemode: Optional[str] = None,
 178                  imgopts: Optional[str] = None,
 179                  misalign: bool = False,
 180                  debug: bool = False,
 181                  valgrind: bool = False) -> None:
 182         self.imgfmt = imgfmt
 183         self.imgproto = imgproto
 184         self.aiomode = aiomode
 185         self.imgopts = imgopts
 186         self.misalign = misalign
 187         self.debug = debug
 188
 189         if valgrind:
 190             self.valgrind_qemu = 'y'
 191
 192         if cachemode is None:
 193             self.cachemode_is_default = 'true'
 194             self.cachemode = 'writeback'
 195         else:
 196             self.cachemode_is_default = 'false'
 197             self.cachemode = cachemode
 198
 199         # Initialize generic paths: build_root, build_iotests, source_iotests,
 200         # which are needed to initialize some environment variables. They are
 201         # used by init_*() functions as well.
 202
 203         if os.path.islink(sys.argv[0]):
 204             # called from the build tree
 205             self.source_iotests = os.path.dirname(os.readlink(sys.argv[0]))
 206             self.build_iotests = os.path.dirname(os.path.abspath(sys.argv[0]))
 207         else:
 208             # called from the source tree
 209             self.source_iotests = os.getcwd()
 210             self.build_iotests = self.source_iotests
 211
 212         self.build_root = os.path.join(self.build_iotests, '..', '..')
 213
 214         self.init_directories()
 215         self.init_binaries()
 216
 217         self.malloc_perturb_ = os.getenv('MALLOC_PERTURB_',
 218                                          str(random.randrange(1, 255)))
 219
 220         # QEMU_OPTIONS
 221         self.qemu_options = '-nodefaults -display none -accel qtest'
 222         machine_map = (
 223             ('arm', 'virt'),
 224             ('aarch64', 'virt'),
 225             ('avr', 'mega2560'),
 226             ('m68k', 'virt'),
 227             ('rx', 'gdbsim-r5f562n8'),
 228             ('tricore', 'tricore_testboard')
 229         )
 230         for suffix, machine in machine_map:
 231             if self.qemu_prog.endswith(f'qemu-system-{suffix}'):
 232                 self.qemu_options += f' -machine {machine}'
 233
 234         # QEMU_DEFAULT_MACHINE
 235         self.qemu_default_machine = get_default_machine(self.qemu_prog)
 236
 237         self.qemu_img_options = os.getenv('QEMU_IMG_OPTIONS')
 238         self.qemu_nbd_options = os.getenv('QEMU_NBD_OPTIONS')
 239
 240         is_generic = self.imgfmt not in ['bochs', 'cloop', 'dmg']
 241         self.imgfmt_generic = 'true' if is_generic else 'false'
 242
 243         self.qemu_io_options = f'--cache {self.cachemode} --aio {self.aiomode}'
 244         if self.misalign:
 245             self.qemu_io_options += ' --misalign'
 246
 247         self.qemu_io_options_no_fmt = self.qemu_io_options
 248
 249         if self.imgfmt == 'luks':
 250             self.imgoptssyntax = 'true'
 251             self.imgkeysecret = '123456'
 252             if not self.imgopts:
 253                 self.imgopts = 'iter-time=10'
 254             elif 'iter-time=' not in self.imgopts:
 255                 self.imgopts += ',iter-time=10'
 256         else:
 257             self.imgoptssyntax = 'false'
 258             self.qemu_io_options += ' -f ' + self.imgfmt
 259
 260         if self.imgfmt == 'vmdk':
 261             if not self.imgopts:
 262                 self.imgopts = 'zeroed_grain=on'
 263             elif 'zeroed_grain=' not in self.imgopts:
 264                 self.imgopts += ',zeroed_grain=on'
 265
 266     def close(self) -> None:
 267         if self.tmp_sock_dir:
 268             shutil.rmtree(self.sock_dir)
 269
 270     def __enter__(self) -> 'TestEnv':
 271         return self
 272
 273     def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
 274         self.close()
 275
 276     def print_env(self) -> None:
 277         template = """\
 278 QEMU          -- "{QEMU_PROG}" {QEMU_OPTIONS}
 279 QEMU_IMG      -- "{QEMU_IMG_PROG}" {QEMU_IMG_OPTIONS}
 280 QEMU_IO       -- "{QEMU_IO_PROG}" {QEMU_IO_OPTIONS}
 281 QEMU_NBD      -- "{QEMU_NBD_PROG}" {QEMU_NBD_OPTIONS}
 282 IMGFMT        -- {IMGFMT}{imgopts}
 283 IMGPROTO      -- {IMGPROTO}
 284 PLATFORM      -- {platform}
 285 TEST_DIR      -- {TEST_DIR}
 286 SOCK_DIR      -- {SOCK_DIR}
 287 SOCKET_SCM_HELPER -- {SOCKET_SCM_HELPER}
 288 """
 289
 290         args = collections.defaultdict(str, self.get_env())
 291
 292         if 'IMGOPTS' in args:
 293             args['imgopts'] = f" ({args['IMGOPTS']})"
 294
 295         u = os.uname()
 296         args['platform'] = f'{u.sysname}/{u.machine} {u.nodename} {u.release}'
 297
 298         print(template.format_map(args))