1
0
mirror of https://github.com/ARM-software/devlib.git synced 2024-10-05 18:30:50 +01:00

target: tests: Address review comments on PR#667

PR#667: https://github.com/ARM-software/devlib/pull/667

- Implement a test module initializer with a tear down method in
  test/test_target.py
- Make various cleanups in test/test_target.py
- Improve structure of test/test_config.yml (previously
  target_configs.yaml)
- Make docstrings Sphinx compatible
- Make ``TargetRunner`` and its subclasses private
- Cleanup tests/test_target.py
- Replace print()'s with appropriate logging calls
- Implement ``NOPTargetRunner`` class for simplifying tests
- Improve Python v3.7 compatibility
- Relax host machine type checking
- Escape user input strings

and more..

Signed-off-by: Metin Kaya <metin.kaya@arm.com>
This commit is contained in:
Metin Kaya 2024-03-27 16:50:22 +00:00 committed by Marc Bonnici
parent 7276097d4e
commit 492d42dddb
11 changed files with 319 additions and 265 deletions

View File

@ -22,8 +22,6 @@ from devlib.target import (
ChromeOsTarget,
)
from devlib.target_runner import QEMUTargetRunner
from devlib.host import (
PACKAGE_BIN_DIRECTORY,
LocalConnection,

View File

@ -19,8 +19,6 @@ Target runner and related classes are implemented here.
import logging
import os
import signal
import subprocess
import time
from platform import machine
@ -37,20 +35,41 @@ class TargetRunner:
It mainly aims to provide framework support for QEMU like target runners
(e.g., :class:`QEMUTargetRunner`).
:param target: Specifies type of target per :class:`Target` based classes.
:type target: Target
"""
def __init__(self,
target):
self.target = target
self.logger = logging.getLogger(self.__class__.__name__)
def __enter__(self):
return self
def __exit__(self, *_):
pass
class SubprocessTargetRunner(TargetRunner):
"""
Class for providing subprocess support to the target runners.
:param runner_cmd: The command to start runner process (e.g.,
``qemu-system-aarch64 -kernel Image -append "console=ttyAMA0" ...``).
:type runner_cmd: str
:type runner_cmd: list(str)
:param target: Specifies type of target per :class:`Target` based classes.
:type target: Target
: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.
@ -62,15 +81,14 @@ class TargetRunner:
target,
connect=True,
boot_timeout=60):
self.boot_timeout = boot_timeout
self.target = target
super().__init__(target=target)
self.logger = logging.getLogger(self.__class__.__name__)
self.boot_timeout = boot_timeout
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
@ -78,20 +96,13 @@ class TargetRunner:
self.wait_boot_complete()
def __enter__(self):
"""
Complementary method for contextmanager.
:return: Self object.
:rtype: TargetRunner
"""
return self
def __exit__(self, *_):
"""
Exit routine for contextmanager.
Ensure :attr:`TargetRunner.runner_process` is terminated on exit.
Ensure ``SubprocessTargetRunner.runner_process`` is terminated on exit.
"""
self.terminate()
@ -99,7 +110,7 @@ class TargetRunner:
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.
``SubprocessTargetRunner.boot_timeout`` seconds.
:raises TargetStableError: In case of timeout.
"""
@ -109,10 +120,10 @@ class TargetRunner:
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,33 +134,41 @@ class TargetRunner:
def terminate(self):
"""
Terminate :attr:`TargetRunner.runner_process`.
Terminate ``SubprocessTargetRunner.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()
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)
self.logger.debug('Killing target runner...')
self.runner_process.kill()
self.runner_process.__exit__(None, None, None)
class QEMUTargetRunner(TargetRunner):
class NOPTargetRunner(TargetRunner):
"""
Class for interacting with QEMU runners.
Class for implementing a target runner which does nothing except providing .target attribute.
:class:`QEMUTargetRunner` is a subclass of :class:`TargetRunner` which performs necessary
groundwork for launching a guest OS on QEMU.
: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
def __exit__(self, *_):
pass
def terminate(self):
"""
Nothing to terminate for NOP target runners.
Defined to be compliant with other runners (e.g., ``SubprocessTargetRunner``).
"""
class QEMUTargetRunner(SubprocessTargetRunner):
"""
Class for preparing necessary groundwork for launching a guest OS on QEMU.
:param qemu_settings: A dictionary which has QEMU related parameters. The full list
of QEMU parameters is below:
@ -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,61 @@ class QEMUTargetRunner(TargetRunner):
'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!')
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!')
# 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"],
)
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["initrd_image"]:
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["enable_kvm"]:
qemu_cmd += ' --enable-kvm'
else:
qemu_cmd += f' -machine virt -cpu {qemu_args["cpu_type"]}'
if qemu_args["enable_kvm"]:
# Enable KVM accelerator if host and guest architectures match.
# Comparison is done based on x86 for the sake of simplicity.
if (qemu_args['arch'].startswith('x86') and machine().startswith('x86')) or (
qemu_args['arch'].startswith('x86') and machine().startswith('x86')):
qemu_cmd.append('--enable-kvm')
self.target = make_target(connect=False,
conn_cls=SshConnection,
connection_settings=self.connection_settings)
# qemu-system-x86_64 does not support -machine virt as of now.
if not qemu_args['arch'].startswith('x86'):
qemu_cmd.extend(['-machine', 'virt', '-cpu', qemu_args["cpu_type"]])
target = make_target(connect=False,
conn_cls=SshConnection,
connection_settings=self.connection_settings)
super().__init__(runner_cmd=qemu_cmd,
target=self.target,
target=target,
**args)

