[elbe-devel] [PATCH 1/6] commands: check-build: Create base for checkers

[Olivier Dion]

Checker for a build should inherit from 'CheckBase' and register
themself with 'CheckBase.register(skip_if)' where 'skip_if' is a
command line option that will automatically be created to skip the
test if set. Checker should then implement the 'run' method to do
their testing of the build.  The 'run' method should return 0 for
success.

Checker can assume that their working directory is set to the build
directory before entering 'run'.

Any uncatch exception will result in a failure.  Calling
'self.fail(reason)', returning none 0 from 'run' or setting 'self.ret'
to none 0 will also result in a failure.

Finally, all checkers are run in parallel in different threads.  All
outputs are redirect to the Elbe's logging facility.

Signed-off-by: Olivier Dion <dion at linutronix.de>
---
 elbepack/commands/check-build.py | 113 +++++++++++++++++++++++++++++++
 1 file changed, 113 insertions(+)
 create mode 100644 elbepack/commands/check-build.py

diff --git a/elbepack/commands/check-build.py b/elbepack/commands/check-build.py
new file mode 100644
index 00000000..64da4d9c
--- /dev/null
+++ b/elbepack/commands/check-build.py
@@ -0,0 +1,113 @@
+# ELBE - Debian Based Embedded Rootfilesystem Builder
+# Copyright (c) 2020 Olivier Dion <dion at linutronix.de>
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+import logging
+import multiprocessing.pool
+import optparse
+import os
+import tempfile
+import traceback
+
+from elbepack.log import elbe_logging
+
+# TODO:py3 - Replace with tempfile.TempDirectory
+import shutil
+class TempDirectory(object):
+
+    def __init__(self):
+        self._dir = tempfile.mkdtemp(prefix='elbe')
+
+    def __enter__(self):
+        return self._dir
+
+    def __exit__(self, exec_type, exec_value, tb):
+        shutil.rmtree(self._dir)
+        return False
+
+def run_command(argv):
+
+    oparser = optparse.OptionParser(usage="usage: %prog check-buildf [options] <build-dir>")
+
+    # pylint: disable=protected-access
+    for _, skip_if in CheckBase._tests:
+        oparser.add_option("--%s" % skip_if, action="store_true",
+                           dest=skip_if, default=False,
+                           help="Skip test cdroms integrity")
+
+    (opt, args) = oparser.parse_args(argv)
+
+    if not args:
+        print("No build directory specified")
+        oparser.print_help()
+        os.sys.exit(20)
+
+    # Set working directory for all checkers
+    os.chdir(args[0])
+
+    opt       = vars(opt)
+    total_cnt = 0
+    fail_cnt  = 0
+
+    # pylint: disable=protected-access
+    with elbe_logging({"streams":None}):
+
+        pool = multiprocessing.pool.ThreadPool()
+        results = []
+
+        for cls, skip_if in CheckBase._tests:
+
+            if not opt[skip_if]:
+                logging.info("Starting test %s (%s)", cls.__name__, cls.__doc__)
+                results.append((pool.apply_async(cls()), cls))
+                total_cnt += 1
+
+        for test, cls in results:
+
+            ret = test.get()
+
+            if ret:
+                fail_cnt += 1
+                logging.error("FAILED test %s (%s)", cls.__name__, cls.__doc__)
+
+        logging.info("Passed %d tests ouf of %d",
+                     total_cnt - fail_cnt, total_cnt)
+
+    os.sys.exit(fail_cnt)
+
+class CheckException(Exception):
+    pass
+
+class CheckBase(object):
+
+    _tests = []
+
+    def __init__(self):
+        self.ret = 0
+
+    def __call__(self):
+        with elbe_logging({"streams":None}):
+            try:
+                self.ret = self.run()
+            except CheckException as E:
+                logging.exception(E)
+                self.ret = 1
+            except: # pylint: disable=bare-except
+                logging.error(traceback.format_exc())
+                self.ret = 1
+            return self.ret
+
+    @classmethod
+    def register(cls, skip_if):
+        def _register(test):
+            cls._tests.append((test, skip_if))
+            return test
+        return _register
+
+    # pylint: disable=no-self-use
+    def run(self):
+        raise Exception("Check run method not implemented")
+
+    def fail(self, reason):
+        raise CheckException(reason)
-- 
2.27.0