tools/testing/selftests/tc-testing/plugin-lib/valgrindPlugin.py

   1 '''
   2 run the command under test, under valgrind and collect memory leak info
   3 as a separate test.
   4 '''
   5
   6
   7 import os
   8 import re
   9 import signal
  10 from string import Template
  11 import subprocess
  12 import time
  13 from TdcPlugin import TdcPlugin
  14 from TdcResults import *
  15
  16 from tdc_config import *
  17
  18 def vp_extract_num_from_string(num_as_string_maybe_with_commas):
  19     return int(num_as_string_maybe_with_commas.replace(',',''))
  20
  21 class SubPlugin(TdcPlugin):
  22     def __init__(self):
  23         self.sub_class = 'valgrind/SubPlugin'
  24         self.tap = ''
  25         self._tsr = TestSuiteReport()
  26         super().__init__()
  27
  28     def pre_suite(self, testcount, testist):
  29         '''run commands before test_runner goes into a test loop'''
  30         self.testidlist = [tidx['id'] for tidx in testlist]
  31         super().pre_suite(testcount, testlist)
  32         if self.args.verbose > 1:
  33             print('{}.pre_suite'.format(self.sub_class))
  34         if self.args.valgrind:
  35             self._add_to_tap('1..{}\n'.format(self.testcount))
  36
  37     def post_suite(self, index):
  38         '''run commands after test_runner goes into a test loop'''
  39         super().post_suite(index)
  40         if self.args.verbose > 1:
  41             print('{}.post_suite'.format(self.sub_class))
  42         #print('{}'.format(self.tap))
  43         for xx in range(index - 1, self.testcount):
  44             res = TestResult('{}-mem'.format(self.testidlist[xx]), 'Test skipped')
  45             res.set_result(ResultState.skip)
  46             res.set_errormsg('Skipped because of prior setup/teardown failure')
  47             self._add_results(res)
  48         if self.args.verbose < 4:
  49             subprocess.check_output('rm -f vgnd-*.log', shell=True)
  50
  51     def add_args(self, parser):
  52         super().add_args(parser)
  53         self.argparser_group = self.argparser.add_argument_group(
  54             'valgrind',
  55             'options for valgrindPlugin (run command under test under Valgrind)')
  56
  57         self.argparser_group.add_argument(
  58             '-V', '--valgrind', action='store_true',
  59             help='Run commands under valgrind')
  60
  61         return self.argparser
  62
  63     def adjust_command(self, stage, command):
  64         super().adjust_command(stage, command)
  65         cmdform = 'list'
  66         cmdlist = list()
  67
  68         if not self.args.valgrind:
  69             return command
  70
  71         if self.args.verbose > 1:
  72             print('{}.adjust_command'.format(self.sub_class))
  73
  74         if not isinstance(command, list):
  75             cmdform = 'str'
  76             cmdlist = command.split()
  77         else:
  78             cmdlist = command
  79
  80         if stage == 'execute':
  81             if self.args.verbose > 1:
  82                 print('adjust_command:  stage is {}; inserting valgrind stuff in command [{}] list [{}]'.
  83                       format(stage, command, cmdlist))
  84             cmdlist.insert(0, '--track-origins=yes')
  85             cmdlist.insert(0, '--show-leak-kinds=definite,indirect')
  86             cmdlist.insert(0, '--leak-check=full')
  87             cmdlist.insert(0, '--log-file=vgnd-{}.log'.format(self.args.testid))
  88             cmdlist.insert(0, '-v')  # ask for summary of non-leak errors
  89             cmdlist.insert(0, ENVIR['VALGRIND_BIN'])
  90         else:
  91             pass
  92
  93         if cmdform == 'str':
  94             command = ' '.join(cmdlist)
  95         else:
  96             command = cmdlist
  97
  98         if self.args.verbose > 1:
  99             print('adjust_command:  return command [{}]'.format(command))
 100         return command
 101
 102     def post_execute(self):
 103         if not self.args.valgrind:
 104             return
 105
 106         res = TestResult('{}-mem'.format(self.args.testid),
 107               '{} memory leak check'.format(self.args.test_name))
 108         if self.args.test_skip:
 109             res.set_result(ResultState.skip)
 110             res.set_errormsg('Test case designated as skipped.')
 111             self._add_results(res)
 112             return
 113
 114         self.definitely_lost_re = re.compile(
 115             r'definitely lost:\s+([,0-9]+)\s+bytes in\s+([,0-9]+)\sblocks', re.MULTILINE | re.DOTALL)
 116         self.indirectly_lost_re = re.compile(
 117             r'indirectly lost:\s+([,0-9]+)\s+bytes in\s+([,0-9]+)\s+blocks', re.MULTILINE | re.DOTALL)
 118         self.possibly_lost_re = re.compile(
 119             r'possibly lost:\s+([,0-9]+)bytes in\s+([,0-9]+)\s+blocks', re.MULTILINE | re.DOTALL)
 120         self.non_leak_error_re = re.compile(
 121             r'ERROR SUMMARY:\s+([,0-9]+) errors from\s+([,0-9]+)\s+contexts', re.MULTILINE | re.DOTALL)
 122
 123         def_num = 0
 124         ind_num = 0
 125         pos_num = 0
 126         nle_num = 0
 127
 128         # what about concurrent test runs?  Maybe force them to be in different directories?
 129         with open('vgnd-{}.log'.format(self.args.testid)) as vfd:
 130             content = vfd.read()
 131             def_mo = self.definitely_lost_re.search(content)
 132             ind_mo = self.indirectly_lost_re.search(content)
 133             pos_mo = self.possibly_lost_re.search(content)
 134             nle_mo = self.non_leak_error_re.search(content)
 135
 136             if def_mo:
 137                 def_num = int(def_mo.group(2))
 138             if ind_mo:
 139                 ind_num = int(ind_mo.group(2))
 140             if pos_mo:
 141                 pos_num = int(pos_mo.group(2))
 142             if nle_mo:
 143                 nle_num = int(nle_mo.group(1))
 144
 145         mem_results = ''
 146         if (def_num > 0) or (ind_num > 0) or (pos_num > 0) or (nle_num > 0):
 147             mem_results += 'not '
 148             res.set_result(ResultState.fail)
 149             res.set_failmsg('Memory leak detected')
 150             res.append_failmsg(content)
 151         else:
 152             res.set_result(ResultState.success)
 153
 154         self._add_results(res)
 155
 156
 157     def _add_results(self, res):
 158         self._tsr.add_resultdata(res)
 159
 160     def _add_to_tap(self, more_tap_output):
 161         self.tap += more_tap_output