diff --git a/devlib/__init__.py b/devlib/__init__.py index dceda0d62..e496299b1 100644 --- a/devlib/__init__.py +++ b/devlib/__init__.py @@ -22,8 +22,6 @@ ChromeOsTarget, ) -from devlib.target_runner import QEMUTargetRunner - from devlib.host import ( PACKAGE_BIN_DIRECTORY, LocalConnection, diff --git a/devlib/target_runner.py b/devlib/_target_runner.py similarity index 67% rename from devlib/target_runner.py rename to devlib/_target_runner.py index c08c3a1ab..1a463ed81 100644 --- a/devlib/target_runner.py +++ b/devlib/_target_runner.py @@ -19,8 +19,6 @@ import logging import os -import signal -import subprocess import time from platform import machine @@ -37,20 +35,20 @@ class TargetRunner: It mainly aims to provide framework support for QEMU like target runners (e.g., :class:`QEMUTargetRunner`). - :param runner_cmd: The command to start runner process (e.g., - ``qemu-system-aarch64 -kernel Image -append "console=ttyAMA0" ...``). - :type runner_cmd: str - :param target: Specifies type of target per :class:`Target` based classes. :type target: Target + :param runner_cmd: The command to start runner process (e.g., + ``qemu-system-aarch64 -kernel Image -append "console=ttyAMA0" ...``). + :type runner_cmd: list(str) or None + :param connect: Specifies if :class:`TargetRunner` should try to connect target after launching it, defaults to True. - :type connect: bool, optional + :type connect: bool or None :param boot_timeout: Timeout for target's being ready for SSH access in seconds, defaults to 60. - :type boot_timeout: int, optional + :type boot_timeout: int or None :raises HostError: if it cannot execute runner command successfully. @@ -58,19 +56,23 @@ class TargetRunner: """ def __init__(self, - runner_cmd, target, + runner_cmd=None, connect=True, boot_timeout=60): self.boot_timeout = boot_timeout self.target = target + self.runner_process = None self.logger = logging.getLogger(self.__class__.__name__) + if runner_cmd is None: + return + self.logger.info('runner_cmd: %s', runner_cmd) try: - self.runner_process = get_subprocess(list(runner_cmd.split())) + self.runner_process = get_subprocess(runner_cmd) except Exception as ex: raise HostError(f'Error while running "{runner_cmd}": {ex}') from ex @@ -91,7 +93,7 @@ def __exit__(self, *_): """ Exit routine for contextmanager. - Ensure :attr:`TargetRunner.runner_process` is terminated on exit. + Ensure ``TargetRunner.runner_process`` is terminated on exit. """ self.terminate() @@ -99,7 +101,7 @@ def __exit__(self, *_): def wait_boot_complete(self): """ Wait for target OS to finish boot up and become accessible over SSH in at most - :attr:`TargetRunner.boot_timeout` seconds. + ``TargetRunner.boot_timeout`` seconds. :raises TargetStableError: In case of timeout. """ @@ -109,10 +111,10 @@ def wait_boot_complete(self): while self.boot_timeout >= elapsed: try: self.target.connect(timeout=self.boot_timeout - elapsed) - self.logger.info('Target is ready.') + self.logger.debug('Target is ready.') return # pylint: disable=broad-except - except BaseException as ex: + except Exception as ex: self.logger.info('Cannot connect target: %s', ex) time.sleep(1) @@ -123,25 +125,42 @@ def wait_boot_complete(self): def terminate(self): """ - Terminate :attr:`TargetRunner.runner_process`. + Terminate ``TargetRunner.runner_process``. """ if self.runner_process is None: return - try: - self.runner_process.stdin.close() - self.runner_process.stdout.close() - self.runner_process.stderr.close() + self.logger.debug('Killing target runner...') + self.runner_process.kill() + self.runner_process.__exit__(None, None, None) + + +class NOPTargetRunner(TargetRunner): + """ + Class for implementing a target runner which does nothing except providing .target attribute. + + :class:`NOPTargetRunner` is a subclass of :class:`TargetRunner` which is implemented only for + testing purposes. + + :param target: Specifies type of target per :class:`Target` based classes. + :type target: Target + """ + + def __init__(self, target): + super().__init__(target=target) + + def __enter__(self): + return self - if self.runner_process.poll() is None: - self.logger.debug('Terminating target runner...') - os.killpg(self.runner_process.pid, signal.SIGTERM) - # Wait 3 seconds before killing the runner. - self.runner_process.wait(timeout=3) - except subprocess.TimeoutExpired: - self.logger.info('Killing target runner...') - os.killpg(self.runner_process.pid, signal.SIGKILL) + def __exit__(self, *_): + pass + + def wait_boot_complete(self): + pass + + def terminate(self): + pass class QEMUTargetRunner(TargetRunner): @@ -181,12 +200,11 @@ class QEMUTargetRunner(TargetRunner): :type qemu_settings: Dict :param connection_settings: the dictionary to store connection settings - of :attr:`Target.connection_settings`, defaults to None. - :type connection_settings: Dict, optional + of ``Target.connection_settings``, defaults to None. + :type connection_settings: Dict or None - :param make_target: Lambda function for creating :class:`Target` based - object, defaults to :func:`lambda **kwargs: LinuxTarget(**kwargs)`. - :type make_target: func, optional + :param make_target: Lambda function for creating :class:`Target` based object. + :type make_target: func or None :Variable positional arguments: Forwarded to :class:`TargetRunner`. @@ -196,9 +214,9 @@ class QEMUTargetRunner(TargetRunner): def __init__(self, qemu_settings, connection_settings=None, - # pylint: disable=unnecessary-lambda - make_target=lambda **kwargs: LinuxTarget(**kwargs), + make_target=LinuxTarget, **args): + self.connection_settings = { 'host': '127.0.0.1', 'port': 8022, @@ -206,62 +224,56 @@ def __init__(self, 'password': 'root', 'strict_host_check': False, } - - if connection_settings is not None: - self.connection_settings = self.connection_settings | connection_settings + self.connection_settings = {**self.connection_settings, **(connection_settings or {})} qemu_args = { - 'kernel_image': '', 'arch': 'aarch64', 'cpu_type': 'cortex-a72', - 'initrd_image': '', 'mem_size': 512, 'num_cores': 2, 'num_threads': 2, 'cmdline': 'console=ttyAMA0', 'enable_kvm': True, } - - qemu_args = qemu_args | qemu_settings + qemu_args = {**qemu_args, **qemu_settings} qemu_executable = f'qemu-system-{qemu_args["arch"]}' qemu_path = which(qemu_executable) if qemu_path is None: raise FileNotFoundError(f'Cannot find {qemu_executable} executable!') - if not os.path.exists(qemu_args["kernel_image"]): - raise FileNotFoundError(f'{qemu_args["kernel_image"]} does not exist!') - - # pylint: disable=consider-using-f-string - qemu_cmd = '''\ -{} -kernel {} -append "{}" -m {} -smp cores={},threads={} -netdev user,id=net0,hostfwd=tcp::{}-:22 \ --device virtio-net-pci,netdev=net0 --nographic\ -'''.format( - qemu_path, - qemu_args["kernel_image"], - qemu_args["cmdline"], - qemu_args["mem_size"], - qemu_args["num_cores"], - qemu_args["num_threads"], - self.connection_settings["port"], - ) - - if qemu_args["initrd_image"]: + if qemu_args.get("kernel_image"): + if not os.path.exists(qemu_args["kernel_image"]): + raise FileNotFoundError(f'{qemu_args["kernel_image"]} does not exist!') + else: + raise KeyError('qemu_settings must have kernel_image!') + + qemu_cmd = [qemu_path, + '-kernel', qemu_args["kernel_image"], + '-append', f"'{qemu_args['cmdline']}'", + '-m', str(qemu_args["mem_size"]), + '-smp', f'cores={qemu_args["num_cores"]},threads={qemu_args["num_threads"]}', + '-netdev', f'user,id=net0,hostfwd=tcp::{self.connection_settings["port"]}-:22', + '-device', 'virtio-net-pci,netdev=net0', + '--nographic', + ] + + if qemu_args.get("initrd_image"): if not os.path.exists(qemu_args["initrd_image"]): raise FileNotFoundError(f'{qemu_args["initrd_image"]} does not exist!') - qemu_cmd += f' -initrd {qemu_args["initrd_image"]}' + qemu_cmd.extend(['-initrd', qemu_args["initrd_image"]]) - if qemu_args["arch"] == machine(): + if qemu_args['arch'].startswith('x86') and machine().startswith('x86'): if qemu_args["enable_kvm"]: - qemu_cmd += ' --enable-kvm' + qemu_cmd.append('--enable-kvm') else: - qemu_cmd += f' -machine virt -cpu {qemu_args["cpu_type"]}' + qemu_cmd.extend(['-machine', 'virt', '-cpu', qemu_args["cpu_type"]]) - self.target = make_target(connect=False, - conn_cls=SshConnection, - connection_settings=self.connection_settings) + target = make_target(connect=False, + conn_cls=SshConnection, + connection_settings=self.connection_settings) - super().__init__(runner_cmd=qemu_cmd, - target=self.target, + super().__init__(target=target, + runner_cmd=qemu_cmd, **args) diff --git a/devlib/target.py b/devlib/target.py index 0f957201e..0976614c7 100644 --- a/devlib/target.py +++ b/devlib/target.py @@ -1078,14 +1078,14 @@ def make_temp(self, is_directory=True, directory='', prefix='devlib-test'): Creates temporary file/folder on target and deletes it once it's done. :param is_directory: Specifies if temporary object is a directory, defaults to True. - :type is_directory: bool, optional + :type is_directory: bool or None :param directory: Temp object will be created under this directory, - defaults to :attr:`Target.working_directory`. - :type directory: str, optional + defaults to ``Target.working_directory``. + :type directory: str or None - :param prefix: Prefix of temp object's name, defaults to 'devlib-test'. - :type prefix: str, optional + :param prefix: Prefix of temp object's name. + :type prefix: str or None :yield: Full path of temp object. :rtype: str @@ -1094,7 +1094,7 @@ def make_temp(self, is_directory=True, directory='', prefix='devlib-test'): directory = directory or self.working_directory temp_obj = None try: - cmd = f'mktemp -p {directory} {prefix}-XXXXXX' + cmd = f'mktemp -p {quote(directory)} {quote(prefix)}-XXXXXX' if is_directory: cmd += ' -d' diff --git a/devlib/utils/ssh.py b/devlib/utils/ssh.py index fbff5ca94..99baeb983 100644 --- a/devlib/utils/ssh.py +++ b/devlib/utils/ssh.py @@ -391,10 +391,13 @@ def __init__(self, ) ) - except BaseException: - if self.client is not None: - self.client.close() - raise + # pylint: disable=broad-except + except BaseException as e: + try: + if self.client is not None: + self.client.close() + finally: + raise e def _make_client(self): if self.strict_host_check: diff --git a/tests/target_configs.yaml.example b/tests/target_configs.yaml.example index 1154ea8b9..497635693 100644 --- a/tests/target_configs.yaml.example +++ b/tests/target_configs.yaml.example @@ -31,7 +31,7 @@ QEMUTargetRunner: entry-1: connection_settings: - port : 8023 + port: 8023 qemu_settings: kernel_image: '/path/to/devlib/tools/buildroot/buildroot-v2023.11.1-x86_64/output/images/bzImage' diff --git a/tests/test_target.py b/tests/test_target.py index 2d59a2374..1e68933dd 100644 --- a/tests/test_target.py +++ b/tests/test_target.py @@ -14,136 +14,158 @@ # limitations under the License. # -"""Module for testing targets.""" +""" +Module for testing targets. +Sample run with log level is set to DEBUG (see +https://docs.pytest.org/en/7.1.x/how-to/logging.html#live-logs for logging details): + +$ python -m pytest --log-cli-level DEBUG test_target.py +""" + +import logging import os -from pprint import pp -import pytest +from unittest import TestCase -from devlib import AndroidTarget, ChromeOsTarget, LinuxTarget, LocalLinuxTarget, QEMUTargetRunner +from devlib import AndroidTarget, ChromeOsTarget, LinuxTarget, LocalLinuxTarget +from devlib._target_runner import NOPTargetRunner, QEMUTargetRunner from devlib.utils.android import AdbConnection from devlib.utils.misc import load_struct_from_yaml -def build_targets(): - """Read targets from a YAML formatted config file""" - - config_file = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'target_configs.yaml') - - target_configs = load_struct_from_yaml(config_file) - if target_configs is None: - raise ValueError(f'{config_file} looks empty!') - - targets = [] - - if target_configs.get('AndroidTarget') is not None: - print('> Android targets:') - for entry in target_configs['AndroidTarget'].values(): - pp(entry) - a_target = AndroidTarget( - connect=False, - connection_settings=entry['connection_settings'], - conn_cls=lambda **kwargs: AdbConnection(adb_as_root=True, **kwargs), - ) - a_target.connect(timeout=entry.get('timeout', 60)) - targets.append((a_target, None)) - - if target_configs.get('LinuxTarget') is not None: - print('> Linux targets:') - for entry in target_configs['LinuxTarget'].values(): - pp(entry) - l_target = LinuxTarget(connection_settings=entry['connection_settings']) - targets.append((l_target, None)) - - if target_configs.get('ChromeOsTarget') is not None: - print('> ChromeOS targets:') - for entry in target_configs['ChromeOsTarget'].values(): - pp(entry) - c_target = ChromeOsTarget( - connection_settings=entry['connection_settings'], - working_directory='/tmp/devlib-target', - ) - targets.append((c_target, None)) - - if target_configs.get('LocalLinuxTarget') is not None: - print('> LocalLinux targets:') - for entry in target_configs['LocalLinuxTarget'].values(): - pp(entry) - ll_target = LocalLinuxTarget(connection_settings=entry['connection_settings']) - targets.append((ll_target, None)) - - if target_configs.get('QEMUTargetRunner') is not None: - print('> QEMU target runners:') - for entry in target_configs['QEMUTargetRunner'].values(): - pp(entry) - qemu_settings = entry.get('qemu_settings') and entry['qemu_settings'] - connection_settings = entry.get( - 'connection_settings') and entry['connection_settings'] - - qemu_runner = QEMUTargetRunner( - qemu_settings=qemu_settings, - connection_settings=connection_settings, - ) - - if entry.get('ChromeOsTarget') is None: - targets.append((qemu_runner.target, qemu_runner)) - continue - - # Leave termination of QEMU runner to ChromeOS target. - targets.append((qemu_runner.target, None)) - - print('> ChromeOS targets:') - pp(entry['ChromeOsTarget']) - c_target = ChromeOsTarget( - connection_settings={ - **entry['ChromeOsTarget']['connection_settings'], - **qemu_runner.target.connection_settings, - }, - working_directory='/tmp/devlib-target', - ) - targets.append((c_target, qemu_runner)) - - return targets - - -@pytest.mark.parametrize("target, target_runner", build_targets()) -def test_read_multiline_values(target, target_runner): - """ - Test Target.read_tree_values_flat() +logger = logging.getLogger('test_target') - :param target: Type of target per :class:`Target` based classes. - :type target: Target - :param target_runner: Target runner object to terminate target (if necessary). - :type target: TargetRunner +class TestBaseFunctionalities(TestCase): + """ + Class for verifying basic functionalities of various targets. """ - data = { - 'test1': '1', - 'test2': '2\n\n', - 'test3': '3\n\n4\n\n', - } - - print(f'target={target.__class__.__name__} os={target.os} hostname={target.hostname}') - - with target.make_temp() as tempdir: - print(f'Created {tempdir}.') - - for key, value in data.items(): - path = os.path.join(tempdir, key) - print(f'Writing {value!r} to {path}...') - target.write_value(path, value, verify=False, - as_root=target.conn.connected_as_root) - - print('Reading values from target...') - raw_result = target.read_tree_values_flat(tempdir) - result = {os.path.basename(k): v for k, v in raw_result.items()} - - print(f'Removing {target.working_directory}...') - target.remove(target.working_directory) - - if target_runner is not None: - print('Terminating target runner...') - target_runner.terminate() - - assert {k: v.strip() for k, v in data.items()} == result + @classmethod + def setUpClass(cls): + logger.debug("Initializing resources...") + + def build_target_runners(): + """Read targets from a YAML formatted config file and create runners for them""" + + config_file = os.path.join(os.path.dirname( + os.path.realpath(__file__)), 'target_configs.yaml') + + target_configs = load_struct_from_yaml(config_file) + if target_configs is None: + raise ValueError(f'{config_file} looks empty!') + + target_runners = [] + + if target_configs.get('AndroidTarget') is not None: + logger.info('> Android targets:') + for entry in target_configs['AndroidTarget'].values(): + logger.info('%s', repr(entry)) + a_target = AndroidTarget( + connect=False, + connection_settings=entry['connection_settings'], + conn_cls=lambda **kwargs: AdbConnection(adb_as_root=True, **kwargs), + ) + a_target.connect(timeout=entry.get('timeout', 60)) + target_runners.append(NOPTargetRunner(a_target)) + + if target_configs.get('LinuxTarget') is not None: + logger.info('> Linux targets:') + for entry in target_configs['LinuxTarget'].values(): + logger.info('%s', repr(entry)) + l_target = LinuxTarget(connection_settings=entry['connection_settings']) + target_runners.append(NOPTargetRunner(l_target)) + + if target_configs.get('ChromeOsTarget') is not None: + logger.info('> ChromeOS targets:') + for entry in target_configs['ChromeOsTarget'].values(): + logger.info('%s', repr(entry)) + c_target = ChromeOsTarget( + connection_settings=entry['connection_settings'], + working_directory='/tmp/devlib-target', + ) + target_runners.append(NOPTargetRunner(c_target)) + + if target_configs.get('LocalLinuxTarget') is not None: + logger.info('> LocalLinux targets:') + for entry in target_configs['LocalLinuxTarget'].values(): + logger.info('%s', repr(entry)) + ll_target = LocalLinuxTarget(connection_settings=entry['connection_settings']) + target_runners.append(NOPTargetRunner(ll_target)) + + if target_configs.get('QEMUTargetRunner') is not None: + logger.info('> QEMU target runners:') + for entry in target_configs['QEMUTargetRunner'].values(): + logger.info('%s', repr(entry)) + + qemu_runner = QEMUTargetRunner( + qemu_settings=entry.get('qemu_settings'), + connection_settings=entry.get('connection_settings'), + ) + + if entry.get('ChromeOsTarget') is not None: + # Leave termination of QEMU runner to ChromeOS target. + target_runners.append(NOPTargetRunner(qemu_runner.target)) + + logger.info('>> ChromeOS target: %s', repr(entry["ChromeOsTarget"])) + qemu_runner.target = ChromeOsTarget( + connection_settings={ + **entry['ChromeOsTarget']['connection_settings'], + **qemu_runner.target.connection_settings, + }, + working_directory='/tmp/devlib-target', + ) + + target_runners.append(qemu_runner) + + return target_runners + + cls._target_runners = build_target_runners() + + def test_read_multiline_values(self): + """ + Test Target.read_tree_values_flat() + + Runs tests around ``Target.read_tree_values_flat()`` for ``TargetRunner`` objects. + """ + + logger.info('Running test_read_multiline_values test...') + + data = { + 'test1': '1', + 'test2': '2\n\n', + 'test3': '3\n\n4\n\n', + } + + for target_runner in self._target_runners: + target = target_runner.target + + logger.info('target=%s os=%s hostname=%s', + target.__class__.__name__, target.os, target.hostname) + + with target.make_temp() as tempdir: + logger.debug('Created %s.', tempdir) + + for key, value in data.items(): + path = os.path.join(tempdir, key) + logger.debug('Writing %s to %s...', repr(value), path) + target.write_value(path, value, verify=False, + as_root=target.conn.connected_as_root) + + logger.debug('Reading values from target...') + raw_result = target.read_tree_values_flat(tempdir) + result = {os.path.basename(k): v for k, v in raw_result.items()} + + assert {k: v.strip() for k, v in data.items()} == result + + @classmethod + def tearDownClass(cls): + logger.info("Destroying resources...") + + for target_runner in cls._target_runners: + target = target_runner.target + + logger.debug('Removing %s...', target.working_directory) + target.remove(target.working_directory) + + target_runner.terminate()