View File

@ -1131,14 +1131,14 @@ fi
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
@ -1147,7 +1147,7 @@ fi
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'

View File

@ -392,10 +392,13 @@ class SshConnection(SshConnectionBase):
)
)
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:

View File

@ -1,5 +0,0 @@
LocalLinuxTarget:
entry-0:
connection_settings:
unrooted: True

View File

@ -1,39 +0,0 @@
AndroidTarget:
entry-0:
timeout: 60
connection_settings:
device: 'emulator-5554'
ChromeOsTarget:
entry-0:
connection_settings:
device: 'emulator-5556'
host: 'example.com'
username: 'username'
password: 'password'
LinuxTarget:
entry-0:
connection_settings:
host: 'example.com'
username: 'username'
password: 'password'
LocalLinuxTarget:
entry-0:
connection_settings:
unrooted: True
QEMUTargetRunner:
entry-0:
qemu_settings:
kernel_image: '/path/to/devlib/tools/buildroot/buildroot-v2023.11.1-aarch64/output/images/Image'
entry-1:
connection_settings:
port : 8023
qemu_settings:
kernel_image: '/path/to/devlib/tools/buildroot/buildroot-v2023.11.1-x86_64/output/images/bzImage'
arch: 'x86_64'
cmdline: 'console=ttyS0'

5
tests/test_config.yml Normal file
View File

@ -0,0 +1,5 @@
target-configs:
entry-0:
LocalLinuxTarget:
connection_settings:
unrooted: True

View File

@ -0,0 +1,40 @@
target-configs:
entry-0:
AndroidTarget:
timeout: 60
connection_settings:
device: 'emulator-5554'
entry-1:
ChromeOsTarget:
connection_settings:
device: 'emulator-5556'
host: 'example.com'
username: 'username'
password: 'password'
entry-2:
LinuxTarget:
connection_settings:
host: 'example.com'
username: 'username'
password: 'password'
entry-3:
LocalLinuxTarget:
connection_settings:
unrooted: True
entry-4:
QEMUTargetRunner:
qemu_settings:
kernel_image: '/path/to/devlib/tools/buildroot/buildroot-v2023.11.1-aarch64/output/images/Image'
entry-5:
QEMUTargetRunner:
connection_settings:
port: 8023
qemu_settings:
kernel_image: '/path/to/devlib/tools/buildroot/buildroot-v2023.11.1-x86_64/output/images/bzImage'
arch: 'x86_64'
cmdline: 'console=ttyS0'

View File

