#!/usr/bin/env python3 # ex:ts=4:sw=4:sts=4:et # -*- tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- # # patchtest: execute all unittest test cases discovered for a single patch # # Copyright (C) 2016 Intel Corporation # # SPDX-License-Identifier: GPL-2.0-only # import json import logging import os import signal import subprocess import sys import traceback import unittest # Include current path so test cases can see it sys.path.insert(0, os.path.dirname(os.path.realpath(__file__))) # Include patchtest library sys.path.insert(0, os.path.join(os.path.dirname(os.path.realpath(__file__)), '../meta/lib/patchtest')) from patchtest_parser import PatchtestParser from repo import PatchTestRepo logger = logging.getLogger("patchtest") loggerhandler = logging.StreamHandler() loggerhandler.setFormatter(logging.Formatter("%(message)s")) logger.addHandler(loggerhandler) logger.setLevel(logging.INFO) def _format_test_description(test): return (test.id().split('.')[-1] .replace('_', ' ') .replace("cve", "CVE") .replace("signed off by", "Signed-off-by") .replace("upstream status", "Upstream-Status") .replace("non auh", "non-AUH") .replace("presence format", "presence")) def _write_patchtest_result(line, logfile=None): print(line) if logfile: with open(logfile, "a") as f: f.write(line + "\n") def getResult(patch, mergepatch, logfile=None): class PatchTestResult(unittest.TextTestResult): """ Patchtest TextTestResult """ shouldStop = True longMessage = False success = 'PASS' fail = 'FAIL' skip = 'SKIP' def startTestRun(self): self.repo_error = False self.test_error = False self.test_failure = False try: self.repo = PatchtestParser.repo = PatchTestRepo( patch=patch, repodir=PatchtestParser.repodir, commit=PatchtestParser.basecommit, branch=PatchtestParser.basebranch, ) except: logger.error(traceback.print_exc()) self.repo_error = True self.stop() return if mergepatch: self.repo.merge() def addError(self, test, err): self.test_error = True (ty, va, trace) = err traceback.print_exc() def addFailure(self, test, err): self.test_failure = True desc = _format_test_description(test) issue = json.loads(str(err[1]))["issue"] _write_patchtest_result('{}: {}: {} ({})'.format(self.fail, desc, issue, test.id()), logfile) def addSuccess(self, test): desc = _format_test_description(test) _write_patchtest_result('{}: {} ({})'.format(self.success, desc, test.id()), logfile) def addSkip(self, test, reason): desc = _format_test_description(test) issue = json.loads(str(reason))["issue"] _write_patchtest_result('{}: {}: {} ({})'.format(self.skip, desc, issue, test.id()), logfile) def stopTestRun(self): # in case there was an error on repo object creation, just return if self.repo_error: return self.repo.clean() return PatchTestResult def _runner(resultklass, prefix=None): # load test with the corresponding prefix loader = unittest.TestLoader() if prefix: loader.testMethodPrefix = prefix # create the suite with discovered tests and the corresponding runner suite = loader.discover( start_dir=PatchtestParser.testdir, pattern=PatchtestParser.pattern, top_level_dir=PatchtestParser.topdir, ) # if there are no test cases, just quit if not suite.countTestCases(): return 2 runner = unittest.TextTestRunner(resultclass=resultklass, verbosity=0) try: result = runner.run(suite) except Exception: traceback.print_exc() logger.error('patchtest: something went wrong') return 1 if (result.test_failure or result.test_error) else 0 def run(patch, logfile=None): """ Load, setup and run pre and post-merge tests """ premerge_result = _runner(getResult(patch, False, logfile), 'pretest') postmerge_result = _runner(getResult(patch, True, logfile), 'test') print_result_message(premerge_result, postmerge_result) return premerge_result or postmerge_result def print_result_message(preresult, postresult): print("----------------------------------------------------------------------\n") if preresult == 2 and postresult == 2: logger.error( "patchtest: No test cases found - did you specify the correct suite directory?" ) if preresult == 1 or postresult == 1: logger.error( "WARNING: patchtest: At least one patchtest caused a failure or an error - please check https://wiki.yoctoproject.org/wiki/Patchtest for further guidance" ) else: logger.info("OK: patchtest: All patchtests passed") print("----------------------------------------------------------------------\n") def main(): patch_path = PatchtestParser.patch_path git_status = subprocess.run( ['git', '-C', PatchtestParser.repodir, 'status'], capture_output=True, text=True, ).stdout status_matches = ["Changes not staged for commit", "Changes to be committed"] if any(match in git_status for match in status_matches): logger.error("patchtest: there are uncommitted changes in the target repo that would be overwritten. Please commit or restore them before running patchtest") return 1 builddir = os.environ.get('BUILDDIR') if builddir: bblayers_conf = os.path.join(builddir, 'conf', 'bblayers.conf') if os.path.exists(bblayers_conf): with open(bblayers_conf) as f: if 'meta-selftest' not in f.read(): logger.error( "patchtest: meta-selftest layer not found in %s - add it to BBLAYERS before running patchtest" % bblayers_conf ) return 1 if os.path.isdir(patch_path): patch_list = [os.path.join(patch_path, filename) for filename in sorted(os.listdir(patch_path))] else: patch_list = [patch_path] interrupted = False previous_sigint = signal.getsignal(signal.SIGINT) def _sigint_handler(signum, frame): nonlocal interrupted interrupted = True # Restore previous handler so a second CTRL+C exits immediately signal.signal(signal.SIGINT, previous_sigint) signal.default_int_handler(signum, frame) signal.signal(signal.SIGINT, _sigint_handler) ret = 0 for patch in patch_list: if interrupted: break if os.path.getsize(patch) == 0: logger.error('patchtest: patch is empty') signal.signal(signal.SIGINT, previous_sigint) return 1 logger.info('Testing patch %s' % patch) log_path = None if PatchtestParser.log_results: log_path = patch + ".testresult" with open(log_path, "a") as f: f.write("Patchtest results for patch '%s':\n\n" % patch) try: result = run(patch, log_path) ret = ret or result except KeyboardInterrupt: interrupted = True if interrupted: logger.error('\npatchtest: interrupted') signal.signal(signal.SIGINT, previous_sigint) return 1 signal.signal(signal.SIGINT, previous_sigint) return ret if __name__ == '__main__': ret = 1 # Parse the command line arguments and store it on the PatchtestParser namespace PatchtestParser.set_namespace() # set debugging level if PatchtestParser.debug: logger.setLevel(logging.DEBUG) # if topdir not define, default it to testdir if not PatchtestParser.topdir: PatchtestParser.topdir = PatchtestParser.testdir try: ret = main() except Exception: traceback.print_exc(5) sys.exit(ret)