2 # SPDX-License-Identifier: GPL-2.0
4 from __future__ import print_function
17 import ConfigParser as configparser
20 # Allow multiple values in assignment separated by '|'
26 if (a_item == b_item):
28 elif (a_item == '*') or (b_item == '*'):
33 class Fail(Exception):
34 def __init__(self, test, msg):
38 return '\'%s\' - %s' % (self.test.path, self.msg)
40 class Notest(Exception):
41 def __init__(self, test, arch):
45 return '[%s] \'%s\'' % (self.arch, self.test.path)
47 class Unsup(Exception):
48 def __init__(self, test):
51 return '\'%s\'' % self.test.path
83 'exclude_callchain_kernel',
84 'exclude_callchain_user',
96 log.debug(" %s = %s" % (key, val))
99 def __init__(self, name, data, base):
100 log.debug(" Event %s" % name);
106 def equal(self, other):
107 for t in Event.terms:
108 log.debug(" [%s] %s %s" % (t, self[t], other[t]));
109 if t not in self or t not in other:
111 if not data_equal(self[t], other[t]):
116 if 'optional' in self and self['optional'] == '1':
120 def diff(self, other):
121 for t in Event.terms:
122 if t not in self or t not in other:
124 if not data_equal(self[t], other[t]):
125 log.warning("expected %s=%s, got %s" % (t, self[t], other[t]))
127 # Test file description needs to have following sections:
129 # - just single instance in file
130 # - needs to specify:
131 # 'command' - perf command name
132 # 'args' - special command arguments
133 # 'ret' - expected command return value (0 by default)
134 # 'arch' - architecture specific test (optional)
135 # comma separated list, ! at the beginning
139 # - one or multiple instances in file
140 # - expected values assignments
142 def __init__(self, path, options):
143 parser = configparser.SafeConfigParser()
146 log.warning("running '%s'" % path)
149 self.test_dir = options.test_dir
150 self.perf = options.perf
151 self.command = parser.get('config', 'command')
152 self.args = parser.get('config', 'args')
155 self.ret = parser.get('config', 'ret')
160 self.arch = parser.get('config', 'arch')
161 log.warning("test limitation '%s'" % self.arch)
167 log.debug(" loading expected events");
168 self.load_events(path, self.expect)
170 def is_event(self, name):
171 if name.find("event") == -1:
176 def skip_test(self, myarch):
177 # If architecture not set always run test
179 # log.warning("test for arch %s is ok" % myarch)
182 # Allow multiple values in assignment separated by ','
183 arch_list = self.arch.split(',')
185 # Handle negated list such as !s390x,ppc
186 if arch_list[0][0] == '!':
187 arch_list[0] = arch_list[0][1:]
188 log.warning("excluded architecture list %s" % arch_list)
189 for arch_item in arch_list:
190 # log.warning("test for %s arch is %s" % (arch_item, myarch))
191 if arch_item == myarch:
195 for arch_item in arch_list:
196 # log.warning("test for architecture '%s' current '%s'" % (arch_item, myarch))
197 if arch_item == myarch:
201 def load_events(self, path, events):
202 parser_event = configparser.SafeConfigParser()
203 parser_event.read(path)
205 # The event record section header contains 'event' word,
206 # optionaly followed by ':' allowing to load 'parent
207 # event' first as a base
208 for section in filter(self.is_event, parser_event.sections()):
210 parser_items = parser_event.items(section);
213 # Read parent event if there's any
215 base = section[section.index(':') + 1:]
216 parser_base = configparser.SafeConfigParser()
217 parser_base.read(self.test_dir + '/' + base)
218 base_items = parser_base.items('event')
220 e = Event(section, parser_items, base_items)
223 def run_cmd(self, tempdir):
224 junk1, junk2, junk3, junk4, myarch = (os.uname())
226 if self.skip_test(myarch):
227 raise Notest(self, myarch)
229 cmd = "PERF_TEST_ATTR=%s %s %s -o %s/perf.data %s" % (tempdir,
230 self.perf, self.command, tempdir, self.args)
231 ret = os.WEXITSTATUS(os.system(cmd))
233 log.info(" '%s' ret '%s', expected '%s'" % (cmd, str(ret), str(self.ret)))
235 if not data_equal(str(ret), str(self.ret)):
238 def compare(self, expect, result):
241 log.debug(" compare");
243 # For each expected event find all matching
244 # events in result. Fail if there's not any.
245 for exp_name, exp_event in expect.items():
248 log.debug(" matching [%s]" % exp_name)
249 for res_name, res_event in result.items():
250 log.debug(" to [%s]" % res_name)
251 if (exp_event.equal(res_event)):
252 exp_list.append(res_name)
255 log.debug(" ->FAIL");
257 log.debug(" match: [%s] matches %s" % (exp_name, str(exp_list)))
259 # we did not any matching event - fail
261 if exp_event.optional():
262 log.debug(" %s does not match, but is optional" % exp_name)
265 log.debug(" res_event is empty");
267 exp_event.diff(res_event)
268 raise Fail(self, 'match failure');
270 match[exp_name] = exp_list
272 # For each defined group in the expected events
273 # check we match the same group in the result.
274 for exp_name, exp_event in expect.items():
275 group = exp_event.group
280 for res_name in match[exp_name]:
281 res_group = result[res_name].group
282 if res_group not in match[group]:
283 raise Fail(self, 'group failure')
285 log.debug(" group: [%s] matches group leader %s" %
286 (exp_name, str(match[group])))
288 log.debug(" matched")
290 def resolve_groups(self, events):
291 for name, event in events.items():
292 group_fd = event['group_fd'];
296 for iname, ievent in events.items():
297 if (ievent['fd'] == group_fd):
299 log.debug('[%s] has group leader [%s]' % (name, iname))
303 tempdir = tempfile.mkdtemp();
306 # run the test script
307 self.run_cmd(tempdir);
309 # load events expectation for the test
310 log.debug(" loading result events");
311 for f in glob.glob(tempdir + '/event*'):
312 self.load_events(f, self.result);
314 # resolve group_fd to event names
315 self.resolve_groups(self.expect);
316 self.resolve_groups(self.result);
318 # do the expectation - results matching - both ways
319 self.compare(self.expect, self.result)
320 self.compare(self.result, self.expect)
324 shutil.rmtree(tempdir)
327 def run_tests(options):
328 for f in glob.glob(options.test_dir + '/' + options.test):
330 Test(f, options).run()
332 log.warning("unsupp %s" % obj.getMsg())
333 except Notest as obj:
334 log.warning("skipped %s" % obj.getMsg())
336 def setup_log(verbose):
338 level = logging.CRITICAL
341 level = logging.WARNING
345 level = logging.DEBUG
347 log = logging.getLogger('test')
349 ch = logging.StreamHandler()
351 formatter = logging.Formatter('%(message)s')
352 ch.setFormatter(formatter)
355 USAGE = '''%s [OPTIONS]
357 -p path # perf binary
358 -t test # single test
363 parser = optparse.OptionParser(usage=USAGE)
365 parser.add_option("-t", "--test",
366 action="store", type="string", dest="test")
367 parser.add_option("-d", "--test-dir",
368 action="store", type="string", dest="test_dir")
369 parser.add_option("-p", "--perf",
370 action="store", type="string", dest="perf")
371 parser.add_option("-v", "--verbose",
372 default=0, action="count", dest="verbose")
374 options, args = parser.parse_args()
376 parser.error('FAILED wrong arguments %s' % ' '.join(args))
379 setup_log(options.verbose)
381 if not options.test_dir:
382 print('FAILED no -d option specified')
386 options.test = 'test*'
392 print("FAILED %s" % obj.getMsg())
397 if __name__ == '__main__':