[elbe-devel] [RFC PATCH 2/3] commands check-build: Add CheckImage test

[Olivier Dion]

The CheckImage build test can be used to boot images from a build into
an interpreter and make communication with it using a serial line.

This test can be skipped by using the '--skip-img' flag on the CLI.

The test failed if any image emulation failed.  An image emulation can
failed if there's a timeout during an expectation from the guest on
the serial line or if the guest breaks the serial line unexpectedly.

A transcript of the communications between the host and the guest is
printed at the end of each image emulation.

Signed-off-by: Olivier Dion <dion at linutronix.de>
---
 elbepack/commands/check-build.py | 100 +++++++++++++++++++++++++++++--
 1 file changed, 96 insertions(+), 4 deletions(-)

diff --git a/elbepack/commands/check-build.py b/elbepack/commands/check-build.py
index 408c7b62..2027f8c2 100644
--- a/elbepack/commands/check-build.py
+++ b/elbepack/commands/check-build.py
@@ -6,11 +6,12 @@
 import multiprocessing
 import optparse
 import os
-import subprocess
 import sys
 import tempfile
 import traceback
 
+import pexpect
+
 from elbepack.treeutils import etree
 from elbepack.shellhelper import command_out, system_out
 
@@ -32,7 +33,6 @@ class TempDirectory(object):
         shutil.rmtree(self._dir)
         return False
 
-
 def get_files_with_prefix(prefix, _dir="."):
     # TODO - Change os.listdir for os.scandir in Python3
     return [path
@@ -46,6 +46,17 @@ def dpkg_get_infos(path, fmt):
         cmd = 'grep -E "^(%s):" %s' % ("|".join(fmt), path)
     return system_out(cmd)
 
+def open_tgz(path):
+    tmp = tempfile.NamedTemporaryFile(prefix='elbe')
+    command_out("tar --to-stdout --extract --gunzip --file %s" % path,
+                output=tmp)
+    return tmp
+
+def open_img(path):
+    if path.endswith(".tar.gz"):
+        return open_tgz(path)
+    return open(path)
+
 class CheckBase(object):
 
     _tests = []
@@ -279,10 +290,10 @@ def run_command(argv):
     oparser = optparse.OptionParser(usage="usage: %prog check-build [options] <build-dir>")
 
     # pylint: disable=protected-access
-    for _, skip_if in CheckBase._tests:
+    for cls, skip_if in CheckBase._tests:
         oparser.add_option("--%s" % skip_if, action="store_true",
                            dest=skip_if, default=False,
-                           help="Skip test cdroms integrity")
+                           help=cls.__doc__)
 
     (opt, args) = oparser.parse_args(argv)
 
@@ -332,4 +343,84 @@ def run_command(argv):
     print("Passed %d tests ouf of %d" % (total_cnt - fail_cnt, total_cnt))
 
     sys.exit(0 if fail_cnt == 0 else 1)
+
+
+ at CheckBase.register("skip-img")
+class CheckImage(CheckBase):
+
+    """Check if image can boot"""
+
+    def run(self):
+
+        xml = etree("source.xml")
+
+        # For all image
+        for tag in xml.all("./check-img"):
+
+            img_name = tag.text("./name")
+
+            # Open it
+            with open_img(img_name) as img:
+
+                # ELBE_IMG always point to the opened image
+                os.environ["ELBE_IMG"] = img.name
+
+                qemu   = tag.text("./interpreter-name")
+                opts   = os.path.expandvars(tag.text("./interpreter-opts").strip(' \t\n')).split(' ')
+                serial = tag.all("./serial/*")
+
+                transcript = []
+                child      = pexpect.spawn(qemu, opts)
+
+                try:
+                    for action in serial:
+
+                        if "expect" == action.tag:
+
+                            # Try to expect something from the guest
+                            # If there's a timeout; the test fail
+                            # Otherwise; Add to the transcript what we
+                            # received
+                            try:
+                                child.expect(action.et.text)
+                            except pexpect.exceptions.TIMEOUT:
+                                print('Was expecting "%s" but got timeout (%ds)' %
+                                      (action.et.text, child.timeout))
+                                self.ret = 1
+                                break
+                            else:
+                                transcript.append(child.before.decode('utf-8'))
+                                transcript.append(child.after.decode('utf-8'))
+
+                        elif "sendline" == action.tag:
+                            child.sendline(action.et.text)
+
+                        # We're expecting the line to be closed by the
+                        # guest.  If there's a timeout, this means
+                        # that the guest has not closed the line and
+                        # the test failed.  In every case the test
+                        # ends here.
+                        elif "EOF" == action.tag:
+                            try:
+                                child.expect(pexpect.EOF)
+                            except pexpect.exceptions.TIMEOUT:
+                                print('Was expecting EOF but got timeout (%d)' %
+                                      child.timeout)
+                                self.ret = 1
+                            else:
+                                transcript.append(child.before.decode('utf-8'))
+                            break
+
+                # Woops. The guest has die and we didn't expect that!
+                except pexpect.exceptions.EOF:
+                    print("Communication was interrupted unexpectedly")
+                    self.ret = 1
+
+                child.close()
+
+                print("Transcript for image %s:" % img_name)
+                print("----------------------------------------------------------------------")
+                print(''.join(transcript))
+                print("----------------------------------------------------------------------")
+
+                self.ret = self.ret or child.exitstatus
-- 
2.26.2