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