[mirror_qemu.git] / tests / qemu-iotests / testenv.py

# TestEnv class to manage test environment variables.
#
# Copyright (c) 2020-2021 Virtuozzo International GmbH
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

import os
import sys
import tempfile
from pathlib import Path
import shutil
import collections
import random
import subprocess
import glob
from typing import List, Dict, Any, Optional, ContextManager


def isxfile(path: str) -> bool:
    return os.path.isfile(path) and os.access(path, os.X_OK)


def get_default_machine(qemu_prog: str) -> str:
    outp = subprocess.run([qemu_prog, '-machine', 'help'], check=True,
                          universal_newlines=True,
                          stdout=subprocess.PIPE).stdout

    machines = outp.split('\n')
    try:
        default_machine = next(m for m in machines if m.endswith(' (default)'))
    except StopIteration:
        return ''
    default_machine = default_machine.split(' ', 1)[0]

    alias_suf = ' (alias of {})'.format(default_machine)
    alias = next((m for m in machines if m.endswith(alias_suf)), None)
    if alias is not None:
        default_machine = alias.split(' ', 1)[0]

    return default_machine


class TestEnv(ContextManager['TestEnv']):
    """
    Manage system environment for running tests

    The following variables are supported/provided. They are represented by
    lower-cased TestEnv attributes.
    """

    # We store environment variables as instance attributes, and there are a
    # lot of them. Silence pylint:
    # pylint: disable=too-many-instance-attributes

    env_variables = ['PYTHONPATH', 'TEST_DIR', 'SOCK_DIR', 'SAMPLE_IMG_DIR',
                     'OUTPUT_DIR', 'PYTHON', 'QEMU_PROG', 'QEMU_IMG_PROG',
                     'QEMU_IO_PROG', 'QEMU_NBD_PROG', 'QSD_PROG',
                     'SOCKET_SCM_HELPER', 'QEMU_OPTIONS', 'QEMU_IMG_OPTIONS',
                     'QEMU_IO_OPTIONS', 'QEMU_IO_OPTIONS_NO_FMT',
                     'QEMU_NBD_OPTIONS', 'IMGOPTS', 'IMGFMT', 'IMGPROTO',
                     'AIOMODE', 'CACHEMODE', 'VALGRIND_QEMU',
                     'CACHEMODE_IS_DEFAULT', 'IMGFMT_GENERIC', 'IMGOPTSSYNTAX',
                     'IMGKEYSECRET', 'QEMU_DEFAULT_MACHINE', 'MALLOC_PERTURB_']

    def prepare_subprocess(self, args: List[str]) -> Dict[str, str]:
        if self.debug:
            args.append('-d')

        with open(args[0], encoding="utf-8") as f:
            try:
                if f.readline().rstrip() == '#!/usr/bin/env python3':
                    args.insert(0, self.python)
            except UnicodeDecodeError:  # binary test? for future.
                pass

        os_env = os.environ.copy()
        os_env.update(self.get_env())
        return os_env

    def get_env(self) -> Dict[str, str]:
        env = {}
        for v in self.env_variables:
            val = getattr(self, v.lower(), None)
            if val is not None:
                env[v] = val

        return env

    def init_directories(self) -> None:
        """Init directory variables:
             PYTHONPATH
             TEST_DIR
             SOCK_DIR
             SAMPLE_IMG_DIR
             OUTPUT_DIR
        """
        self.pythonpath = os.getenv('PYTHONPATH')
        if self.pythonpath:
            self.pythonpath = self.source_iotests + os.pathsep + \
                self.pythonpath
        else:
            self.pythonpath = self.source_iotests

        self.test_dir = os.getenv('TEST_DIR',
                                  os.path.join(os.getcwd(), 'scratch'))
        Path(self.test_dir).mkdir(parents=True, exist_ok=True)

        try:
            self.sock_dir = os.environ['SOCK_DIR']
            self.tmp_sock_dir = False
            Path(self.sock_dir).mkdir(parents=True, exist_ok=True)
        except KeyError:
            self.sock_dir = tempfile.mkdtemp()
            self.tmp_sock_dir = True

        self.sample_img_dir = os.getenv('SAMPLE_IMG_DIR',
                                        os.path.join(self.source_iotests,
                                                     'sample_images'))

        self.output_dir = os.getcwd()  # OUTPUT_DIR

    def init_binaries(self) -> None:
        """Init binary path variables:
             PYTHON (for bash tests)
             QEMU_PROG, QEMU_IMG_PROG, QEMU_IO_PROG, QEMU_NBD_PROG, QSD_PROG
             SOCKET_SCM_HELPER
        """
        self.python = sys.executable

        def root(*names: str) -> str:
            return os.path.join(self.build_root, *names)

        arch = os.uname().machine
        if 'ppc64' in arch:
            arch = 'ppc64'

        self.qemu_prog = os.getenv('QEMU_PROG', root(f'qemu-system-{arch}'))
        if not os.path.exists(self.qemu_prog):
            pattern = root('qemu-system-*')
            try:
                progs = sorted(glob.iglob(pattern))
                self.qemu_prog = next(p for p in progs if isxfile(p))
            except StopIteration:
                sys.exit("Not found any Qemu executable binary by pattern "
                         f"'{pattern}'")

        self.qemu_img_prog = os.getenv('QEMU_IMG_PROG', root('qemu-img'))
        self.qemu_io_prog = os.getenv('QEMU_IO_PROG', root('qemu-io'))
        self.qemu_nbd_prog = os.getenv('QEMU_NBD_PROG', root('qemu-nbd'))
        self.qsd_prog = os.getenv('QSD_PROG', root('storage-daemon',
                                                   'qemu-storage-daemon'))

        for b in [self.qemu_img_prog, self.qemu_io_prog, self.qemu_nbd_prog,
                  self.qemu_prog, self.qsd_prog]:
            if not os.path.exists(b):
                sys.exit('No such file: ' + b)
            if not isxfile(b):
                sys.exit('Not executable: ' + b)

        helper_path = os.path.join(self.build_iotests, 'socket_scm_helper')
        if isxfile(helper_path):
            self.socket_scm_helper = helper_path  # SOCKET_SCM_HELPER

    def __init__(self, imgfmt: str, imgproto: str, aiomode: str,
                 cachemode: Optional[str] = None,
                 imgopts: Optional[str] = None,
                 misalign: bool = False,
                 debug: bool = False,
                 valgrind: bool = False) -> None:
        self.imgfmt = imgfmt
        self.imgproto = imgproto
        self.aiomode = aiomode
        self.imgopts = imgopts
        self.misalign = misalign
        self.debug = debug

        if valgrind:
            self.valgrind_qemu = 'y'

        if cachemode is None:
            self.cachemode_is_default = 'true'
            self.cachemode = 'writeback'
        else:
            self.cachemode_is_default = 'false'
            self.cachemode = cachemode

        # Initialize generic paths: build_root, build_iotests, source_iotests,
        # which are needed to initialize some environment variables. They are
        # used by init_*() functions as well.

        if os.path.islink(sys.argv[0]):
            # called from the build tree
            self.source_iotests = os.path.dirname(os.readlink(sys.argv[0]))
            self.build_iotests = os.path.dirname(os.path.abspath(sys.argv[0]))
        else:
            # called from the source tree
            self.source_iotests = os.getcwd()
            self.build_iotests = self.source_iotests

        self.build_root = os.path.join(self.build_iotests, '..', '..')

        self.init_directories()
        self.init_binaries()

        self.malloc_perturb_ = os.getenv('MALLOC_PERTURB_',
                                         str(random.randrange(1, 255)))

        # QEMU_OPTIONS
        self.qemu_options = '-nodefaults -display none -accel qtest'
        machine_map = (
            ('arm', 'virt'),
            ('aarch64', 'virt'),
            ('avr', 'mega2560'),
            ('m68k', 'virt'),
            ('rx', 'gdbsim-r5f562n8'),
            ('tricore', 'tricore_testboard')
        )
        for suffix, machine in machine_map:
            if self.qemu_prog.endswith(f'qemu-system-{suffix}'):
                self.qemu_options += f' -machine {machine}'

        # QEMU_DEFAULT_MACHINE
        self.qemu_default_machine = get_default_machine(self.qemu_prog)

        self.qemu_img_options = os.getenv('QEMU_IMG_OPTIONS')
        self.qemu_nbd_options = os.getenv('QEMU_NBD_OPTIONS')

        is_generic = self.imgfmt not in ['bochs', 'cloop', 'dmg']
        self.imgfmt_generic = 'true' if is_generic else 'false'

        self.qemu_io_options = f'--cache {self.cachemode} --aio {self.aiomode}'
        if self.misalign:
            self.qemu_io_options += ' --misalign'

        self.qemu_io_options_no_fmt = self.qemu_io_options

        if self.imgfmt == 'luks':
            self.imgoptssyntax = 'true'
            self.imgkeysecret = '123456'
            if not self.imgopts:
                self.imgopts = 'iter-time=10'
            elif 'iter-time=' not in self.imgopts:
                self.imgopts += ',iter-time=10'
        else:
            self.imgoptssyntax = 'false'
            self.qemu_io_options += ' -f ' + self.imgfmt

        if self.imgfmt == 'vmdk':
            if not self.imgopts:
                self.imgopts = 'zeroed_grain=on'
            elif 'zeroed_grain=' not in self.imgopts:
                self.imgopts += ',zeroed_grain=on'

    def close(self) -> None:
        if self.tmp_sock_dir:
            shutil.rmtree(self.sock_dir)

    def __enter__(self) -> 'TestEnv':
        return self

    def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
        self.close()

    def print_env(self) -> None:
        template = """\
QEMU          -- "{QEMU_PROG}" {QEMU_OPTIONS}
QEMU_IMG      -- "{QEMU_IMG_PROG}" {QEMU_IMG_OPTIONS}
QEMU_IO       -- "{QEMU_IO_PROG}" {QEMU_IO_OPTIONS}
QEMU_NBD      -- "{QEMU_NBD_PROG}" {QEMU_NBD_OPTIONS}
IMGFMT        -- {IMGFMT}{imgopts}
IMGPROTO      -- {IMGPROTO}
PLATFORM      -- {platform}
TEST_DIR      -- {TEST_DIR}
SOCK_DIR      -- {SOCK_DIR}
SOCKET_SCM_HELPER -- {SOCKET_SCM_HELPER}
"""

        args = collections.defaultdict(str, self.get_env())

        if 'IMGOPTS' in args:
            args['imgopts'] = f" ({args['IMGOPTS']})"

        u = os.uname()
        args['platform'] = f'{u.sysname}/{u.machine} {u.nodename} {u.release}'

        print(template.format_map(args))
Commit	Line	Data
2e5a2f57 VSO	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
c64430d2	28	from typing import List, Dict, Any, Optional, ContextManager
2e5a2f57 VSO	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
c64430d2 PB	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
2e5a2f57 VSO	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
c3d479aa	123	Path(self.sock_dir).mkdir(parents=True, exist_ok=True)
2e5a2f57 VSO	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:
ca502ca6	153	progs = sorted(glob.iglob(pattern))
2e5a2f57 VSO	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'),
7033f1fd	226	('m68k', 'virt'),
2e5a2f57 VSO	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}
480b75ee PB	287	SOCKET_SCM_HELPER -- {SOCKET_SCM_HELPER}
480b75ee PB	288	"""
2e5a2f57 VSO	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))