diff options
Diffstat (limited to 'tools/testing/selftests/tc-testing/tdc.py')
| -rwxr-xr-x | tools/testing/selftests/tc-testing/tdc.py | 663 |
1 files changed, 663 insertions, 0 deletions
diff --git a/tools/testing/selftests/tc-testing/tdc.py b/tools/testing/selftests/tc-testing/tdc.py new file mode 100755 index 000000000..7607ba3e3 --- /dev/null +++ b/tools/testing/selftests/tc-testing/tdc.py @@ -0,0 +1,663 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: GPL-2.0 + +""" +tdc.py - Linux tc (Traffic Control) unit test driver + +Copyright (C) 2017 Lucas Bates <lucasb@mojatatu.com> +""" + +import re +import os +import sys +import argparse +import importlib +import json +import subprocess +import time +import traceback +from collections import OrderedDict +from string import Template + +from tdc_config import * +from tdc_helper import * + +import TdcPlugin + + +class PluginMgrTestFail(Exception): + def __init__(self, stage, output, message): + self.stage = stage + self.output = output + self.message = message + +class PluginMgr: + def __init__(self, argparser): + super().__init__() + self.plugins = {} + self.plugin_instances = [] + self.args = [] + self.argparser = argparser + + # TODO, put plugins in order + plugindir = os.getenv('TDC_PLUGIN_DIR', './plugins') + for dirpath, dirnames, filenames in os.walk(plugindir): + for fn in filenames: + if (fn.endswith('.py') and + not fn == '__init__.py' and + not fn.startswith('#') and + not fn.startswith('.#')): + mn = fn[0:-3] + foo = importlib.import_module('plugins.' + mn) + self.plugins[mn] = foo + self.plugin_instances.append(foo.SubPlugin()) + + def call_pre_suite(self, testcount, testidlist): + for pgn_inst in self.plugin_instances: + pgn_inst.pre_suite(testcount, testidlist) + + def call_post_suite(self, index): + for pgn_inst in reversed(self.plugin_instances): + pgn_inst.post_suite(index) + + def call_pre_case(self, test_ordinal, testid): + for pgn_inst in self.plugin_instances: + try: + pgn_inst.pre_case(test_ordinal, testid) + except Exception as ee: + print('exception {} in call to pre_case for {} plugin'. + format(ee, pgn_inst.__class__)) + print('test_ordinal is {}'.format(test_ordinal)) + print('testid is {}'.format(testid)) + raise + + def call_post_case(self): + for pgn_inst in reversed(self.plugin_instances): + pgn_inst.post_case() + + def call_pre_execute(self): + for pgn_inst in self.plugin_instances: + pgn_inst.pre_execute() + + def call_post_execute(self): + for pgn_inst in reversed(self.plugin_instances): + pgn_inst.post_execute() + + def call_add_args(self, parser): + for pgn_inst in self.plugin_instances: + parser = pgn_inst.add_args(parser) + return parser + + def call_check_args(self, args, remaining): + for pgn_inst in self.plugin_instances: + pgn_inst.check_args(args, remaining) + + def call_adjust_command(self, stage, command): + for pgn_inst in self.plugin_instances: + command = pgn_inst.adjust_command(stage, command) + return command + + @staticmethod + def _make_argparser(args): + self.argparser = argparse.ArgumentParser( + description='Linux TC unit tests') + + +def replace_keywords(cmd): + """ + For a given executable command, substitute any known + variables contained within NAMES with the correct values + """ + tcmd = Template(cmd) + subcmd = tcmd.safe_substitute(NAMES) + return subcmd + + +def exec_cmd(args, pm, stage, command): + """ + Perform any required modifications on an executable command, then run + it in a subprocess and return the results. + """ + if len(command.strip()) == 0: + return None, None + if '$' in command: + command = replace_keywords(command) + + command = pm.call_adjust_command(stage, command) + if args.verbose > 0: + print('command "{}"'.format(command)) + proc = subprocess.Popen(command, + shell=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=ENVIR) + (rawout, serr) = proc.communicate() + + if proc.returncode != 0 and len(serr) > 0: + foutput = serr.decode("utf-8", errors="ignore") + else: + foutput = rawout.decode("utf-8", errors="ignore") + + proc.stdout.close() + proc.stderr.close() + return proc, foutput + + +def prepare_env(args, pm, stage, prefix, cmdlist, output = None): + """ + Execute the setup/teardown commands for a test case. + Optionally terminate test execution if the command fails. + """ + if args.verbose > 0: + print('{}'.format(prefix)) + for cmdinfo in cmdlist: + if isinstance(cmdinfo, list): + exit_codes = cmdinfo[1:] + cmd = cmdinfo[0] + else: + exit_codes = [0] + cmd = cmdinfo + + if not cmd: + continue + + (proc, foutput) = exec_cmd(args, pm, stage, cmd) + + if proc and (proc.returncode not in exit_codes): + print('', file=sys.stderr) + print("{} *** Could not execute: \"{}\"".format(prefix, cmd), + file=sys.stderr) + print("\n{} *** Error message: \"{}\"".format(prefix, foutput), + file=sys.stderr) + print("returncode {}; expected {}".format(proc.returncode, + exit_codes)) + print("\n{} *** Aborting test run.".format(prefix), file=sys.stderr) + print("\n\n{} *** stdout ***".format(proc.stdout), file=sys.stderr) + print("\n\n{} *** stderr ***".format(proc.stderr), file=sys.stderr) + raise PluginMgrTestFail( + stage, output, + '"{}" did not complete successfully'.format(prefix)) + +def run_one_test(pm, args, index, tidx): + global NAMES + result = True + tresult = "" + tap = "" + if args.verbose > 0: + print("\t====================\n=====> ", end="") + print("Test " + tidx["id"] + ": " + tidx["name"]) + + # populate NAMES with TESTID for this test + NAMES['TESTID'] = tidx['id'] + + pm.call_pre_case(index, tidx['id']) + prepare_env(args, pm, 'setup', "-----> prepare stage", tidx["setup"]) + + if (args.verbose > 0): + print('-----> execute stage') + pm.call_pre_execute() + (p, procout) = exec_cmd(args, pm, 'execute', tidx["cmdUnderTest"]) + if p: + exit_code = p.returncode + else: + exit_code = None + + pm.call_post_execute() + + if (exit_code is None or exit_code != int(tidx["expExitCode"])): + result = False + print("exit: {!r}".format(exit_code)) + print("exit: {}".format(int(tidx["expExitCode"]))) + #print("exit: {!r} {}".format(exit_code, int(tidx["expExitCode"]))) + print(procout) + else: + if args.verbose > 0: + print('-----> verify stage') + match_pattern = re.compile( + str(tidx["matchPattern"]), re.DOTALL | re.MULTILINE) + (p, procout) = exec_cmd(args, pm, 'verify', tidx["verifyCmd"]) + if procout: + match_index = re.findall(match_pattern, procout) + if len(match_index) != int(tidx["matchCount"]): + result = False + elif int(tidx["matchCount"]) != 0: + result = False + + if not result: + tresult += 'not ' + tresult += 'ok {} - {} # {}\n'.format(str(index), tidx['id'], tidx['name']) + tap += tresult + + if result == False: + if procout: + tap += procout + else: + tap += 'No output!\n' + + prepare_env(args, pm, 'teardown', '-----> teardown stage', tidx['teardown'], procout) + pm.call_post_case() + + index += 1 + + # remove TESTID from NAMES + del(NAMES['TESTID']) + return tap + +def test_runner(pm, args, filtered_tests): + """ + Driver function for the unit tests. + + Prints information about the tests being run, executes the setup and + teardown commands and the command under test itself. Also determines + success/failure based on the information in the test case and generates + TAP output accordingly. + """ + testlist = filtered_tests + tcount = len(testlist) + index = 1 + tap = '' + badtest = None + stage = None + emergency_exit = False + emergency_exit_message = '' + + if args.notap: + if args.verbose: + tap = 'notap requested: omitting test plan\n' + else: + tap = str(index) + ".." + str(tcount) + "\n" + try: + pm.call_pre_suite(tcount, [tidx['id'] for tidx in testlist]) + except Exception as ee: + ex_type, ex, ex_tb = sys.exc_info() + print('Exception {} {} (caught in pre_suite).'. + format(ex_type, ex)) + # when the extra print statements are uncommented, + # the traceback does not appear between them + # (it appears way earlier in the tdc.py output) + # so don't bother ... + # print('--------------------(') + # print('traceback') + traceback.print_tb(ex_tb) + # print('--------------------)') + emergency_exit_message = 'EMERGENCY EXIT, call_pre_suite failed with exception {} {}\n'.format(ex_type, ex) + emergency_exit = True + stage = 'pre-SUITE' + + if emergency_exit: + pm.call_post_suite(index) + return emergency_exit_message + if args.verbose > 1: + print('give test rig 2 seconds to stabilize') + time.sleep(2) + for tidx in testlist: + if "flower" in tidx["category"] and args.device == None: + if args.verbose > 1: + print('Not executing test {} {} because DEV2 not defined'. + format(tidx['id'], tidx['name'])) + continue + try: + badtest = tidx # in case it goes bad + tap += run_one_test(pm, args, index, tidx) + except PluginMgrTestFail as pmtf: + ex_type, ex, ex_tb = sys.exc_info() + stage = pmtf.stage + message = pmtf.message + output = pmtf.output + print(message) + print('Exception {} {} (caught in test_runner, running test {} {} {} stage {})'. + format(ex_type, ex, index, tidx['id'], tidx['name'], stage)) + print('---------------') + print('traceback') + traceback.print_tb(ex_tb) + print('---------------') + if stage == 'teardown': + print('accumulated output for this test:') + if pmtf.output: + print(pmtf.output) + print('---------------') + break + index += 1 + + # if we failed in setup or teardown, + # fill in the remaining tests with ok-skipped + count = index + if not args.notap: + tap += 'about to flush the tap output if tests need to be skipped\n' + if tcount + 1 != index: + for tidx in testlist[index - 1:]: + msg = 'skipped - previous {} failed'.format(stage) + tap += 'ok {} - {} # {} {} {}\n'.format( + count, tidx['id'], msg, index, badtest.get('id', '--Unknown--')) + count += 1 + + tap += 'done flushing skipped test tap output\n' + + if args.pause: + print('Want to pause\nPress enter to continue ...') + if input(sys.stdin): + print('got something on stdin') + + pm.call_post_suite(index) + + return tap + +def has_blank_ids(idlist): + """ + Search the list for empty ID fields and return true/false accordingly. + """ + return not(all(k for k in idlist)) + + +def load_from_file(filename): + """ + Open the JSON file containing the test cases and return them + as list of ordered dictionary objects. + """ + try: + with open(filename) as test_data: + testlist = json.load(test_data, object_pairs_hook=OrderedDict) + except json.JSONDecodeError as jde: + print('IGNORING test case file {}\n\tBECAUSE: {}'.format(filename, jde)) + testlist = list() + else: + idlist = get_id_list(testlist) + if (has_blank_ids(idlist)): + for k in testlist: + k['filename'] = filename + return testlist + + +def args_parse(): + """ + Create the argument parser. + """ + parser = argparse.ArgumentParser(description='Linux TC unit tests') + return parser + + +def set_args(parser): + """ + Set the command line arguments for tdc. + """ + parser.add_argument( + '-p', '--path', type=str, + help='The full path to the tc executable to use') + sg = parser.add_argument_group( + 'selection', 'select which test cases: ' + + 'files plus directories; filtered by categories plus testids') + ag = parser.add_argument_group( + 'action', 'select action to perform on selected test cases') + + sg.add_argument( + '-D', '--directory', nargs='+', metavar='DIR', + help='Collect tests from the specified directory(ies) ' + + '(default [tc-tests])') + sg.add_argument( + '-f', '--file', nargs='+', metavar='FILE', + help='Run tests from the specified file(s)') + sg.add_argument( + '-c', '--category', nargs='*', metavar='CATG', default=['+c'], + help='Run tests only from the specified category/ies, ' + + 'or if no category/ies is/are specified, list known categories.') + sg.add_argument( + '-e', '--execute', nargs='+', metavar='ID', + help='Execute the specified test cases with specified IDs') + ag.add_argument( + '-l', '--list', action='store_true', + help='List all test cases, or those only within the specified category') + ag.add_argument( + '-s', '--show', action='store_true', dest='showID', + help='Display the selected test cases') + ag.add_argument( + '-i', '--id', action='store_true', dest='gen_id', + help='Generate ID numbers for new test cases') + parser.add_argument( + '-v', '--verbose', action='count', default=0, + help='Show the commands that are being run') + parser.add_argument( + '-N', '--notap', action='store_true', + help='Suppress tap results for command under test') + parser.add_argument('-d', '--device', + help='Execute the test case in flower category') + parser.add_argument( + '-P', '--pause', action='store_true', + help='Pause execution just before post-suite stage') + return parser + + +def check_default_settings(args, remaining, pm): + """ + Process any arguments overriding the default settings, + and ensure the settings are correct. + """ + # Allow for overriding specific settings + global NAMES + + if args.path != None: + NAMES['TC'] = args.path + if args.device != None: + NAMES['DEV2'] = args.device + if not os.path.isfile(NAMES['TC']): + print("The specified tc path " + NAMES['TC'] + " does not exist.") + exit(1) + + pm.call_check_args(args, remaining) + + +def get_id_list(alltests): + """ + Generate a list of all IDs in the test cases. + """ + return [x["id"] for x in alltests] + + +def check_case_id(alltests): + """ + Check for duplicate test case IDs. + """ + idl = get_id_list(alltests) + return [x for x in idl if idl.count(x) > 1] + + +def does_id_exist(alltests, newid): + """ + Check if a given ID already exists in the list of test cases. + """ + idl = get_id_list(alltests) + return (any(newid == x for x in idl)) + + +def generate_case_ids(alltests): + """ + If a test case has a blank ID field, generate a random hex ID for it + and then write the test cases back to disk. + """ + import random + for c in alltests: + if (c["id"] == ""): + while True: + newid = str('{:04x}'.format(random.randrange(16**4))) + if (does_id_exist(alltests, newid)): + continue + else: + c['id'] = newid + break + + ufilename = [] + for c in alltests: + if ('filename' in c): + ufilename.append(c['filename']) + ufilename = get_unique_item(ufilename) + for f in ufilename: + testlist = [] + for t in alltests: + if 'filename' in t: + if t['filename'] == f: + del t['filename'] + testlist.append(t) + outfile = open(f, "w") + json.dump(testlist, outfile, indent=4) + outfile.write("\n") + outfile.close() + +def filter_tests_by_id(args, testlist): + ''' + Remove tests from testlist that are not in the named id list. + If id list is empty, return empty list. + ''' + newlist = list() + if testlist and args.execute: + target_ids = args.execute + + if isinstance(target_ids, list) and (len(target_ids) > 0): + newlist = list(filter(lambda x: x['id'] in target_ids, testlist)) + return newlist + +def filter_tests_by_category(args, testlist): + ''' + Remove tests from testlist that are not in a named category. + ''' + answer = list() + if args.category and testlist: + test_ids = list() + for catg in set(args.category): + if catg == '+c': + continue + print('considering category {}'.format(catg)) + for tc in testlist: + if catg in tc['category'] and tc['id'] not in test_ids: + answer.append(tc) + test_ids.append(tc['id']) + + return answer + +def get_test_cases(args): + """ + If a test case file is specified, retrieve tests from that file. + Otherwise, glob for all json files in subdirectories and load from + each one. + Also, if requested, filter by category, and add tests matching + certain ids. + """ + import fnmatch + + flist = [] + testdirs = ['tc-tests'] + + if args.file: + # at least one file was specified - remove the default directory + testdirs = [] + + for ff in args.file: + if not os.path.isfile(ff): + print("IGNORING file " + ff + "\n\tBECAUSE does not exist.") + else: + flist.append(os.path.abspath(ff)) + + if args.directory: + testdirs = args.directory + + for testdir in testdirs: + for root, dirnames, filenames in os.walk(testdir): + for filename in fnmatch.filter(filenames, '*.json'): + candidate = os.path.abspath(os.path.join(root, filename)) + if candidate not in testdirs: + flist.append(candidate) + + alltestcases = list() + for casefile in flist: + alltestcases = alltestcases + (load_from_file(casefile)) + + allcatlist = get_test_categories(alltestcases) + allidlist = get_id_list(alltestcases) + + testcases_by_cats = get_categorized_testlist(alltestcases, allcatlist) + idtestcases = filter_tests_by_id(args, alltestcases) + cattestcases = filter_tests_by_category(args, alltestcases) + + cat_ids = [x['id'] for x in cattestcases] + if args.execute: + if args.category: + alltestcases = cattestcases + [x for x in idtestcases if x['id'] not in cat_ids] + else: + alltestcases = idtestcases + else: + if cat_ids: + alltestcases = cattestcases + else: + # just accept the existing value of alltestcases, + # which has been filtered by file/directory + pass + + return allcatlist, allidlist, testcases_by_cats, alltestcases + + +def set_operation_mode(pm, args): + """ + Load the test case data and process remaining arguments to determine + what the script should do for this run, and call the appropriate + function. + """ + ucat, idlist, testcases, alltests = get_test_cases(args) + + if args.gen_id: + if (has_blank_ids(idlist)): + alltests = generate_case_ids(alltests) + else: + print("No empty ID fields found in test files.") + exit(0) + + duplicate_ids = check_case_id(alltests) + if (len(duplicate_ids) > 0): + print("The following test case IDs are not unique:") + print(str(set(duplicate_ids))) + print("Please correct them before continuing.") + exit(1) + + if args.showID: + for atest in alltests: + print_test_case(atest) + exit(0) + + if isinstance(args.category, list) and (len(args.category) == 0): + print("Available categories:") + print_sll(ucat) + exit(0) + + if args.list: + if args.list: + list_test_cases(alltests) + exit(0) + + if len(alltests): + catresults = test_runner(pm, args, alltests) + else: + catresults = 'No tests found\n' + if args.notap: + print('Tap output suppression requested\n') + else: + print('All test results: \n\n{}'.format(catresults)) + +def main(): + """ + Start of execution; set up argument parser and get the arguments, + and start operations. + """ + parser = args_parse() + parser = set_args(parser) + pm = PluginMgr(parser) + parser = pm.call_add_args(parser) + (args, remaining) = parser.parse_known_args() + args.NAMES = NAMES + check_default_settings(args, remaining, pm) + if args.verbose > 2: + print('args is {}'.format(args)) + + set_operation_mode(pm, args) + + exit(0) + + +if __name__ == "__main__": + main() |