| #!/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) |