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