1#!/usr/bin/env python3
2# SPDX-License-Identifier: GPL-2.0
3
4"""
5tdc.py - Linux tc (Traffic Control) unit test driver
6
7Copyright (C) 2017 Lucas Bates <lucasb@mojatatu.com>
8"""
9
10import re
11import os
12import sys
13import argparse
14import importlib
15import json
16import subprocess
17import time
18import traceback
19from collections import OrderedDict
20from string import Template
21
22from tdc_config import *
23from tdc_helper import *
24
25import TdcPlugin
26from TdcResults import *
27
28class PluginDependencyException(Exception):
29    def __init__(self, missing_pg):
30        self.missing_pg = missing_pg
31
32class PluginMgrTestFail(Exception):
33    def __init__(self, stage, output, message):
34        self.stage = stage
35        self.output = output
36        self.message = message
37
38class PluginMgr:
39    def __init__(self, argparser):
40        super().__init__()
41        self.plugins = {}
42        self.plugin_instances = []
43        self.failed_plugins = {}
44        self.argparser = argparser
45
46        # TODO, put plugins in order
47        plugindir = os.getenv('TDC_PLUGIN_DIR', './plugins')
48        for dirpath, dirnames, filenames in os.walk(plugindir):
49            for fn in filenames:
50                if (fn.endswith('.py') and
51                    not fn == '__init__.py' and
52                    not fn.startswith('#') and
53                    not fn.startswith('.#')):
54                    mn = fn[0:-3]
55                    foo = importlib.import_module('plugins.' + mn)
56                    self.plugins[mn] = foo
57                    self.plugin_instances.append(foo.SubPlugin())
58
59    def load_plugin(self, pgdir, pgname):
60        pgname = pgname[0:-3]
61        foo = importlib.import_module('{}.{}'.format(pgdir, pgname))
62        self.plugins[pgname] = foo
63        self.plugin_instances.append(foo.SubPlugin())
64        self.plugin_instances[-1].check_args(self.args, None)
65
66    def get_required_plugins(self, testlist):
67        '''
68        Get all required plugins from the list of test cases and return
69        all unique items.
70        '''
71        reqs = []
72        for t in testlist:
73            try:
74                if 'requires' in t['plugins']:
75                    if isinstance(t['plugins']['requires'], list):
76                        reqs.extend(t['plugins']['requires'])
77                    else:
78                        reqs.append(t['plugins']['requires'])
79            except KeyError:
80                continue
81        reqs = get_unique_item(reqs)
82        return reqs
83
84    def load_required_plugins(self, reqs, parser, args, remaining):
85        '''
86        Get all required plugins from the list of test cases and load any plugin
87        that is not already enabled.
88        '''
89        pgd = ['plugin-lib', 'plugin-lib-custom']
90        pnf = []
91
92        for r in reqs:
93            if r not in self.plugins:
94                fname = '{}.py'.format(r)
95                source_path = []
96                for d in pgd:
97                    pgpath = '{}/{}'.format(d, fname)
98                    if os.path.isfile(pgpath):
99                        source_path.append(pgpath)
100                if len(source_path) == 0:
101                    print('ERROR: unable to find required plugin {}'.format(r))
102                    pnf.append(fname)
103                    continue
104                elif len(source_path) > 1:
105                    print('WARNING: multiple copies of plugin {} found, using version found')
106                    print('at {}'.format(source_path[0]))
107                pgdir = source_path[0]
108                pgdir = pgdir.split('/')[0]
109                self.load_plugin(pgdir, fname)
110        if len(pnf) > 0:
111            raise PluginDependencyException(pnf)
112
113        parser = self.call_add_args(parser)
114        (args, remaining) = parser.parse_known_args(args=remaining, namespace=args)
115        return args
116
117    def call_pre_suite(self, testcount, testidlist):
118        for pgn_inst in self.plugin_instances:
119            pgn_inst.pre_suite(testcount, testidlist)
120
121    def call_post_suite(self, index):
122        for pgn_inst in reversed(self.plugin_instances):
123            pgn_inst.post_suite(index)
124
125    def call_pre_case(self, caseinfo, *, test_skip=False):
126        for pgn_inst in self.plugin_instances:
127            try:
128                pgn_inst.pre_case(caseinfo, test_skip)
129            except Exception as ee:
130                print('exception {} in call to pre_case for {} plugin'.
131                      format(ee, pgn_inst.__class__))
132                print('test_ordinal is {}'.format(test_ordinal))
133                print('testid is {}'.format(caseinfo['id']))
134                raise
135
136    def call_post_case(self):
137        for pgn_inst in reversed(self.plugin_instances):
138            pgn_inst.post_case()
139
140    def call_pre_execute(self):
141        for pgn_inst in self.plugin_instances:
142            pgn_inst.pre_execute()
143
144    def call_post_execute(self):
145        for pgn_inst in reversed(self.plugin_instances):
146            pgn_inst.post_execute()
147
148    def call_add_args(self, parser):
149        for pgn_inst in self.plugin_instances:
150            parser = pgn_inst.add_args(parser)
151        return parser
152
153    def call_check_args(self, args, remaining):
154        for pgn_inst in self.plugin_instances:
155            pgn_inst.check_args(args, remaining)
156
157    def call_adjust_command(self, stage, command):
158        for pgn_inst in self.plugin_instances:
159            command = pgn_inst.adjust_command(stage, command)
160        return command
161
162    def set_args(self, args):
163        self.args = args
164
165    @staticmethod
166    def _make_argparser(args):
167        self.argparser = argparse.ArgumentParser(
168            description='Linux TC unit tests')
169
170def replace_keywords(cmd):
171    """
172    For a given executable command, substitute any known
173    variables contained within NAMES with the correct values
174    """
175    tcmd = Template(cmd)
176    subcmd = tcmd.safe_substitute(NAMES)
177    return subcmd
178
179
180def exec_cmd(args, pm, stage, command):
181    """
182    Perform any required modifications on an executable command, then run
183    it in a subprocess and return the results.
184    """
185    if len(command.strip()) == 0:
186        return None, None
187    if '$' in command:
188        command = replace_keywords(command)
189
190    command = pm.call_adjust_command(stage, command)
191    if args.verbose > 0:
192        print('command "{}"'.format(command))
193    proc = subprocess.Popen(command,
194        shell=True,
195        stdout=subprocess.PIPE,
196        stderr=subprocess.PIPE,
197        env=ENVIR)
198
199    try:
200        (rawout, serr) = proc.communicate(timeout=NAMES['TIMEOUT'])
201        if proc.returncode != 0 and len(serr) > 0:
202            foutput = serr.decode("utf-8", errors="ignore")
203        else:
204            foutput = rawout.decode("utf-8", errors="ignore")
205    except subprocess.TimeoutExpired:
206        foutput = "Command \"{}\" timed out\n".format(command)
207        proc.returncode = 255
208
209    proc.stdout.close()
210    proc.stderr.close()
211    return proc, foutput
212
213
214def prepare_env(args, pm, stage, prefix, cmdlist, output = None):
215    """
216    Execute the setup/teardown commands for a test case.
217    Optionally terminate test execution if the command fails.
218    """
219    if args.verbose > 0:
220        print('{}'.format(prefix))
221    for cmdinfo in cmdlist:
222        if isinstance(cmdinfo, list):
223            exit_codes = cmdinfo[1:]
224            cmd = cmdinfo[0]
225        else:
226            exit_codes = [0]
227            cmd = cmdinfo
228
229        if not cmd:
230            continue
231
232        (proc, foutput) = exec_cmd(args, pm, stage, cmd)
233
234        if proc and (proc.returncode not in exit_codes):
235            print('', file=sys.stderr)
236            print("{} *** Could not execute: \"{}\"".format(prefix, cmd),
237                  file=sys.stderr)
238            print("\n{} *** Error message: \"{}\"".format(prefix, foutput),
239                  file=sys.stderr)
240            print("returncode {}; expected {}".format(proc.returncode,
241                                                      exit_codes))
242            print("\n{} *** Aborting test run.".format(prefix), file=sys.stderr)
243            print("\n\n{} *** stdout ***".format(proc.stdout), file=sys.stderr)
244            print("\n\n{} *** stderr ***".format(proc.stderr), file=sys.stderr)
245            raise PluginMgrTestFail(
246                stage, output,
247                '"{}" did not complete successfully'.format(prefix))
248
249def verify_by_json(procout, res, tidx, args, pm):
250    try:
251        outputJSON = json.loads(procout)
252    except json.JSONDecodeError:
253        res.set_result(ResultState.fail)
254        res.set_failmsg('Cannot decode verify command\'s output. Is it JSON?')
255        return res
256
257    matchJSON = json.loads(json.dumps(tidx['matchJSON']))
258
259    if type(outputJSON) != type(matchJSON):
260        failmsg = 'Original output and matchJSON value are not the same type: output: {} != matchJSON: {} '
261        failmsg = failmsg.format(type(outputJSON).__name__, type(matchJSON).__name__)
262        res.set_result(ResultState.fail)
263        res.set_failmsg(failmsg)
264        return res
265
266    if len(matchJSON) > len(outputJSON):
267        failmsg = "Your matchJSON value is an array, and it contains more elements than the command under test\'s output:\ncommand output (length: {}):\n{}\nmatchJSON value (length: {}):\n{}"
268        failmsg = failmsg.format(len(outputJSON), outputJSON, len(matchJSON), matchJSON)
269        res.set_result(ResultState.fail)
270        res.set_failmsg(failmsg)
271        return res
272    res = find_in_json(res, outputJSON, matchJSON, 0)
273
274    return res
275
276def find_in_json(res, outputJSONVal, matchJSONVal, matchJSONKey=None):
277    if res.get_result() == ResultState.fail:
278        return res
279
280    if type(matchJSONVal) == list:
281        res = find_in_json_list(res, outputJSONVal, matchJSONVal, matchJSONKey)
282
283    elif type(matchJSONVal) == dict:
284        res = find_in_json_dict(res, outputJSONVal, matchJSONVal)
285    else:
286        res = find_in_json_other(res, outputJSONVal, matchJSONVal, matchJSONKey)
287
288    if res.get_result() != ResultState.fail:
289        res.set_result(ResultState.success)
290        return res
291
292    return res
293
294def find_in_json_list(res, outputJSONVal, matchJSONVal, matchJSONKey=None):
295    if (type(matchJSONVal) != type(outputJSONVal)):
296        failmsg = 'Original output and matchJSON value are not the same type: output: {} != matchJSON: {}'
297        failmsg = failmsg.format(outputJSONVal, matchJSONVal)
298        res.set_result(ResultState.fail)
299        res.set_failmsg(failmsg)
300        return res
301
302    if len(matchJSONVal) > len(outputJSONVal):
303        failmsg = "Your matchJSON value is an array, and it contains more elements than the command under test\'s output:\ncommand output (length: {}):\n{}\nmatchJSON value (length: {}):\n{}"
304        failmsg = failmsg.format(len(outputJSONVal), outputJSONVal, len(matchJSONVal), matchJSONVal)
305        res.set_result(ResultState.fail)
306        res.set_failmsg(failmsg)
307        return res
308
309    for matchJSONIdx, matchJSONVal in enumerate(matchJSONVal):
310        res = find_in_json(res, outputJSONVal[matchJSONIdx], matchJSONVal,
311                           matchJSONKey)
312    return res
313
314def find_in_json_dict(res, outputJSONVal, matchJSONVal):
315    for matchJSONKey, matchJSONVal in matchJSONVal.items():
316        if type(outputJSONVal) == dict:
317            if matchJSONKey not in outputJSONVal:
318                failmsg = 'Key not found in json output: {}: {}\nMatching against output: {}'
319                failmsg = failmsg.format(matchJSONKey, matchJSONVal, outputJSONVal)
320                res.set_result(ResultState.fail)
321                res.set_failmsg(failmsg)
322                return res
323
324        else:
325            failmsg = 'Original output and matchJSON value are not the same type: output: {} != matchJSON: {}'
326            failmsg = failmsg.format(type(outputJSON).__name__, type(matchJSON).__name__)
327            res.set_result(ResultState.fail)
328            res.set_failmsg(failmsg)
329            return rest
330
331        if type(outputJSONVal) == dict and (type(outputJSONVal[matchJSONKey]) == dict or
332                type(outputJSONVal[matchJSONKey]) == list):
333            if len(matchJSONVal) > 0:
334                res = find_in_json(res, outputJSONVal[matchJSONKey], matchJSONVal, matchJSONKey)
335            # handling corner case where matchJSONVal == [] or matchJSONVal == {}
336            else:
337                res = find_in_json_other(res, outputJSONVal, matchJSONVal, matchJSONKey)
338        else:
339            res = find_in_json(res, outputJSONVal, matchJSONVal, matchJSONKey)
340    return res
341
342def find_in_json_other(res, outputJSONVal, matchJSONVal, matchJSONKey=None):
343    if matchJSONKey in outputJSONVal:
344        if matchJSONVal != outputJSONVal[matchJSONKey]:
345            failmsg = 'Value doesn\'t match: {}: {} != {}\nMatching against output: {}'
346            failmsg = failmsg.format(matchJSONKey, matchJSONVal, outputJSONVal[matchJSONKey], outputJSONVal)
347            res.set_result(ResultState.fail)
348            res.set_failmsg(failmsg)
349            return res
350
351    return res
352
353def run_one_test(pm, args, index, tidx):
354    global NAMES
355    result = True
356    tresult = ""
357    tap = ""
358    res = TestResult(tidx['id'], tidx['name'])
359    if args.verbose > 0:
360        print("\t====================\n=====> ", end="")
361    print("Test " + tidx["id"] + ": " + tidx["name"])
362
363    if 'skip' in tidx:
364        if tidx['skip'] == 'yes':
365            res = TestResult(tidx['id'], tidx['name'])
366            res.set_result(ResultState.skip)
367            res.set_errormsg('Test case designated as skipped.')
368            pm.call_pre_case(tidx, test_skip=True)
369            pm.call_post_execute()
370            return res
371
372    if 'dependsOn' in tidx:
373        if (args.verbose > 0):
374            print('probe command for test skip')
375        (p, procout) = exec_cmd(args, pm, 'execute', tidx['dependsOn'])
376        if p:
377            if (p.returncode != 0):
378                res = TestResult(tidx['id'], tidx['name'])
379                res.set_result(ResultState.skip)
380                res.set_errormsg('probe command: test skipped.')
381                pm.call_pre_case(tidx, test_skip=True)
382                pm.call_post_execute()
383                return res
384
385    # populate NAMES with TESTID for this test
386    NAMES['TESTID'] = tidx['id']
387
388    pm.call_pre_case(tidx)
389    prepare_env(args, pm, 'setup', "-----> prepare stage", tidx["setup"])
390
391    if (args.verbose > 0):
392        print('-----> execute stage')
393    pm.call_pre_execute()
394    (p, procout) = exec_cmd(args, pm, 'execute', tidx["cmdUnderTest"])
395    if p:
396        exit_code = p.returncode
397    else:
398        exit_code = None
399
400    pm.call_post_execute()
401
402    if (exit_code is None or exit_code != int(tidx["expExitCode"])):
403        print("exit: {!r}".format(exit_code))
404        print("exit: {}".format(int(tidx["expExitCode"])))
405        #print("exit: {!r} {}".format(exit_code, int(tidx["expExitCode"])))
406        res.set_result(ResultState.fail)
407        res.set_failmsg('Command exited with {}, expected {}\n{}'.format(exit_code, tidx["expExitCode"], procout))
408        print(procout)
409    else:
410        if args.verbose > 0:
411            print('-----> verify stage')
412        (p, procout) = exec_cmd(args, pm, 'verify', tidx["verifyCmd"])
413        if procout:
414            if 'matchJSON' in tidx:
415                verify_by_json(procout, res, tidx, args, pm)
416            elif 'matchPattern' in tidx:
417                match_pattern = re.compile(
418                    str(tidx["matchPattern"]), re.DOTALL | re.MULTILINE)
419                match_index = re.findall(match_pattern, procout)
420                if len(match_index) != int(tidx["matchCount"]):
421                    res.set_result(ResultState.fail)
422                    res.set_failmsg('Could not match regex pattern. Verify command output:\n{}'.format(procout))
423                else:
424                    res.set_result(ResultState.success)
425            else:
426                res.set_result(ResultState.fail)
427                res.set_failmsg('Must specify a match option: matchJSON or matchPattern\n{}'.format(procout))
428        elif int(tidx["matchCount"]) != 0:
429            res.set_result(ResultState.fail)
430            res.set_failmsg('No output generated by verify command.')
431        else:
432            res.set_result(ResultState.success)
433
434    prepare_env(args, pm, 'teardown', '-----> teardown stage', tidx['teardown'], procout)
435    pm.call_post_case()
436
437    index += 1
438
439    # remove TESTID from NAMES
440    del(NAMES['TESTID'])
441    return res
442
443def test_runner(pm, args, filtered_tests):
444    """
445    Driver function for the unit tests.
446
447    Prints information about the tests being run, executes the setup and
448    teardown commands and the command under test itself. Also determines
449    success/failure based on the information in the test case and generates
450    TAP output accordingly.
451    """
452    testlist = filtered_tests
453    tcount = len(testlist)
454    index = 1
455    tap = ''
456    badtest = None
457    stage = None
458    emergency_exit = False
459    emergency_exit_message = ''
460
461    tsr = TestSuiteReport()
462
463    try:
464        pm.call_pre_suite(tcount, [tidx['id'] for tidx in testlist])
465    except Exception as ee:
466        ex_type, ex, ex_tb = sys.exc_info()
467        print('Exception {} {} (caught in pre_suite).'.
468              format(ex_type, ex))
469        traceback.print_tb(ex_tb)
470        emergency_exit_message = 'EMERGENCY EXIT, call_pre_suite failed with exception {} {}\n'.format(ex_type, ex)
471        emergency_exit = True
472        stage = 'pre-SUITE'
473
474    if emergency_exit:
475        pm.call_post_suite(index)
476        return emergency_exit_message
477    if args.verbose > 1:
478        print('give test rig 2 seconds to stabilize')
479    time.sleep(2)
480    for tidx in testlist:
481        if "flower" in tidx["category"] and args.device == None:
482            errmsg = "Tests using the DEV2 variable must define the name of a "
483            errmsg += "physical NIC with the -d option when running tdc.\n"
484            errmsg += "Test has been skipped."
485            if args.verbose > 1:
486                print(errmsg)
487            res = TestResult(tidx['id'], tidx['name'])
488            res.set_result(ResultState.skip)
489            res.set_errormsg(errmsg)
490            tsr.add_resultdata(res)
491            index += 1
492            continue
493        try:
494            badtest = tidx  # in case it goes bad
495            res = run_one_test(pm, args, index, tidx)
496            tsr.add_resultdata(res)
497        except PluginMgrTestFail as pmtf:
498            ex_type, ex, ex_tb = sys.exc_info()
499            stage = pmtf.stage
500            message = pmtf.message
501            output = pmtf.output
502            res = TestResult(tidx['id'], tidx['name'])
503            res.set_result(ResultState.skip)
504            res.set_errormsg(pmtf.message)
505            res.set_failmsg(pmtf.output)
506            tsr.add_resultdata(res)
507            index += 1
508            print(message)
509            print('Exception {} {} (caught in test_runner, running test {} {} {} stage {})'.
510                  format(ex_type, ex, index, tidx['id'], tidx['name'], stage))
511            print('---------------')
512            print('traceback')
513            traceback.print_tb(ex_tb)
514            print('---------------')
515            if stage == 'teardown':
516                print('accumulated output for this test:')
517                if pmtf.output:
518                    print(pmtf.output)
519            print('---------------')
520            break
521        index += 1
522
523    # if we failed in setup or teardown,
524    # fill in the remaining tests with ok-skipped
525    count = index
526
527    if tcount + 1 != count:
528        for tidx in testlist[count - 1:]:
529            res = TestResult(tidx['id'], tidx['name'])
530            res.set_result(ResultState.skip)
531            msg = 'skipped - previous {} failed {} {}'.format(stage,
532                index, badtest.get('id', '--Unknown--'))
533            res.set_errormsg(msg)
534            tsr.add_resultdata(res)
535            count += 1
536
537    if args.pause:
538        print('Want to pause\nPress enter to continue ...')
539        if input(sys.stdin):
540            print('got something on stdin')
541
542    pm.call_post_suite(index)
543
544    return tsr
545
546def has_blank_ids(idlist):
547    """
548    Search the list for empty ID fields and return true/false accordingly.
549    """
550    return not(all(k for k in idlist))
551
552
553def load_from_file(filename):
554    """
555    Open the JSON file containing the test cases and return them
556    as list of ordered dictionary objects.
557    """
558    try:
559        with open(filename) as test_data:
560            testlist = json.load(test_data, object_pairs_hook=OrderedDict)
561    except json.JSONDecodeError as jde:
562        print('IGNORING test case file {}\n\tBECAUSE:  {}'.format(filename, jde))
563        testlist = list()
564    else:
565        idlist = get_id_list(testlist)
566        if (has_blank_ids(idlist)):
567            for k in testlist:
568                k['filename'] = filename
569    return testlist
570
571
572def args_parse():
573    """
574    Create the argument parser.
575    """
576    parser = argparse.ArgumentParser(description='Linux TC unit tests')
577    return parser
578
579
580def set_args(parser):
581    """
582    Set the command line arguments for tdc.
583    """
584    parser.add_argument(
585        '--outfile', type=str,
586        help='Path to the file in which results should be saved. ' +
587        'Default target is the current directory.')
588    parser.add_argument(
589        '-p', '--path', type=str,
590        help='The full path to the tc executable to use')
591    sg = parser.add_argument_group(
592        'selection', 'select which test cases: ' +
593        'files plus directories; filtered by categories plus testids')
594    ag = parser.add_argument_group(
595        'action', 'select action to perform on selected test cases')
596
597    sg.add_argument(
598        '-D', '--directory', nargs='+', metavar='DIR',
599        help='Collect tests from the specified directory(ies) ' +
600        '(default [tc-tests])')
601    sg.add_argument(
602        '-f', '--file', nargs='+', metavar='FILE',
603        help='Run tests from the specified file(s)')
604    sg.add_argument(
605        '-c', '--category', nargs='*', metavar='CATG', default=['+c'],
606        help='Run tests only from the specified category/ies, ' +
607        'or if no category/ies is/are specified, list known categories.')
608    sg.add_argument(
609        '-e', '--execute', nargs='+', metavar='ID',
610        help='Execute the specified test cases with specified IDs')
611    ag.add_argument(
612        '-l', '--list', action='store_true',
613        help='List all test cases, or those only within the specified category')
614    ag.add_argument(
615        '-s', '--show', action='store_true', dest='showID',
616        help='Display the selected test cases')
617    ag.add_argument(
618        '-i', '--id', action='store_true', dest='gen_id',
619        help='Generate ID numbers for new test cases')
620    parser.add_argument(
621        '-v', '--verbose', action='count', default=0,
622        help='Show the commands that are being run')
623    parser.add_argument(
624        '--format', default='tap', const='tap', nargs='?',
625        choices=['none', 'xunit', 'tap'],
626        help='Specify the format for test results. (Default: TAP)')
627    parser.add_argument('-d', '--device',
628                        help='Execute test cases that use a physical device, ' +
629                        'where DEVICE is its name. (If not defined, tests ' +
630                        'that require a physical device will be skipped)')
631    parser.add_argument(
632        '-P', '--pause', action='store_true',
633        help='Pause execution just before post-suite stage')
634    return parser
635
636
637def check_default_settings(args, remaining, pm):
638    """
639    Process any arguments overriding the default settings,
640    and ensure the settings are correct.
641    """
642    # Allow for overriding specific settings
643    global NAMES
644
645    if args.path != None:
646        NAMES['TC'] = args.path
647    if args.device != None:
648        NAMES['DEV2'] = args.device
649    if 'TIMEOUT' not in NAMES:
650        NAMES['TIMEOUT'] = None
651    if not os.path.isfile(NAMES['TC']):
652        print("The specified tc path " + NAMES['TC'] + " does not exist.")
653        exit(1)
654
655    pm.call_check_args(args, remaining)
656
657
658def get_id_list(alltests):
659    """
660    Generate a list of all IDs in the test cases.
661    """
662    return [x["id"] for x in alltests]
663
664
665def check_case_id(alltests):
666    """
667    Check for duplicate test case IDs.
668    """
669    idl = get_id_list(alltests)
670    return [x for x in idl if idl.count(x) > 1]
671
672
673def does_id_exist(alltests, newid):
674    """
675    Check if a given ID already exists in the list of test cases.
676    """
677    idl = get_id_list(alltests)
678    return (any(newid == x for x in idl))
679
680
681def generate_case_ids(alltests):
682    """
683    If a test case has a blank ID field, generate a random hex ID for it
684    and then write the test cases back to disk.
685    """
686    import random
687    for c in alltests:
688        if (c["id"] == ""):
689            while True:
690                newid = str('{:04x}'.format(random.randrange(16**4)))
691                if (does_id_exist(alltests, newid)):
692                    continue
693                else:
694                    c['id'] = newid
695                    break
696
697    ufilename = []
698    for c in alltests:
699        if ('filename' in c):
700            ufilename.append(c['filename'])
701    ufilename = get_unique_item(ufilename)
702    for f in ufilename:
703        testlist = []
704        for t in alltests:
705            if 'filename' in t:
706                if t['filename'] == f:
707                    del t['filename']
708                    testlist.append(t)
709        outfile = open(f, "w")
710        json.dump(testlist, outfile, indent=4)
711        outfile.write("\n")
712        outfile.close()
713
714def filter_tests_by_id(args, testlist):
715    '''
716    Remove tests from testlist that are not in the named id list.
717    If id list is empty, return empty list.
718    '''
719    newlist = list()
720    if testlist and args.execute:
721        target_ids = args.execute
722
723        if isinstance(target_ids, list) and (len(target_ids) > 0):
724            newlist = list(filter(lambda x: x['id'] in target_ids, testlist))
725    return newlist
726
727def filter_tests_by_category(args, testlist):
728    '''
729    Remove tests from testlist that are not in a named category.
730    '''
731    answer = list()
732    if args.category and testlist:
733        test_ids = list()
734        for catg in set(args.category):
735            if catg == '+c':
736                continue
737            print('considering category {}'.format(catg))
738            for tc in testlist:
739                if catg in tc['category'] and tc['id'] not in test_ids:
740                    answer.append(tc)
741                    test_ids.append(tc['id'])
742
743    return answer
744
745
746def get_test_cases(args):
747    """
748    If a test case file is specified, retrieve tests from that file.
749    Otherwise, glob for all json files in subdirectories and load from
750    each one.
751    Also, if requested, filter by category, and add tests matching
752    certain ids.
753    """
754    import fnmatch
755
756    flist = []
757    testdirs = ['tc-tests']
758
759    if args.file:
760        # at least one file was specified - remove the default directory
761        testdirs = []
762
763        for ff in args.file:
764            if not os.path.isfile(ff):
765                print("IGNORING file " + ff + "\n\tBECAUSE does not exist.")
766            else:
767                flist.append(os.path.abspath(ff))
768
769    if args.directory:
770        testdirs = args.directory
771
772    for testdir in testdirs:
773        for root, dirnames, filenames in os.walk(testdir):
774            for filename in fnmatch.filter(filenames, '*.json'):
775                candidate = os.path.abspath(os.path.join(root, filename))
776                if candidate not in testdirs:
777                    flist.append(candidate)
778
779    alltestcases = list()
780    for casefile in flist:
781        alltestcases = alltestcases + (load_from_file(casefile))
782
783    allcatlist = get_test_categories(alltestcases)
784    allidlist = get_id_list(alltestcases)
785
786    testcases_by_cats = get_categorized_testlist(alltestcases, allcatlist)
787    idtestcases = filter_tests_by_id(args, alltestcases)
788    cattestcases = filter_tests_by_category(args, alltestcases)
789
790    cat_ids = [x['id'] for x in cattestcases]
791    if args.execute:
792        if args.category:
793            alltestcases = cattestcases + [x for x in idtestcases if x['id'] not in cat_ids]
794        else:
795            alltestcases = idtestcases
796    else:
797        if cat_ids:
798            alltestcases = cattestcases
799        else:
800            # just accept the existing value of alltestcases,
801            # which has been filtered by file/directory
802            pass
803
804    return allcatlist, allidlist, testcases_by_cats, alltestcases
805
806
807def set_operation_mode(pm, parser, args, remaining):
808    """
809    Load the test case data and process remaining arguments to determine
810    what the script should do for this run, and call the appropriate
811    function.
812    """
813    ucat, idlist, testcases, alltests = get_test_cases(args)
814
815    if args.gen_id:
816        if (has_blank_ids(idlist)):
817            alltests = generate_case_ids(alltests)
818        else:
819            print("No empty ID fields found in test files.")
820        exit(0)
821
822    duplicate_ids = check_case_id(alltests)
823    if (len(duplicate_ids) > 0):
824        print("The following test case IDs are not unique:")
825        print(str(set(duplicate_ids)))
826        print("Please correct them before continuing.")
827        exit(1)
828
829    if args.showID:
830        for atest in alltests:
831            print_test_case(atest)
832        exit(0)
833
834    if isinstance(args.category, list) and (len(args.category) == 0):
835        print("Available categories:")
836        print_sll(ucat)
837        exit(0)
838
839    if args.list:
840        list_test_cases(alltests)
841        exit(0)
842
843    exit_code = 0 # KSFT_PASS
844    if len(alltests):
845        req_plugins = pm.get_required_plugins(alltests)
846        try:
847            args = pm.load_required_plugins(req_plugins, parser, args, remaining)
848        except PluginDependencyException as pde:
849            print('The following plugins were not found:')
850            print('{}'.format(pde.missing_pg))
851        catresults = test_runner(pm, args, alltests)
852        if catresults.count_failures() != 0:
853            exit_code = 1 # KSFT_FAIL
854        if args.format == 'none':
855            print('Test results output suppression requested\n')
856        else:
857            print('\nAll test results: \n')
858            if args.format == 'xunit':
859                suffix = 'xml'
860                res = catresults.format_xunit()
861            elif args.format == 'tap':
862                suffix = 'tap'
863                res = catresults.format_tap()
864            print(res)
865            print('\n\n')
866            if not args.outfile:
867                fname = 'test-results.{}'.format(suffix)
868            else:
869                fname = args.outfile
870            with open(fname, 'w') as fh:
871                fh.write(res)
872                fh.close()
873                if os.getenv('SUDO_UID') is not None:
874                    os.chown(fname, uid=int(os.getenv('SUDO_UID')),
875                        gid=int(os.getenv('SUDO_GID')))
876    else:
877        print('No tests found\n')
878        exit_code = 4 # KSFT_SKIP
879    exit(exit_code)
880
881def main():
882    """
883    Start of execution; set up argument parser and get the arguments,
884    and start operations.
885    """
886    parser = args_parse()
887    parser = set_args(parser)
888    pm = PluginMgr(parser)
889    parser = pm.call_add_args(parser)
890    (args, remaining) = parser.parse_known_args()
891    args.NAMES = NAMES
892    pm.set_args(args)
893    check_default_settings(args, remaining, pm)
894    if args.verbose > 2:
895        print('args is {}'.format(args))
896
897    set_operation_mode(pm, parser, args, remaining)
898
899if __name__ == "__main__":
900    main()
901