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
Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.