@ -14,136 +14,168 @@
# 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 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"""
logger = logging.getLogger('test_target')
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:
def get_class_object(name):
"""
Get associated class object from string formatted class name
:param name: Class name
:type name: str
:return: Class object
:rtype: object or None
"""
if globals().get(name) is None:
return None
return globals()[name] if issubclass(globals()[name], object) else None
@pytest.fixture(scope="module")
# pylint: disable=too-many-branches
def build_target_runners():
"""Read targets from a YAML formatted config file and create runners for them"""
logger.info("Initializing resources...")
config_file = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'test_config.yml')
test_config = load_struct_from_yaml(config_file)
if test_config is None:
raise ValueError(f'{config_file} looks empty!')
targets = []
target_configs = test_config.get('target-configs')
if target_configs is None:
raise ValueError('No targets found!')
if target_configs.get('AndroidTarget') is not None:
print('> Android targets:')
for entry in target_configs['AndroidTarget'].values():
pp(entry)
target_runners = []
for entry in target_configs.values():
key, target_info = entry.popitem()
target_class = get_class_object(key)
if target_class is AndroidTarget:
logger.info('> Android target: %s', repr(target_info))
a_target = AndroidTarget(
connect=False,
connection_settings=entry['connection_settings'],
connection_settings=target_info['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))
a_target.connect(timeout=target_info.get('timeout', 60))
target_runners.append(NOPTargetRunner(a_target))
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)
elif target_class is ChromeOsTarget:
logger.info('> ChromeOS target: %s', repr(target_info))
c_target = ChromeOsTarget(
connection_settings=entry['connection_settings'],
connection_settings=target_info['connection_settings'],
working_directory='/tmp/devlib-target',
)
targets.append((c_target, None))
target_runners.append(NOPTargetRunner(c_target))
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))
elif target_class is LinuxTarget:
logger.info('> Linux target: %s', repr(target_info))
l_target = LinuxTarget(connection_settings=target_info['connection_settings'])
target_runners.append(NOPTargetRunner(l_target))
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']
elif target_class is LocalLinuxTarget:
logger.info('> LocalLinux target: %s', repr(target_info))
ll_target = LocalLinuxTarget(connection_settings=target_info['connection_settings'])
target_runners.append(NOPTargetRunner(ll_target))
elif target_class is QEMUTargetRunner:
logger.info('> QEMU target runner: %s', repr(target_info))
qemu_runner = QEMUTargetRunner(
qemu_settings=qemu_settings,
connection_settings=connection_settings,
qemu_settings=target_info.get('qemu_settings'),
connection_settings=target_info.get('connection_settings'),
)
if entry.get('ChromeOsTarget') is None:
targets.append((qemu_runner.target, qemu_runner))
continue
if target_info.get('ChromeOsTarget') is not None:
# Leave termination of QEMU runner to ChromeOS target.
target_runners.append(NOPTargetRunner(qemu_runner.target))
# Leave termination of QEMU runner to ChromeOS target.
targets.append((qemu_runner.target, None))
logger.info('>> ChromeOS target: %s', repr(target_info["ChromeOsTarget"]))
qemu_runner.target = ChromeOsTarget(
connection_settings={
**target_info['ChromeOsTarget']['connection_settings'],
**qemu_runner.target.connection_settings,
},
working_directory='/tmp/devlib-target',
)
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))
target_runners.append(qemu_runner)
return targets
else:
raise ValueError(f'Unknown target type {key}!')
yield target_runners
logger.info("Destroying resources...")
for target_runner in target_runners:
target = target_runner.target
# TODO: Revisit per https://github.com/ARM-software/devlib/issues/680.
logger.debug('Removing %s...', target.working_directory)
target.remove(target.working_directory)
target_runner.terminate()
@pytest.mark.parametrize("target, target_runner", build_targets())
def test_read_multiline_values(target, target_runner):
# pylint: disable=redefined-outer-name
def test_read_multiline_values(build_target_runners):
"""
Test Target.read_tree_values_flat()
: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
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',
}
print(f'target={target.__class__.__name__} os={target.os} hostname={target.hostname}')
target_runners = build_target_runners
for target_runner in target_runners:
target = target_runner.target
with target.make_temp() as tempdir:
print(f'Created {tempdir}.')
logger.info('target=%s os=%s hostname=%s',
target.__class__.__name__, target.os, target.hostname)
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)
with target.make_temp() as tempdir:
logger.debug('Created %s.', tempdir)
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()}
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)
print(f'Removing {target.working_directory}...')
target.remove(target.working_directory)
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()}
if target_runner is not None:
print('Terminating target runner...')
target_runner.terminate()
assert {k: v.strip() for k, v in data.items()} == result
assert {k: v.strip() for k, v in data.items()} == result

View File

@ -1,43 +0,0 @@
AndroidTarget:
# Android-12, Pixel-6
entry-0:
timeout: 60
connection_settings:
device: 'emulator-5554'
# Android-14, Pixel-6
entry-1:
connection_settings:
device: 'emulator-5556'
# Android-13, Pixel tablet
entry-2:
connection_settings:
device: 'emulator-5558'
LocalLinuxTarget:
entry-0:
connection_settings:
unrooted: True
QEMUTargetRunner:
entry-0:
qemu_settings:
kernel_image: '/devlib/tools/buildroot/buildroot-v2023.11.1-aarch64/output/images/Image'
ChromeOsTarget:
connection_settings:
device: 'emulator-5558'
entry-1:
connection_settings:
port: 8023
qemu_settings:
kernel_image: '/devlib/tools/buildroot/buildroot-v2023.11.1-x86_64/output/images/bzImage'
arch: 'x86_64'
cmdline: 'console=ttyS0'
ChromeOsTarget:
connection_settings:
device: 'emulator-5558'

View File

@ -0,0 +1,46 @@
target-configs:
entry-0:
# Android-12, Pixel-6
AndroidTarget:
timeout: 60
connection_settings:
device: 'emulator-5554'
entry-1:
# Android-14, Pixel-6
AndroidTarget:
connection_settings:
device: 'emulator-5556'
entry-2:
# Android-13, Pixel tablet
AndroidTarget:
connection_settings:
device: 'emulator-5558'
entry-3:
LocalLinuxTarget:
connection_settings:
unrooted: True
entry-4:
# aarch64 target
QEMUTargetRunner:
qemu_settings:
kernel_image: '/devlib/tools/buildroot/buildroot-v2023.11.1-aarch64/output/images/Image'
ChromeOsTarget:
connection_settings:
device: 'emulator-5558'
entry-5:
# x86_64 target
QEMUTargetRunner:
connection_settings:
port: 8023
qemu_settings:
kernel_image: '/devlib/tools/buildroot/buildroot-v2023.11.1-x86_64/output/images/bzImage'
arch: 'x86_64'
cmdline: 'console=ttyS0'
ChromeOsTarget:
connection_settings:
device: 'emulator-5558'