# SPDX-License-Identifier: GPL-2.0
from __future__ import print_function
import os
import sys
import glob
import optparse
import platform
import tempfile
import logging
import re
import shutil
import subprocess
try:
import configparser
except ImportError:
import ConfigParser as configparser
def data_equal(a, b):
# Allow multiple values in assignment separated by '|'
a_list = a.split('|')
b_list = b.split('|')
for a_item in a_list:
for b_item in b_list:
if (a_item == b_item):
return True
elif (a_item == '*') or (b_item == '*'):
return True
return False
class Fail(Exception):
def __init__(self, test, msg):
self.msg = msg
self.test = test
def getMsg(self):
return '\'%s\' - %s' % (self.test.path, self.msg)
class Notest(Exception):
def __init__(self, test, arch):
self.arch = arch
self.test = test
def getMsg(self):
return '[%s] \'%s\'' % (self.arch, self.test.path)
class Unsup(Exception):
def __init__(self, test):
self.test = test
def getMsg(self):
return '\'%s\'' % self.test.path
class Event(dict):
terms = [
'cpu',
'flags',
'type',
'size',
'config',
'sample_period',
'sample_type',
'read_format',
'disabled',
'inherit',
'pinned',
'exclusive',
'exclude_user',
'exclude_kernel',
'exclude_hv',
'exclude_idle',
'mmap',
'comm',
'freq',
'inherit_stat',
'enable_on_exec',
'task',
'watermark',
'precise_ip',
'mmap_data',
'sample_id_all',
'exclude_host',
'exclude_guest',
'exclude_callchain_kernel',
'exclude_callchain_user',
'wakeup_events',
'bp_type',
'config1',
'config2',
'branch_sample_type',
'sample_regs_user',
'sample_stack_user',
]
def add(self, data):
for key, val in data:
log.debug(" %s = %s" % (key, val))
self[key] = val
def __init__(self, name, data, base):
log.debug(" Event %s" % name);
self.name = name;
self.group = ''
self.add(base)
self.add(data)
def equal(self, other):
for t in Event.terms:
log.debug(" [%s] %s %s" % (t, self[t], other[t]));
if t not in self or t not in other:
return False
if not data_equal(self[t], other[t]):
return False
return True
def optional(self):
if 'optional' in self and self['optional'] == '1':
return True
return False
def diff(self, other):
for t in Event.terms:
if t not in self or t not in other:
continue
if not data_equal(self[t], other[t]):
log.warning("expected %s=%s, got %s" % (t, self[t], other[t]))
def parse_version(version):
if not version:
return None
return [int(v) for v in version.split(".")[0:2]]
# Test file description needs to have following sections:
# [config]
# - just single instance in file
# - needs to specify:
# 'command' - perf command name
# 'args' - special command arguments
# 'ret' - Skip test if Perf doesn't exit with this value (0 by default)
# 'test_ret'- If set to 'true', fail test instead of skipping for 'ret' argument
# 'arch' - architecture specific test (optional)
# comma separated list, ! at the beginning
# negates it.
# 'auxv' - Truthy statement that is evaled in the scope of the auxv map. When false,
# the test is skipped. For example 'auxv["AT_HWCAP"] == 10'. (optional)
# 'kernel_since' - Inclusive kernel version from which the test will start running. Only the
# first two values are supported, for example "6.1" (optional)
# 'kernel_until' - Exclusive kernel version from which the test will stop running. (optional)
# [eventX:base]
# - one or multiple instances in file
# - expected values assignments
class Test(object):
def __init__(self, path, options):
parser = configparser.ConfigParser()
parser.read(path)
log.warning("running '%s'" % path)
self.path = path
self.test_dir = options.test_dir
self.perf = options.perf
self.command = parser.get('config', 'command')
self.args = parser.get('config', 'args')
try:
self.ret = parser.get('config', 'ret')
except:
self.ret = 0
self.test_ret = parser.getboolean('config', 'test_ret', fallback=False)
try:
self.arch = parser.get('config', 'arch')
log.warning("test limitation '%s'" % self.arch)
except:
self.arch = ''
self.auxv = parser.get('config', 'auxv', fallback=None)
self.kernel_since = parse_version(parser.get('config', 'kernel_since', fallback=None))
self.kernel_until = parse_version(parser.get('config', 'kernel_until', fallback=None))
self.expect = {}
self.result = {}
log.debug(" loading expected events");
self.load_events(path, self.expect)
def is_event(self, name):
if name.find("event") == -1:
return False
else:
return True
def skip_test_kernel_since(self):
if not self.kernel_since:
return False
return not self.kernel_since <= parse_version(platform.release())
def skip_test_kernel_until(self):
if not self.kernel_until:
return False
return not parse_version(platform.release()) < self.kernel_until
def skip_test_auxv(self):
def new_auxv(a, pattern):
items = list(filter(None, pattern.split(a)))
# AT_HWCAP is hex but doesn't have a prefix, so special case it
if items[0] == "AT_HWCAP":
value = int(items[-1], 16)
else:
try:
value = int(items[-1], 0)
except:
value = items[-1]
return (items[0], value)
if not self.auxv:
return False
auxv = subprocess.check_output("LD_SHOW_AUXV=1 sleep 0", shell=True) \
.decode(sys.stdout.encoding)
pattern = re.compile(r"[: ]+")
auxv = dict([new_auxv(a, pattern) for a in auxv.splitlines()])
return not eval(self.auxv)
def skip_test_arch(self, myarch):
# If architecture not set always run test
if self.arch == '':
# log.warning("test for arch %s is ok" % myarch)
return False
# Allow multiple values in assignment separated by ','
arch_list = self.arch.split(',')
# Handle negated list such as !s390x,ppc
if arch_list[0][0] == '!':
arch_list[0] = arch_list[0][1:]
log.warning("excluded architecture list %s" % arch_list)
for arch_item in arch_list:
# log.warning("test for %s arch is %s" % (arch_item, myarch))
if arch_item == myarch:
return True
return False
for arch_item in arch_list:
# log.warning("test for architecture '%s' current '%s'" % (arch_item, myarch))
if arch_item == myarch:
return False
return True
def load_events(self, path, events):
parser_event = configparser.ConfigParser()
parser_event.read(path)
# The event record section header contains 'event' word,
# optionaly followed by ':' allowing to load 'parent
# event' first as a base
for section in filter(self.is_event, parser_event.sections()):
parser_items = parser_event.items(section);
base_items = {}
# Read parent event if there's any
if (':' in section):
base = section[section.index(':') + 1:]
parser_base = configparser.ConfigParser()
parser_base.read(self.test_dir + '/' + base)
base_items = parser_base.items('event')
e = Event(section, parser_items, base_items)
events[section] = e
def run_cmd(self, tempdir):
junk1, junk2, junk3, junk4, myarch = (os.uname())
if self.skip_test_arch(myarch):
raise Notest(self, myarch)
if self.skip_test_auxv():
raise Notest(self, "auxv skip")
if self.skip_test_kernel_since():
raise Notest(self, "old kernel skip")
if self.skip_test_kernel_until():
raise Notest(self, "new kernel skip")
cmd = "PERF_TEST_ATTR=%s %s %s -o %s/perf.data %s" % (tempdir,
self.perf, self.command, tempdir, self.args)
ret = os.WEXITSTATUS(os.system(cmd))
log.info(" '%s' ret '%s', expected '%s'" % (cmd, str(ret), str(self.ret)))
if not data_equal(str(ret), str(self.ret)):
if self.test_ret:
raise Fail(self, "Perf exit code failure")
else:
raise Unsup(self)
def compare(self, expect, result):
match = {}
log.debug(" compare");
# For each expected event find all matching
# events in result. Fail if there's not any.
for exp_name, exp_event in expect.items():
exp_list = []
res_event = {}
log.debug(" matching [%s]" % exp_name)
for res_name, res_event in result.items():
log.debug(" to [%s]" % res_name)
if (exp_event.equal(res_event)):
exp_list.append(res_name)
log.debug(" ->OK")
else:
log.debug(" ->FAIL");
log.debug(" match: [%s] matches %s" % (exp_name, str(exp_list)))
# we did not any matching event - fail
if not exp_list:
if exp_event.optional():
log.debug(" %s does not match, but is optional" % exp_name)
else:
if not res_event:
log.debug(" res_event is empty");
else:
exp_event.diff(res_event)
raise Fail(self, 'match failure');
match[exp_name] = exp_list
# For each defined group in the expected events
# check we match the same group in the result.
for exp_name, exp_event in expect.items():
group = exp_event.group
if (group == ''):
continue
for res_name in match[exp_name]:
res_group = result[res_name].group
if res_group not in match[group]:
raise Fail(self, 'group failure')
log.debug(" group: [%s] matches group leader %s" %
(exp_name, str(match[group])))
log.debug(" matched")
def resolve_groups(self, events):
for name, event in events.items():
group_fd = event['group_fd'];
if group_fd == '-1':
continue;
for iname, ievent in events.items():
if (ievent['fd'] == group_fd):
event.group = iname
log.debug('[%s] has group leader [%s]' % (name, iname))
break;
def run(self):
tempdir = tempfile.mkdtemp();
try:
# run the test script
self.run_cmd(tempdir);
# load events expectation for the test
log.debug(" loading result events");
for f in glob.glob(tempdir + '/event*'):
self.load_events(f, self.result);
# resolve group_fd to event names
self.resolve_groups(self.expect);
self.resolve_groups(self.result);
# do the expectation - results matching - both ways
self.compare(self.expect, self.result)
self.compare(self.result, self.expect)
finally:
# cleanup
shutil.rmtree(tempdir)
def run_tests(options):
for f in glob.glob(options.test_dir + '/' + options.test):
try:
Test(f, options).run()
except Unsup as obj:
log.warning("unsupp %s" % obj.getMsg())
except Notest as obj:
log.warning("skipped %s" % obj.getMsg())
def setup_log(verbose):
global log
level = logging.CRITICAL
if verbose == 1:
level = logging.WARNING
if verbose == 2:
level = logging.INFO
if verbose >= 3:
level = logging.DEBUG
log = logging.getLogger('test')
log.setLevel(level)
ch = logging.StreamHandler()
ch.setLevel(level)
formatter = logging.Formatter('%(message)s')
ch.setFormatter(formatter)
log.addHandler(ch)
USAGE = '''%s [OPTIONS]
-d dir # tests dir
-p path # perf binary
-t test # single test
-v # verbose level
''' % sys.argv[0]
def main():
parser = optparse.OptionParser(usage=USAGE)
parser.add_option("-t", "--test",
action="store", type="string", dest="test")
parser.add_option("-d", "--test-dir",
action="store", type="string", dest="test_dir")
parser.add_option("-p", "--perf",
action="store", type="string", dest="perf")
parser.add_option("-v", "--verbose",
default=0, action="count", dest="verbose")
options, args = parser.parse_args()
if args:
parser.error('FAILED wrong arguments %s' % ' '.join(args))
return -1
setup_log(options.verbose)
if not options.test_dir:
print('FAILED no -d option specified')
sys.exit(-1)
if not options.test:
options.test = 'test*'
try:
run_tests(options)
except Fail as obj:
print("FAILED %s" % obj.getMsg())
sys.exit(-1)
sys.exit(0)
if __name__ == '__main__':
main()
Codebase Browser
linux