@@ -12,6 +12,7 @@
import json
import logging
import os
+import subprocess
import sys
import traceback
import unittest
@@ -30,37 +31,50 @@ loggerhandler = logging.StreamHandler()
loggerhandler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(loggerhandler)
logger.setLevel(logging.INFO)
-info = logger.info
-error = logger.error
-def getResult(patch, mergepatch, logfile=None):
+
+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 _emit(line, logfile=None):
+ print(line)
+ if logfile:
+ with open(logfile, "a") as f:
+ f.write(line + "\n")
+
+
+def make_result_class(patch, mergepatch, logfile=None):
class PatchTestResult(unittest.TextTestResult):
""" Patchtest TextTestResult """
shouldStop = True
longMessage = False
- success = 'PASS'
- fail = 'FAIL'
- skip = 'SKIP'
+ success = 'PASS'
+ fail = 'FAIL'
+ skip = 'SKIP'
def startTestRun(self):
- # let's create the repo already, it can be used later on
- repoargs = {
- "repodir": PatchtestParser.repodir,
- "commit": PatchtestParser.basecommit,
- "branch": PatchtestParser.basebranch,
- "patch": patch,
- }
-
- self.repo_error = False
- self.test_error = False
- self.test_failure = False
+ self.repo_error = False
+ self.test_error = False
+ self.test_failure = False
try:
- self.repo = PatchtestParser.repo = PatchTestRepo(**repoargs)
- except:
- logger.error(traceback.print_exc())
+ self.repo = PatchtestParser.repo = PatchTestRepo(
+ patch=patch,
+ repodir=PatchtestParser.repodir,
+ commit=PatchtestParser.basecommit,
+ branch=PatchtestParser.basebranch,
+ )
+ except Exception:
+ traceback.print_exc()
self.repo_error = True
self.stop()
return
@@ -70,100 +84,59 @@ def getResult(patch, mergepatch, logfile=None):
def addError(self, test, err):
self.test_error = True
- (ty, va, trace) = err
- logger.error(traceback.print_exc())
+ traceback.print_exc()
def addFailure(self, test, err):
- test_description = 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")
self.test_failure = True
- fail_str = '{}: {}: {} ({})'.format(self.fail,
- test_description, json.loads(str(err[1]))["issue"],
- test.id())
- print(fail_str)
- if logfile:
- with open(logfile, "a") as f:
- f.write(fail_str + "\n")
+ desc = _format_test_description(test)
+ issue = json.loads(str(err[1]))["issue"]
+ _emit('{}: {}: {} ({})'.format(self.fail, desc, issue, test.id()), logfile)
def addSuccess(self, test):
- test_description = 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")
- success_str = '{}: {} ({})'.format(self.success,
- test_description, test.id())
- print(success_str)
- if logfile:
- with open(logfile, "a") as f:
- f.write(success_str + "\n")
+ desc = _format_test_description(test)
+ _emit('{}: {} ({})'.format(self.success, desc, test.id()), logfile)
def addSkip(self, test, reason):
- test_description = 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")
- skip_str = '{}: {}: {} ({})'.format(self.skip,
- test_description, json.loads(str(reason))["issue"],
- test.id())
- print(skip_str)
- if logfile:
- with open(logfile, "a") as f:
- f.write(skip_str + "\n")
+ desc = _format_test_description(test)
+ issue = json.loads(str(reason))["issue"]
+ _emit('{}: {}: {} ({})'.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()
+ if not self.repo_error:
+ 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,
)
- ntc = suite.countTestCases()
- # if there are no test cases, just quit
- if not ntc:
+ if not suite.countTestCases():
return 2
- runner = unittest.TextTestRunner(resultclass=resultklass, verbosity=0)
+ runner = unittest.TextTestRunner(resultclass=resultklass, verbosity=0)
try:
result = runner.run(suite)
- except:
- logger.error(traceback.print_exc())
+ except Exception:
+ traceback.print_exc()
logger.error('patchtest: something went wrong')
return 1
- if result.test_failure or result.test_error:
- return 1
- return 0
+ 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 """
- # Get the result class and install the control-c handler
unittest.installHandler()
- # run pre-merge tests, meaning those methods with 'pretest' as prefix
- premerge_resultklass = getResult(patch, False, logfile)
- premerge_result = _runner(premerge_resultklass, 'pretest')
-
- # run post-merge tests, meaning those methods with 'test' as prefix
- postmerge_resultklass = getResult(patch, True, logfile)
- postmerge_result = _runner(postmerge_resultklass, 'test')
+ premerge_result = _runner(make_result_class(patch, False, logfile), 'pretest')
+ postmerge_result = _runner(make_result_class(patch, True, logfile), 'test')
print_result_message(premerge_result, postmerge_result)
return premerge_result or postmerge_result
@@ -183,24 +156,34 @@ def print_result_message(preresult, postresult):
print("----------------------------------------------------------------------\n")
def main():
- ret = 0
- tmp_patch = False
patch_path = PatchtestParser.patch_path
- log_results = PatchtestParser.log_results
- log_path = None
- patch_list = None
- git_status = os.popen("(cd %s && git status)" % PatchtestParser.repodir).read()
+ 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]):
+ 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))]
+ patch_list = [os.path.join(patch_path, f) for f in sorted(os.listdir(patch_path))]
else:
patch_list = [patch_path]
+ ret = 0
for patch in patch_list:
if os.path.getsize(patch) == 0:
logger.error('patchtest: patch is empty')
@@ -208,19 +191,13 @@ def main():
logger.info('Testing patch %s' % patch)
- if log_results:
+ 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:
- if log_path:
- ret = run(patch, log_path)
- else:
- ret = run(patch)
- finally:
- if tmp_patch:
- os.remove(patch)
+ ret = run(patch, log_path)
return ret
@@ -241,7 +218,6 @@ if __name__ == '__main__':
try:
ret = main()
except Exception:
- import traceback
traceback.print_exc(5)
sys.exit(ret)
- Stop immediately and print a warning if meta-selftest isn't in bblayers.conf: |patchtest: meta-selftest layer not found in /home/tgamblin/workspace/yocto/openembedded-core/build/conf/bblayers.conf - add it to BBLAYERS before running patchtest - Remove a variety of unused code blocks and inconsistent formatting, then simplify the results-generation logic. This includes removing the 'finally:' block which was never actually reached. AI-Generated: Uses Claude Code Signed-off-by: Trevor Gamblin <tgamblin@baylibre.com> --- scripts/patchtest | 170 ++++++++++++++++++++-------------------------- 1 file changed, 73 insertions(+), 97 deletions(-)