Add busybox SUT implementation

Author: acerv

ltp/busybox.py

@@ -0,0 +1,233 @@
+"""
+.. module:: busybox
+    :platform: Linux
+    :synopsis: module containing busybox SUT implementation
+
+.. moduleauthor:: Andrea Cervesato <[email protected]>
+"""
+import os
+import re
+import time
+import shutil
+from ltp.sut import SUTError
+from ltp.sut import IOBuffer
+from ltp.qemu import QemuBase
+
+
+class BusyboxSUT(QemuBase):
+    """
+    SUT using qemu to start busybox instances.
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._qemu_cmd = None
+        self._tmpdir = None
+        self._env = None
+        self._cwd = None
+        self._initramfs = None
+        self._kernel = None
+        self._prompt = None
+        self._ram = None
+        self._smp = None
+        self._virtfs = None
+        self._serial_type = None
+        self._opts = None
+
+    @property
+    def name(self) -> str:
+        return "busybox"
+
+    @property
+    def config_help(self) -> dict:
+        return {
+            "initramfs": "initramfs location",
+            "kernel": "kernel image location",
+            "first_prompt": "first shell prompt string (default: '/ #')",
+            "system": "system architecture (default: x86_64)",
+            "ram": "RAM of the VM (default: 2G)",
+            "smp": "number of CPUs (default: 2)",
+            "serial": "type of serial protocol. isa|virtio (default: isa)",
+            "virtfs": "directory to mount inside VM",
+            "options": "user defined options",
+        }
+
+    def _get_transport(self) -> str:
+        """
+        Return a couple of transport_dev and transport_file used by
+        qemu instance for transport configuration.
+        """
+        pid = os.getpid()
+        transport_file = os.path.join(self._tmpdir, f"transport-{pid}")
+        transport_dev = ""
+
+        if self._serial_type == "isa":
+            transport_dev = "/dev/ttyS1"
+        elif self._serial_type == "virtio":
+            transport_dev = "/dev/vport1p1"
+
+        return transport_dev, transport_file
+
+    def _get_command(self) -> str:
+        """
+        Return the full qemu command to execute.
+        """
+        pid = os.getpid()
+        tty_log = os.path.join(self._tmpdir, f"ttyS0-{pid}.log")
+
+        params = []
+        params.append("-enable-kvm")
+        params.append("-display none")
+        params.append(f"-m {self._ram}")
+        params.append(f"-smp {self._smp}")
+        params.append("-device virtio-rng-pci")
+        params.append(f"-chardev stdio,id=tty,logfile={tty_log}")
+
+        if self._serial_type == "isa":
+            params.append("-serial chardev:tty")
+            params.append("-serial chardev:transport")
+        elif self._serial_type == "virtio":
+            params.append("-device virtio-serial")
+            params.append("-device virtconsole,chardev=tty")
+            params.append("-device virtserialport,chardev=transport")
+        else:
+            raise SUTError(f"Unsupported serial type {self._serial_type}")
+
+        _, transport_file = self._get_transport()
+        params.append(f"-chardev file,id=transport,path={transport_file}")
+
+        if self._virtfs:
+            params.append(
+                "-virtfs local,"
+                f"path={self._virtfs},"
+                "mount_tag=host0,"
+                "security_model=mapped-xattr,"
+                "readonly=on")
+
+        if self._opts:
+            params.append(self._opts)
+
+        if self._initramfs:
+            # initramfs might be included inside the kernel
+            params.append(f"-initrd {self._initramfs}")
+
+        params.append(f"-kernel {self._kernel}")
+        params.append("-append 'console=ttyS0'")
+
+        cmd = f"{self._qemu_cmd} {' '.join(params)}"
+
+        return cmd
+
+    def _login(self, timeout: float, iobuffer: IOBuffer) -> None:
+        self._wait_for(self._prompt, timeout, iobuffer)
+        time.sleep(0.2)
+
+        self._write_stdin("stty -echo; stty cols 1024\n")
+        self._wait_for(self._prompt, 5, iobuffer)
+
+        _, retcode, _ = self._exec("export PS1=''", 5, iobuffer)
+        if retcode != 0:
+            raise SUTError("Can't setup prompt string")
+
+        if self._virtfs:
+            _, retcode, _ = self._exec(
+                "mount -t 9p -o trans=virtio host0 /mnt",
+                10, iobuffer)
+            if retcode != 0:
+                raise SUTError("Failed to mount virtfs")
+
+        if self._cwd:
+            _, retcode, _ = self._exec(f"cd {self._cwd}", 5, iobuffer)
+            if retcode != 0:
+                raise SUTError("Can't setup current working directory")
+
+        if self._env:
+            for key, value in self._env.items():
+                _, retcode, _ = self._exec(
+                    f"export {key}={value}",
+                    5, iobuffer)
+                if retcode != 0:
+                    raise SUTError(f"Can't setup env {key}={value}")
+
+    def setup(self, **kwargs: dict) -> None:
+        self._logger.info("Initialize SUT")
+
+        self._env = kwargs.get("env", None)
+        self._cwd = kwargs.get("cwd", None)
+        self._tmpdir = kwargs.get("tmpdir", None)
+        self._initramfs = kwargs.get("initramfs", None)
+        self._kernel = kwargs.get("kernel", None)
+        self._prompt = kwargs.get("first_prompt", "/ #")
+        self._ram = kwargs.get("ram", "2G")
+        self._smp = kwargs.get("smp", "2")
+        self._virtfs = kwargs.get("virtfs", None)
+        self._serial_type = kwargs.get("serial", "isa")
+        self._opts = kwargs.get("options", None)
+
+        system = kwargs.get("system", "x86_64")
+        self._qemu_cmd = f"qemu-system-{system}"
+
+        if not shutil.which(self._qemu_cmd):
+            raise SUTError(f"Command not found: {self._qemu_cmd}")
+
+        if not self._tmpdir or not os.path.isdir(self._tmpdir):
+            raise SUTError(f"Temp directory doesn't exist: {self._tmpdir}")
+
+        if self._initramfs and not os.path.isfile(self._initramfs):
+            raise SUTError(f"initramfs doesn't exist: {self._initramfs}")
+
+        if not self._kernel or not os.path.isfile(self._kernel):
+            raise SUTError(f"kernel doesn't exist: {self._kernel}")
+
+        if not self._prompt:
+            raise SUTError("first_prompt is not defined")
+
+        if not self._ram:
+            raise SUTError("RAM is not defined")
+
+        if not self._smp:
+            raise SUTError("CPU is not defined")
+
+        if self._virtfs and not os.path.isdir(self._virtfs):
+            raise SUTError(f"virtfs directory doesn't exist: {self._virtfs}")
+
+        if self._serial_type not in ["isa", "virtio"]:
+            raise SUTError("Serial protocol must be isa or virtio")
+
+    def get_info(self) -> dict:
+        def _run_cmd(cmd: str) -> str:
+            """
+            Run command, check for returncode and return command's stdout.
+            """
+            ret = self.run_command(cmd, timeout=3)
+            if ret["returncode"] != 0:
+                raise SUTError(f"Can't read information from SUT: {cmd}")
+
+            stdout = ret["stdout"].rstrip()
+
+            return stdout
+
+        kernel = _run_cmd("uname -s -r -v")
+        arch = _run_cmd("uname -m")
+        cpu = _run_cmd("uname -p")
+        meminfo = _run_cmd("cat /proc/meminfo")
+
+        swap_m = re.search(r'SwapTotal:\s+(?P<swap>\d+\s+kB)', meminfo)
+        if not swap_m:
+            raise SUTError("Can't read swap information from /proc/meminfo")
+
+        mem_m = re.search(r'MemTotal:\s+(?P<memory>\d+\s+kB)', meminfo)
+        if not mem_m:
+            raise SUTError("Can't read memory information from /proc/meminfo")
+
+        ret = {
+            "distro": "busybox",
+            "distro_ver": "",
+            "kernel": kernel,
+            "arch": arch,
+            "cpu": cpu,
+            "swap": swap_m.group('swap'),
+            "ram": mem_m.group('memory')
+        }
+
+        return ret