target: tests: Address review comments on PR#667

PR#667: ARM-software#667

- 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 <[email protected]>

devlib/__init__.py

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

devlib/target_runner.py → devlib/_target_runner.py

@@ -19,8 +19,6 @@
 
 import logging
 import os
-import signal
-import subprocess
 import time
 from platform import machine
 
@@ -37,13 +35,14 @@ 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" ...``).
+        Defaults to None.
+    :type runner_cmd: str, optional
+
     :param connect: Specifies if :class:`TargetRunner` should try to connect
         target after launching it, defaults to True.
     :type connect: bool, optional
@@ -58,15 +57,19 @@ 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:
@@ -91,15 +94,15 @@ 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()
 
     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 +112,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 +126,41 @@ 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()
+
+
+class NOPTargetRunner(TargetRunner):
+    """
+    Class for implementing a target runner which does nothing except providing .target attribute.
 
-            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)
+    :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
+
+    def __exit__(self, *_):
+        pass
+
+    def wait_boot_complete(self):
+        pass
+
+    def terminate(self):
+        pass
 
 
 class QEMUTargetRunner(TargetRunner):
@@ -181,11 +200,10 @@ class QEMUTargetRunner(TargetRunner):
     :type qemu_settings: Dict
 
     :param connection_settings: the dictionary to store connection settings
-        of :attr:`Target.connection_settings`, defaults to None.
+        of ``Target.connection_settings``, defaults to None.
     :type connection_settings: Dict, optional
 
-    :param make_target: Lambda function for creating :class:`Target` based
-        object, defaults to :func:`lambda **kwargs: LinuxTarget(**kwargs)`.
+    :param make_target: Lambda function for creating :class:`Target` based object.
     :type make_target: func, optional
 
     :Variable positional arguments: Forwarded to :class:`TargetRunner`.
@@ -196,8 +214,7 @@ 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',
@@ -207,61 +224,53 @@ def __init__(self,
             '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 = f'''{qemu_path} -kernel {qemu_args["kernel_image"]} \
+-append "{qemu_args["cmdline"]}" -m {qemu_args["mem_size"]} -smp \
+cores={qemu_args["num_cores"]},threads={qemu_args["num_threads"]} \
+-netdev 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"]}'
 
-        if qemu_args["arch"] == machine():
+        if qemu_args['arch'].startswith('x86') and machine().startswith('x86'):
             if qemu_args["enable_kvm"]:
                 qemu_cmd += ' --enable-kvm'
         else:
             qemu_cmd += f' -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)

devlib/target.py

@@ -1081,10 +1081,10 @@ def make_temp(self, is_directory=True, directory='', prefix='devlib-test'):
         :type is_directory: bool, optional
 
         :param directory: Temp object will be created under this directory,
-            defaults to :attr:`Target.working_directory`.
+            defaults to ``Target.working_directory``.
         :type directory: str, optional
 
-        :param prefix: Prefix of temp object's name, defaults to 'devlib-test'.
+        :param prefix: Prefix of temp object's name.
         :type prefix: str, optional
 
         :yield: Full path of temp object.
@@ -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'
 

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'