Something went wrong. Try again.
Linux kernel mirror (for testing) git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git
Something went wrong. Try again.
5.7 kB · 161 lines
Python
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162'''run the command under test, under valgrind and collect memory leak infoas a separate test.'''
import osimport reimport signalfrom string import Templateimport subprocessimport timefrom TdcPlugin import TdcPluginfrom TdcResults import *
from tdc_config import *
def vp_extract_num_from_string(num_as_string_maybe_with_commas): return int(num_as_string_maybe_with_commas.replace(',',''))
class SubPlugin(TdcPlugin): def __init__(self): self.sub_class = 'valgrind/SubPlugin' self.tap = '' self._tsr = TestSuiteReport() super().__init__()
def pre_suite(self, testcount, testist): '''run commands before test_runner goes into a test loop''' self.testidlist = [tidx['id'] for tidx in testlist] super().pre_suite(testcount, testlist) if self.args.verbose > 1: print('{}.pre_suite'.format(self.sub_class)) if self.args.valgrind: self._add_to_tap('1..{}\n'.format(self.testcount))
def post_suite(self, index): '''run commands after test_runner goes into a test loop''' super().post_suite(index) if self.args.verbose > 1: print('{}.post_suite'.format(self.sub_class)) #print('{}'.format(self.tap)) for xx in range(index - 1, self.testcount): res = TestResult('{}-mem'.format(self.testidlist[xx]), 'Test skipped') res.set_result(ResultState.skip) res.set_errormsg('Skipped because of prior setup/teardown failure') self._add_results(res) if self.args.verbose < 4: subprocess.check_output('rm -f vgnd-*.log', shell=True)
def add_args(self, parser): super().add_args(parser) self.argparser_group = self.argparser.add_argument_group( 'valgrind', 'options for valgrindPlugin (run command under test under Valgrind)')
self.argparser_group.add_argument( '-V', '--valgrind', action='store_true', help='Run commands under valgrind')
return self.argparser
def adjust_command(self, stage, command): super().adjust_command(stage, command) cmdform = 'list' cmdlist = list()
if not self.args.valgrind: return command
if self.args.verbose > 1: print('{}.adjust_command'.format(self.sub_class))
if not isinstance(command, list): cmdform = 'str' cmdlist = command.split() else: cmdlist = command
if stage == 'execute': if self.args.verbose > 1: print('adjust_command: stage is {}; inserting valgrind stuff in command [{}] list [{}]'. format(stage, command, cmdlist)) cmdlist.insert(0, '--track-origins=yes') cmdlist.insert(0, '--show-leak-kinds=definite,indirect') cmdlist.insert(0, '--leak-check=full') cmdlist.insert(0, '--log-file=vgnd-{}.log'.format(self.args.testid)) cmdlist.insert(0, '-v') # ask for summary of non-leak errors cmdlist.insert(0, ENVIR['VALGRIND_BIN']) else: pass
if cmdform == 'str': command = ' '.join(cmdlist) else: command = cmdlist
if self.args.verbose > 1: print('adjust_command: return command [{}]'.format(command)) return command
def post_execute(self): if not self.args.valgrind: return
res = TestResult('{}-mem'.format(self.args.testid), '{} memory leak check'.format(self.args.test_name)) if self.args.test_skip: res.set_result(ResultState.skip) res.set_errormsg('Test case designated as skipped.') self._add_results(res) return
self.definitely_lost_re = re.compile( r'definitely lost:\s+([,0-9]+)\s+bytes in\s+([,0-9]+)\sblocks', re.MULTILINE | re.DOTALL) self.indirectly_lost_re = re.compile( r'indirectly lost:\s+([,0-9]+)\s+bytes in\s+([,0-9]+)\s+blocks', re.MULTILINE | re.DOTALL) self.possibly_lost_re = re.compile( r'possibly lost:\s+([,0-9]+)bytes in\s+([,0-9]+)\s+blocks', re.MULTILINE | re.DOTALL) self.non_leak_error_re = re.compile( r'ERROR SUMMARY:\s+([,0-9]+) errors from\s+([,0-9]+)\s+contexts', re.MULTILINE | re.DOTALL)
def_num = 0 ind_num = 0 pos_num = 0 nle_num = 0
# what about concurrent test runs? Maybe force them to be in different directories? with open('vgnd-{}.log'.format(self.args.testid)) as vfd: content = vfd.read() def_mo = self.definitely_lost_re.search(content) ind_mo = self.indirectly_lost_re.search(content) pos_mo = self.possibly_lost_re.search(content) nle_mo = self.non_leak_error_re.search(content)
if def_mo: def_num = int(def_mo.group(2)) if ind_mo: ind_num = int(ind_mo.group(2)) if pos_mo: pos_num = int(pos_mo.group(2)) if nle_mo: nle_num = int(nle_mo.group(1))
mem_results = '' if (def_num > 0) or (ind_num > 0) or (pos_num > 0) or (nle_num > 0): mem_results += 'not ' res.set_result(ResultState.fail) res.set_failmsg('Memory leak detected') res.append_failmsg(content) else: res.set_result(ResultState.success)
self._add_results(res)
def _add_results(self, res): self._tsr.add_resultdata(res)
def _add_to_tap(self, more_tap_output): self.tap += more_tap_output