#! /usr/bin/env python3 # # Copyright (c) 2009 University of Washington # # SPDX-License-Identifier: GPL-2.0-only # import argparse import fnmatch import os import queue import re import shutil import signal import subprocess import sys import threading import time import xml.etree.ElementTree as ET from utils import get_list_from_file # Global variable args = None # imported from waflib Logs colors_lst = { "USE": True, "BOLD": "\x1b[01;1m", "RED": "\x1b[01;31m", "GREEN": "\x1b[32m", "YELLOW": "\x1b[33m", "PINK": "\x1b[35m", "BLUE": "\x1b[01;34m", "CYAN": "\x1b[36m", "GREY": "\x1b[37m", "NORMAL": "\x1b[0m", "cursor_on": "\x1b[?25h", "cursor_off": "\x1b[?25l", } def get_color(cl): if colors_lst["USE"]: return colors_lst.get(cl, "") return "" class color_dict(object): def __getattr__(self, a): return get_color(a) def __call__(self, a): return get_color(a) colors = color_dict() # # XXX This should really be part of a ns3 command to list the configuration # items relative to optional ns-3 pieces. # # A list of interesting configuration items in the ns3 configuration # cache which we may be interested in when deciding on which examples # to run and how to run them. These are set by ns3 during the # configuration phase and the corresponding assignments are usually # found in the associated subdirectory CMakeLists.txt files. # interesting_config_items = [ "NS3_ENABLED_MODULES", "NS3_ENABLED_CONTRIBUTED_MODULES", "NS3_MODULE_PATH", "ENABLE_EXAMPLES", "ENABLE_TESTS", "EXAMPLE_DIRECTORIES", "ENABLE_PYTHON_BINDINGS", "NSCLICK", "ENABLE_BRITE", "ENABLE_OPENFLOW", "APPNAME", "BUILD_PROFILE", "VERSION", "PYTHON", "VALGRIND_FOUND", ] ENABLE_EXAMPLES = True ENABLE_TESTS = True NSCLICK = False ENABLE_BRITE = False ENABLE_OPENFLOW = False ENABLE_PYTHON_BINDINGS = False EXAMPLE_DIRECTORIES = [] APPNAME = "" BUILD_PROFILE = "" BUILD_PROFILE_SUFFIX = "" VERSION = "" PYTHON = "" VALGRIND_FOUND = True # # This will be given a prefix and a suffix when the ns3 config file is # read. # test_runner_name = "test-runner" # # If the user has constrained us to run certain kinds of tests, we can tell ns3 # to only build # core_kinds = ["core", "performance", "system", "unit"] # # Exclude tests that are problematic for valgrind. # core_valgrind_skip_tests = [ "routing-click", "lte-rr-ff-mac-scheduler", "lte-tdmt-ff-mac-scheduler", "lte-fdmt-ff-mac-scheduler", "lte-pf-ff-mac-scheduler", "lte-tta-ff-mac-scheduler", "lte-fdbet-ff-mac-scheduler", "lte-ttbet-ff-mac-scheduler", "lte-fdtbfq-ff-mac-scheduler", "lte-tdtbfq-ff-mac-scheduler", "lte-pss-ff-mac-scheduler", ] # # Parse the examples-to-run file if it exists. # # This function adds any C++ examples or Python examples that are to be run # to the lists in example_tests and python_tests, respectively. # def parse_examples_to_run_file( examples_to_run_path, cpp_executable_dir, python_script_dir, example_tests, example_names_original, python_tests, ): # Look for the examples-to-run file exists. if not os.path.exists(examples_to_run_path): # Also tests for contribs OUTSIDE the ns-3-dev directory possible_external_contrib_path = examples_to_run_path.replace( "contrib", f"{os.path.dirname(os.path.dirname(__file__))}/ns-3-external-contrib" ) if os.path.exists(possible_external_contrib_path): examples_to_run_path = possible_external_contrib_path else: return # Each tuple in the C++ list of examples to run contains # # (example_name, do_run, do_valgrind_run) # # where example_name is the executable to be run, do_run is a # condition under which to run the example, and do_valgrind_run is # a condition under which to run the example under valgrind. This # is needed because NSC causes illegal instruction crashes with # some tests when they are run under valgrind. # # Note that the two conditions are Python statements that # can depend on ns3 configuration variables. For example, # when NSC was in the codebase, we could write: # # ("tcp-nsc-lfn", "NSC_ENABLED == True", "NSC_ENABLED == False", "QUICK"), # cpp_examples = get_list_from_file(examples_to_run_path, "cpp_examples") for cpp_example in cpp_examples: # Old example specification did not include # 'fullness', so for compatibility, # allow 3 components, & set the 'fullness' to QUICK if len(cpp_example) == 3: example_name, do_run, do_valgrind_run = cpp_example fullness = "QUICK" elif len(cpp_example) == 4: example_name, do_run, do_valgrind_run, fullness = cpp_example fullness: str = fullness.upper() if fullness != "QUICK" and fullness != "EXTENSIVE" and fullness != "TAKES_FOREVER": raise ValueError( f"Invalid value provided for example '{example_name}' " + f"expected 'QUICK', 'EXTENSIVE', or 'TAKES_FOREVER', got: '{fullness}'" ) else: # If we have the name of the example we're error-ing for, provide it # Otherwise, just give a generic message if len(cpp_example) >= 1: raise RuntimeError( f"Incorrect number of fields declaration of example '{cpp_example[0]}', " + f"expected 3, or 4 got: {len(cpp_example)}" ) else: raise RuntimeError( f"Incorrect number of fields declaration of example, " + f"expected 3, or 4 got: {len(cpp_example)}" ) # Separate the example name from its arguments. example_name_original = example_name example_name_parts = example_name.split(" ", 1) if len(example_name_parts) == 1: example_name = example_name_parts[0] example_arguments = "" else: example_name = example_name_parts[0] example_arguments = example_name_parts[1] # Add the proper prefix and suffix to the example name to # match what is done in the CMakeLists.txt file. example_path = "%s%s-%s%s" % (APPNAME, VERSION, example_name, BUILD_PROFILE_SUFFIX) # Set the full path for the example. example_path = os.path.join(cpp_executable_dir, example_path) example_path += ".exe" if sys.platform == "win32" else "" example_name = os.path.join(os.path.relpath(cpp_executable_dir, NS3_BUILDDIR), example_name) # Add all of the C++ examples that were built, i.e. found # in the directory, to the list of C++ examples to run. if os.path.exists(example_path): # Handle paths with empty spaces example_path = '"%s"' % example_path # Add any arguments to the path. if len(example_name_parts) != 1: example_path = "%s %s" % (example_path, example_arguments) example_name = "%s %s" % (example_name, example_arguments) # Add this example. example_tests.append((example_name, example_path, do_run, do_valgrind_run, fullness)) example_names_original.append(example_name_original) # Each tuple in the Python list of examples to run contains # # (example_name, do_run, fullness) # # where example_name is the Python script to be run and # do_run is a condition under which to run the example. # # Note that the condition is a Python statement that can # depend on ns3 configuration variables. For example, # # ("brite-generic-example.py", "ENABLE_BRITE == True", "QUICK"), # python_examples = get_list_from_file(examples_to_run_path, "python_examples") # Old example specification did not include # 'fullness', so for compatibility, # allow 2 components, & set the 'fullness' to QUICK for python_example in python_examples: if len(python_example) == 2: example_name, do_run = python_example fullness = "QUICK" elif len(python_example) == 3: example_name, do_run, fullness = python_example else: # If we have the name of the example we're error-ing for, provide it # Otherwise, just give a generic message if len(python_example) >= 1: raise RuntimeError( f"Incorrect number of fields declaration of example '{python_example[0]}', " + f"expected 2, or 3 got: {len(python_example)}" ) else: raise RuntimeError( f"Incorrect number of fields declaration of example, " + f"expected 2, or 3 got: {len(python_example)}" ) # Separate the example name from its arguments. example_name_parts = example_name.split(" ", 1) if len(example_name_parts) == 1: example_name = example_name_parts[0] example_arguments = "" else: example_name = example_name_parts[0] example_arguments = example_name_parts[1] # Set the full path for the example. example_path = os.path.join(python_script_dir, example_name) # Add all of the Python examples that were found to the # list of Python examples to run. if os.path.exists(example_path): # Add any arguments to the path. if len(example_name_parts) != 1: example_path = "%s %s" % (example_path, example_arguments) # Add this example. python_tests.append((example_path, do_run, fullness)) # # The test suites are going to want to output status. They are running # concurrently. This means that unless we are careful, the output of # the test suites will be interleaved. Rather than introducing a lock # file that could unintentionally start serializing execution, we ask # the tests to write their output to a temporary directory and then # put together the final output file when we "join" the test tasks back # to the main thread. In addition to this issue, the example programs # often write lots and lots of trace files which we will just ignore. # We put all of them into the temp directory as well, so they can be # easily deleted. # TMP_OUTPUT_DIR = "testpy-output" def read_test(test): result = test.find("Result").text name = test.find("Name").text if not test.find("Reason") is None: reason = test.find("Reason").text else: reason = "" if not test.find("Time") is None: time_real = test.find("Time").get("real") else: time_real = "" return (result, name, reason, time_real) # # A simple example of writing a text file with a test result summary. It is # expected that this output will be fine for developers looking for problems. # def node_to_text(test, f, test_type="Suite"): result, name, reason, time_real = read_test(test) if reason: reason = " (%s)" % reason output = '%s: Test %s "%s" (%s)%s\n' % (result, test_type, name, time_real, reason) f.write(output) for details in test.findall("FailureDetails"): f.write(" Details:\n") f.write(" Message: %s\n" % details.find("Message").text) f.write(" Condition: %s\n" % details.find("Condition").text) f.write(" Actual: %s\n" % details.find("Actual").text) f.write(" Limit: %s\n" % details.find("Limit").text) f.write(" File: %s\n" % details.find("File").text) f.write(" Line: %s\n" % details.find("Line").text) for child in test.findall("Test"): node_to_text(child, f, "Case") def translate_to_text(results_file, text_file): text_file += ".txt" if ".txt" not in text_file else "" print('Writing results to text file "%s"...' % text_file, end="") et = ET.parse(results_file) with open(text_file, "w", encoding="utf-8") as f: for test in et.findall("Test"): node_to_text(test, f) for example in et.findall("Example"): result = example.find("Result").text name = example.find("Name").text if not example.find("Time") is None: time_real = example.find("Time").get("real") else: time_real = "" output = '%s: Example "%s" (%s)\n' % (result, name, time_real) f.write(output) print("done.") # # A simple example of writing an HTML file with a test result summary. It is # expected that this will eventually be made prettier as time progresses and # we have time to tweak it. This may end up being moved to a separate module # since it will probably grow over time. # def translate_to_html(results_file, html_file): html_file += ".html" if ".html" not in html_file else "" print("Writing results to html file %s..." % html_file, end="") with open(html_file, "w", encoding="utf-8") as f: f.write("\n") f.write("\n") f.write("

ns-3 Test Results

\n") # # Read and parse the whole results file. # et = ET.parse(results_file) # # Iterate through the test suites # f.write("

Test Suites

\n") for suite in et.findall("Test"): # # For each test suite, get its name, result and execution time info # result, name, reason, time = read_test(suite) # # Print a level three header with the result, name and time. If the # test suite passed, the header is printed in green. If the suite was # skipped, print it in orange, otherwise assume something bad happened # and print in red. # if result == "PASS": f.write('

%s: %s (%s)

\n' % (result, name, time)) elif result == "SKIP": f.write( '

%s: %s (%s) (%s)

\n' % (result, name, time, reason) ) else: f.write('

%s: %s (%s)

\n' % (result, name, time)) # # The test case information goes in a table. # f.write('\n') # # The first column of the table has the heading Result # f.write("\n") # # If the suite crashed or is skipped, there is no further information, so just # declare a new table row with the result (CRASH or SKIP) in it. Looks like: # # +--------+ # | Result | # +--------+ # | CRASH | # +--------+ # # Then go on to the next test suite. Valgrind and skipped errors look the same. # if result in ["CRASH", "SKIP", "VALGR"]: f.write("\n") if result == "SKIP": f.write('\n' % result) else: f.write('\n' % result) f.write("\n") f.write("
Result
%s%s
\n") continue # # If the suite didn't crash, we expect more information, so fill out # the table heading row. Like, # # +--------+----------------+------+ # | Result | Test Case Name | Time | # +--------+----------------+------+ # f.write("Test Case Name\n") f.write(" Time \n") # # If the test case failed, we need to print out some failure details # so extend the heading row again. Like, # # +--------+----------------+------+-----------------+ # | Result | Test Case Name | Time | Failure Details | # +--------+----------------+------+-----------------+ # if result == "FAIL": f.write("Failure Details\n") # # Now iterate through all the test cases. # for case in suite.findall("Test"): # # Get the name, result and timing information from xml to use in # printing table below. # result, name, reason, time = read_test(case) # # If the test case failed, we iterate through possibly multiple # failure details # if result == "FAIL": # # There can be multiple failures for each test case. The first # row always gets the result, name and timing information along # with the failure details. Remaining failures don't duplicate # this information but just get blanks for readability. Like, # # +--------+----------------+------+-----------------+ # | Result | Test Case Name | Time | Failure Details | # +--------+----------------+------+-----------------+ # | FAIL | The name | time | It's busted | # +--------+----------------+------+-----------------+ # | | | | Really broken | # +--------+----------------+------+-----------------+ # | | | | Busted bad | # +--------+----------------+------+-----------------+ # first_row = True for details in case.findall("FailureDetails"): # # Start a new row in the table for each possible Failure Detail # f.write("\n") if first_row: first_row = False f.write('%s\n' % result) f.write("%s\n" % name) f.write("%s\n" % time) else: f.write("\n") f.write("\n") f.write("\n") f.write("") f.write("Message: %s, " % details.find("Message").text) f.write("Condition: %s, " % details.find("Condition").text) f.write("Actual: %s, " % details.find("Actual").text) f.write("Limit: %s, " % details.find("Limit").text) f.write("File: %s, " % details.find("File").text) f.write("Line: %s" % details.find("Line").text) f.write("\n") # # End the table row # f.write("\n") else: # # If this particular test case passed, then we just print the PASS # result in green, followed by the test case name and its execution # time information. These go off in ... table data. # The details table entry is left blank. # # +--------+----------------+------+---------+ # | Result | Test Case Name | Time | Details | # +--------+----------------+------+---------+ # | PASS | The name | time | | # +--------+----------------+------+---------+ # f.write("\n") f.write('%s\n' % result) f.write("%s\n" % name) f.write("%s\n" % time) f.write("%s\n" % reason) f.write("\n") # # All of the rows are written, so we need to end the table. # f.write("\n") # # That's it for all of the test suites. Now we have to do something about # our examples. # f.write("

Examples

\n") # # Example status is rendered in a table just like the suites. # f.write('\n') # # The table headings look like, # # +--------+--------------+--------------+---------+ # | Result | Example Name | Elapsed Time | Details | # +--------+--------------+--------------+---------+ # f.write("\n") f.write("\n") f.write("\n") f.write("\n") # # Now iterate through all the examples # for example in et.findall("Example"): # # Start a new row for each example # f.write("\n") # # Get the result and name of the example in question # result, name, reason, time = read_test(example) # # If the example either failed or crashed, print its result status # in red; otherwise green. This goes in a table data # if result == "PASS": f.write('\n' % result) elif result == "SKIP": f.write('\n' % result) # # Write the example name as a new tag data. # f.write("\n" % name) # # Write the elapsed time as a new tag data. # f.write("\n" % time) # # Write the reason, if it exist # f.write("\n" % reason) # # That's it for the current example, so terminate the row. # f.write("\n") # # That's it for the table of examples, so terminate the table. # f.write("
Result Example NameElapsed TimeDetails
... %s%s\n' % result) else: f.write('%s%s%s%s
\n") # # And that's it for the report, so finish up. # f.write("\n") f.write("\n") print("done.") # # Python Control-C handling is broken in the presence of multiple threads. # Signals get delivered to the runnable/running thread by default and if # it is blocked, the signal is simply ignored. So we hook sigint and set # a global variable telling the system to shut down gracefully. # thread_exit = False def sigint_hook(signal, frame): global thread_exit thread_exit = True return 0 # # In general, the build process itself naturally takes care of figuring out # which tests are built into the test runner. For example, if ns3 configure # determines that ENABLE_EMU is false due to some missing dependency, # the tests for the emu net device simply will not be built and will # therefore not be included in the built test runner. # # Examples, however, are a different story. In that case, we are just given # a list of examples that could be run. Instead of just failing, for example, # an example if its library support is not present, we look into the ns3 # saved configuration for relevant configuration items. # # XXX This function pokes around in the ns3 internal state file. To be a # little less hacky, we should add a command to ns3 to return this info # and use that result. # platform = sys.platform platform = "bsd" if "bsd" in platform else platform lock_filename = ".lock-ns3_%s_build" % platform def read_ns3_config(): try: # sys.platform reports linux2 for python2 and linux for python3 with open(lock_filename, "rt", encoding="utf-8") as f: for line in f: if line.startswith("top_dir ="): key, val = line.split("=") top_dir = eval(val.strip()) if line.startswith("out_dir ="): key, val = line.split("=") out_dir = eval(val.strip()) except FileNotFoundError: print( "The .lock-ns3 file was not found. You must configure before running test.py.", file=sys.stderr, ) sys.exit(2) global NS3_BASEDIR NS3_BASEDIR = top_dir global NS3_BUILDDIR NS3_BUILDDIR = out_dir with open(lock_filename, encoding="utf-8") as f: for line in f.readlines(): for item in interesting_config_items: if line.startswith(item): exec(line, globals()) if args.verbose: for item in interesting_config_items: print("%s ==" % item, eval(item)) # # It seems pointless to fork a process to run ns3 to fork a process to run # the test runner, so we just run the test runner directly. The main thing # that ns3 would do for us would be to sort out the shared library path but # we can deal with that easily and do here. # # There can be many different ns-3 repositories on a system, and each has # its own shared libraries, so ns-3 doesn't hardcode a shared library search # path -- it is cooked up dynamically, so we do that too. # def make_paths(): have_DYLD_LIBRARY_PATH = False have_LD_LIBRARY_PATH = False have_PATH = False have_PYTHONPATH = False keys = list(os.environ.keys()) for key in keys: if key == "DYLD_LIBRARY_PATH": have_DYLD_LIBRARY_PATH = True if key == "LD_LIBRARY_PATH": have_LD_LIBRARY_PATH = True if key == "PATH": have_PATH = True if key == "PYTHONPATH": have_PYTHONPATH = True pypath = os.environ["PYTHONPATH"] = os.path.join(NS3_BUILDDIR, "bindings", "python") if not have_PYTHONPATH: os.environ["PYTHONPATH"] = pypath else: os.environ["PYTHONPATH"] += ":" + pypath if args.verbose: print('os.environ["PYTHONPATH"] == %s' % os.environ["PYTHONPATH"]) if sys.platform == "darwin": if not have_DYLD_LIBRARY_PATH: os.environ["DYLD_LIBRARY_PATH"] = "" for path in NS3_MODULE_PATH: os.environ["DYLD_LIBRARY_PATH"] += ":" + path if args.verbose: print('os.environ["DYLD_LIBRARY_PATH"] == %s' % os.environ["DYLD_LIBRARY_PATH"]) elif sys.platform == "win32": if not have_PATH: os.environ["PATH"] = "" for path in NS3_MODULE_PATH: os.environ["PATH"] += ";" + path if args.verbose: print('os.environ["PATH"] == %s' % os.environ["PATH"]) elif sys.platform == "cygwin": if not have_PATH: os.environ["PATH"] = "" for path in NS3_MODULE_PATH: os.environ["PATH"] += ":" + path if args.verbose: print('os.environ["PATH"] == %s' % os.environ["PATH"]) else: if not have_LD_LIBRARY_PATH: os.environ["LD_LIBRARY_PATH"] = "" for path in NS3_MODULE_PATH: os.environ["LD_LIBRARY_PATH"] += ":" + str(path) if args.verbose: print('os.environ["LD_LIBRARY_PATH"] == %s' % os.environ["LD_LIBRARY_PATH"]) # # Short note on generating suppressions: # # See the valgrind documentation for a description of suppressions. The easiest # way to generate a suppression expression is by using the valgrind # --gen-suppressions option. To do that you have to figure out how to run the # test in question. # # If you do "test.py -v -g -s then test.py will output most of what # you need. For example, if you are getting a valgrind error in the # devices-mesh-dot11s-regression test suite, you can run: # # ./test.py -v -g -s devices-mesh-dot11s-regression # # You should see in the verbose output something that looks like: # # Synchronously execute valgrind --suppressions=/home/craigdo/repos/ns-3-allinone-dev/ns-3-dev/testpy.supp # --leak-check=full --error-exitcode=2 /home/craigdo/repos/ns-3-allinone-dev/ns-3-dev/build/debug/utils/ns3-dev-test-runner-debug # --suite=devices-mesh-dot11s-regression --basedir=/home/craigdo/repos/ns-3-allinone-dev/ns-3-dev # --tempdir=testpy-output/2010-01-12-22-47-50-CUT # --out=testpy-output/2010-01-12-22-47-50-CUT/devices-mesh-dot11s-regression.xml # # You need to pull out the useful pieces, and so could run the following to # reproduce your error: # # valgrind --suppressions=/home/craigdo/repos/ns-3-allinone-dev/ns-3-dev/testpy.supp # --leak-check=full --error-exitcode=2 /home/craigdo/repos/ns-3-allinone-dev/ns-3-dev/build/debug/utils/ns3-dev-test-runner-debug # --suite=devices-mesh-dot11s-regression --basedir=/home/craigdo/repos/ns-3-allinone-dev/ns-3-dev # --tempdir=testpy-output # # Hint: Use the first part of the command as is, and point the "tempdir" to # somewhere real. You don't need to specify an "out" file. # # When you run the above command you should see your valgrind error. The # suppression expression(s) can be generated by adding the --gen-suppressions=yes # option to valgrind. Use something like: # # valgrind --gen-suppressions=yes --suppressions=/home/craigdo/repos/ns-3-allinone-dev/ns-3-dev/testpy.supp # --leak-check=full --error-exitcode=2 /home/craigdo/repos/ns-3-allinone-dev/ns-3-dev/build/debug/utils/ns3-dev-test-runner-debug # --suite=devices-mesh-dot11s-regression --basedir=/home/craigdo/repos/ns-3-allinone-dev/ns-3-dev # --tempdir=testpy-output # # Now when valgrind detects an error it will ask: # # ==27235== ---- Print suppression ? --- [Return/N/n/Y/y/C/c] ---- # # to which you just enter 'y'. # # You will be provided with a suppression expression that looks something like # the following: # { # # Memcheck:Addr8 # fun:_ZN3ns36dot11s15HwmpProtocolMac8SendPreqESt6vectorINS0_6IePreqESaIS3_EE # fun:_ZN3ns36dot11s15HwmpProtocolMac10SendMyPreqEv # fun:_ZN3ns36dot11s15HwmpProtocolMac18RequestDestinationENS_12Mac48AddressEjj # ... # the rest of the stack frame # ... # } # # You need to add a suppression name which will only be printed out by valgrind in # verbose mode (but it needs to be there in any case). The entire stack frame is # shown to completely characterize the error, but in most cases you won't need # all of that info. For example, if you want to turn off all errors that happen # when the function (fun:) is called, you can just delete the rest of the stack # frame. You can also use wildcards to make the mangled signatures more readable. # # I added the following to the testpy.supp file for this particular error: # # { # Suppress invalid read size errors in SendPreq() when using HwmpProtocolMac # Memcheck:Addr8 # fun:*HwmpProtocolMac*SendPreq* # } # # Now, when you run valgrind the error will be suppressed. # # Until ns-3.36, we used a suppression in testpy.supp in the top-level # ns-3 directory. It was defined below, but commented out once it was # no longer needed. If it is needed again in the future, define the # below variable again, and remove the alternative definition to None # VALGRIND_SUPPRESSIONS_FILE = ".ns3.supp" # VALGRIND_SUPPRESSIONS_FILE = None # When the TEST_LOGS environment variable is set to 1 or true, # NS_LOG is set to NS_LOG=*, and stdout/stderr # from tests are discarded to prevent running out of memory. TEST_LOGS = bool(os.getenv("TEST_LOGS", False)) def run_job_synchronously(shell_command, directory, valgrind, is_python, build_path=""): if VALGRIND_SUPPRESSIONS_FILE is not None: suppressions_path = os.path.join(NS3_BASEDIR, VALGRIND_SUPPRESSIONS_FILE) if is_python: path_cmd = PYTHON[0] + " " + os.path.join(NS3_BASEDIR, shell_command) else: # Here we handle paths with empty spaces. # To do that, we need to separate program from arguments, then quote the absolute program path with spaces. shell_args = re.findall(r'(?:".*?"|\S)+', shell_command) if len(build_path): path_cmd = os.path.join(build_path, shell_args[0]) else: path_cmd = ( shell_args[0].strip('"') if NS3_BUILDDIR in shell_args[0] else os.path.join(NS3_BUILDDIR, shell_args[0]) ) path_cmd = f'"{path_cmd}" {" ".join(shell_args[1:])}' if valgrind: if VALGRIND_SUPPRESSIONS_FILE: cmd = ( 'valgrind --suppressions="%s" --leak-check=full --show-reachable=yes --error-exitcode=2 --errors-for-leak-kinds=all %s' % (suppressions_path, path_cmd) ) else: cmd = ( "valgrind --leak-check=full --show-reachable=yes --error-exitcode=2 --errors-for-leak-kinds=all %s" % (path_cmd) ) else: cmd = path_cmd if args.verbose: print("Synchronously execute %s" % cmd) start_time = time.time() proc = subprocess.Popen( cmd, shell=True, cwd=directory, stdout=subprocess.PIPE if not TEST_LOGS else subprocess.DEVNULL, stderr=subprocess.PIPE if not TEST_LOGS else subprocess.STDOUT, ) stdout_results, stderr_results = proc.communicate() stdout_results = b"" if stdout_results is None else stdout_results stderr_results = b"" if stderr_results is None else stderr_results elapsed_time = time.time() - start_time retval = proc.returncode def decode_stream_results(stream_results: bytes, stream_name: str) -> str: try: stream_results = stream_results.decode() except UnicodeDecodeError: def decode(byte_array: bytes): try: byte_array.decode() except UnicodeDecodeError: return byte_array # Find lines where the decoding error happened non_utf8_lines = list(map(lambda line: decode(line), stream_results.splitlines())) non_utf8_lines = list(filter(lambda line: line is not None, non_utf8_lines)) print( f"Non-decodable characters found in {stream_name} output of {cmd}: {non_utf8_lines}" ) # Continue decoding on errors stream_results = stream_results.decode(errors="backslashreplace") return stream_results stdout_results = decode_stream_results(stdout_results, "stdout") stderr_results = decode_stream_results(stderr_results, "stderr") if args.verbose: print("Return code = ", retval) print("stderr = ", stderr_results) return (retval, stdout_results, stderr_results, elapsed_time) # # This class defines a unit of testing work. It will typically refer to # a test suite to run using the test-runner, or an example to run directly. # class Job: def __init__(self): self.is_break = False self.is_skip = False self.skip_reason = "" self.is_example = False self.is_pyexample = False self.shell_command = "" self.display_name = "" self.basedir = "" self.tempdir = "" self.cwd = "" self.tmp_file_name = "" self.returncode = False self.elapsed_time = 0 self.build_path = "" # # A job is either a standard job or a special job indicating that a worker # thread should exist. This special job is indicated by setting is_break # to true. # def set_is_break(self, is_break): self.is_break = is_break # # If a job is to be skipped, we actually run it through the worker threads # to keep the PASS, FAIL, CRASH and SKIP processing all in one place. # def set_is_skip(self, is_skip): self.is_skip = is_skip # # If a job is to be skipped, log the reason. # def set_skip_reason(self, skip_reason): self.skip_reason = skip_reason # # Examples are treated differently than standard test suites. This is # mostly because they are completely unaware that they are being run as # tests. So we have to do some special case processing to make them look # like tests. # def set_is_example(self, is_example): self.is_example = is_example # # Examples are treated differently than standard test suites. This is # mostly because they are completely unaware that they are being run as # tests. So we have to do some special case processing to make them look # like tests. # def set_is_pyexample(self, is_pyexample): self.is_pyexample = is_pyexample # # This is the shell command that will be executed in the job. For example, # # "utils/ns3-dev-test-runner-debug --test-name=some-test-suite" # def set_shell_command(self, shell_command): self.shell_command = shell_command # # This is the build path where ns-3 was built. For example, # # "/home/craigdo/repos/ns-3-allinone-test/ns-3-dev/build/debug" # def set_build_path(self, build_path): self.build_path = build_path # # This is the display name of the job, typically the test suite or example # name. For example, # # "some-test-suite" or "udp-echo" # def set_display_name(self, display_name): self.display_name = display_name # # This is the base directory of the repository out of which the tests are # being run. It will be used deep down in the testing framework to determine # where the source directory of the test was, and therefore where to find # provided test vectors. For example, # # "/home/user/repos/ns-3-dev" # def set_basedir(self, basedir): self.basedir = basedir # # This is the directory to which a running test suite should write any # temporary files. # def set_tempdir(self, tempdir): self.tempdir = tempdir # # This is the current working directory that will be given to an executing # test as it is being run. It will be used for examples to tell them where # to write all of the pcap files that we will be carefully ignoring. For # example, # # "/tmp/unchecked-traces" # def set_cwd(self, cwd): self.cwd = cwd # # This is the temporary results file name that will be given to an executing # test as it is being run. We will be running all of our tests in parallel # so there must be multiple temporary output files. These will be collected # into a single XML file at the end and then be deleted. # def set_tmp_file_name(self, tmp_file_name): self.tmp_file_name = tmp_file_name # # The return code received when the job process is executed. # def set_returncode(self, returncode): self.returncode = returncode # # The elapsed real time for the job execution. # def set_elapsed_time(self, elapsed_time): self.elapsed_time = elapsed_time # # The worker thread class that handles the actual running of a given test. # Once spawned, it receives requests for work through its input_queue and # ships the results back through the output_queue. # class worker_thread(threading.Thread): def __init__(self, input_queue, output_queue): threading.Thread.__init__(self) self.input_queue = input_queue self.output_queue = output_queue def run(self): while True: job = self.input_queue.get() # # Worker threads continue running until explicitly told to stop with # a special job. # if job.is_break: return # # If the global interrupt handler sets the thread_exit variable, # we stop doing real work and just report back a "break" in the # normal command processing has happened. # if thread_exit == True: job.set_is_break(True) self.output_queue.put(job) continue # # If we are actually supposed to skip this job, do so. Note that # if is_skip is true, returncode is undefined. # if job.is_skip: if args.verbose: print("Skip %s" % job.shell_command) self.output_queue.put(job) continue # # Otherwise go about the business of running tests as normal. # else: if args.verbose: print("Launch %s" % job.shell_command) if job.is_example or job.is_pyexample: # # If we have an example, the shell command is all we need to # know. It will be something like "examples/udp/udp-echo" or # "examples/wireless/mixed-wireless.py" # ( job.returncode, job.standard_out, job.standard_err, et, ) = run_job_synchronously( job.shell_command, job.cwd, args.valgrind, job.is_pyexample, job.build_path ) else: # # If we're a test suite, we need to provide a little more info # to the test runner, specifically the base directory and temp # file name # if args.update_data: update_data = "--update-data" else: update_data = "" ( job.returncode, job.standard_out, job.standard_err, et, ) = run_job_synchronously( job.shell_command + " --xml --tempdir=%s --out=%s %s" % (job.tempdir, job.tmp_file_name, update_data), job.cwd, args.valgrind, False, ) job.set_elapsed_time(et) if args.verbose: print("returncode = %d" % job.returncode) print("---------- begin standard out ----------") print(job.standard_out) print("---------- begin standard err ----------") print(job.standard_err) print("---------- end standard err ----------") self.output_queue.put(job) # # This function loads the list of previously successful or skipped examples and test suites. # def load_previously_successful_tests(): import glob previously_run_tests_to_skip = {"test": [], "example": []} previous_results = glob.glob(f"{TMP_OUTPUT_DIR}/*-results.xml") if not previous_results: print("No previous runs to rerun") exit(-1) latest_result_file = list( sorted(previous_results, key=lambda x: os.path.basename(x), reverse=True) )[0] try: previous_run_results = ET.parse(latest_result_file) except ET.ParseError: print(f"Failed to parse XML {latest_result_file}") exit(-1) for test_type in ["Test", "Example"]: if previous_run_results.find(test_type) is not None: temp = list( map( lambda x: (x.find("Name").text, x.find("Result").text), previous_run_results.findall(test_type), ) ) temp = list(filter(lambda x: x[1] in ["PASS", "SKIP"], temp)) temp = [x[0] for x in temp] previously_run_tests_to_skip[test_type.lower()] = temp return previously_run_tests_to_skip # # This is the main function that does the work of interacting with the # test-runner itself. # def run_tests(): # # Pull some interesting configuration information out of ns3, primarily # so we can know where executables can be found, but also to tell us what # pieces of the system have been built. This will tell us what examples # are runnable. # read_ns3_config() # # Set the proper suffix. # global BUILD_PROFILE_SUFFIX if BUILD_PROFILE == "release": BUILD_PROFILE_SUFFIX = "" else: BUILD_PROFILE_SUFFIX = "-" + BUILD_PROFILE # # Add the proper prefix and suffix to the test-runner name to # match what is done in the CMakeLists.txt file. # test_runner_name = "%s%s-%s%s" % (APPNAME, VERSION, "test-runner", BUILD_PROFILE_SUFFIX) test_runner_name += ".exe" if sys.platform == "win32" else "" # # Run ns3 to make sure that everything is built, configured and ready to go # unless we are explicitly told not to. We want to be careful about causing # our users pain while waiting for extraneous stuff to compile and link, so # we allow users that know what they're doing to not invoke ns3 at all. # if not args.no_build: # If the user only wants to run a single example, then we can just build # that example. # # If there is no constraint, then we have to build everything since the # user wants to run everything. # if len(args.example): build_cmd = "./ns3 build %s" % os.path.basename(args.example.replace("*", "")) else: build_cmd = "./ns3" if sys.platform == "win32": build_cmd = f'"{sys.executable}" {build_cmd}' if args.verbose: print("Building: %s" % build_cmd) proc = subprocess.run(build_cmd, shell=True) if proc.returncode: print("ns3 died. Not running tests", file=sys.stderr) return proc.returncode # # Dynamically set up paths. # make_paths() # # Get the information from the build status file. # if os.path.exists(lock_filename): ns3_runnable_programs = get_list_from_file(lock_filename, "ns3_runnable_programs") ns3_runnable_scripts = get_list_from_file(lock_filename, "ns3_runnable_scripts") ns3_runnable_scripts = [os.path.basename(script) for script in ns3_runnable_scripts] else: print( "The build status file was not found. You must configure before running test.py.", file=sys.stderr, ) sys.exit(2) # # Make a dictionary that maps the name of a program to its path. # ns3_runnable_programs_dictionary = {} for program in ns3_runnable_programs: # Remove any directory names from path. program_name = os.path.basename(program) ns3_runnable_programs_dictionary[program_name] = program # Generate the lists of examples to run as smoke tests in order to # ensure that they remain buildable and runnable over time. # example_tests = [] example_names_original = [] python_tests = [] for directory in EXAMPLE_DIRECTORIES: # Set the directories and paths for this example. example_directory = os.path.join("examples", directory) examples_to_run_path = os.path.join(example_directory, "examples-to-run.py") cpp_executable_dir = os.path.join(NS3_BUILDDIR, example_directory) python_script_dir = os.path.join(example_directory) # Parse this example directory's file. parse_examples_to_run_file( examples_to_run_path, cpp_executable_dir, python_script_dir, example_tests, example_names_original, python_tests, ) for module in NS3_ENABLED_MODULES: # Remove the "ns3-" from the module name. module = module[len("ns3-") :] # Set the directories and paths for this example. module_directory = os.path.join("src", module) example_directory = os.path.join(module_directory, "examples") examples_to_run_path = os.path.join(module_directory, "test", "examples-to-run.py") cpp_executable_dir = os.path.join(NS3_BUILDDIR, example_directory) python_script_dir = os.path.join(example_directory) # Parse this module's file. parse_examples_to_run_file( examples_to_run_path, cpp_executable_dir, python_script_dir, example_tests, example_names_original, python_tests, ) for module in NS3_ENABLED_CONTRIBUTED_MODULES: # Remove the "ns3-" from the module name. module = module[len("ns3-") :] # Set the directories and paths for this example. module_directory = os.path.join("contrib", module) example_directory = os.path.join(module_directory, "examples") examples_to_run_path = os.path.join(module_directory, "test", "examples-to-run.py") cpp_executable_dir = os.path.join(NS3_BUILDDIR, example_directory) python_script_dir = os.path.join(example_directory) # Parse this module's file. parse_examples_to_run_file( examples_to_run_path, cpp_executable_dir, python_script_dir, example_tests, example_names_original, python_tests, ) # # If lots of logging is enabled, we can crash Python when it tries to # save all of the text. We just don't allow logging to be turned on when # test.py runs. If you want to see logging output from your tests, you # have to run them using the test-runner directly. # os.environ["NS_LOG"] = "*" if TEST_LOGS else "" # # There are a couple of options that imply we can to exit before starting # up a bunch of threads and running tests. Let's detect these cases and # handle them without doing all of the hard work. # if args.kinds: path_cmd = os.path.join("utils", test_runner_name + " --print-test-type-list") rc, standard_out, standard_err, et = run_job_synchronously( path_cmd, os.getcwd(), False, False ) print(standard_out) if args.list: list_items = [] if ENABLE_TESTS: if len(args.constrain): path_cmd = os.path.join( "utils", test_runner_name + " --print-test-name-list --print-test-types --test-type=%s" % args.constrain, ) else: path_cmd = os.path.join( "utils", test_runner_name + " --print-test-name-list --print-test-types" ) rc, standard_out, standard_err, et = run_job_synchronously( path_cmd, os.getcwd(), False, False ) if rc != 0: # This is usually a sign that ns-3 crashed or exited uncleanly print(("test.py error: test-runner return code returned {}".format(rc))) print( ( "To debug, try running {}\n".format( "'./ns3 run \"test-runner --print-test-name-list\"'" ) ) ) return if isinstance(standard_out, bytes): standard_out = standard_out.decode() list_items = standard_out.split("\n") list_items.sort() print("Test Type Test Name") print("--------------- ---------") for item in list_items: if len(item.strip()): print(item) examples_sorted = [] if ENABLE_EXAMPLES: examples_sorted = example_names_original examples_sorted.sort() if ENABLE_PYTHON_BINDINGS: python_examples_sorted = [] for x, y in python_tests: if y == "True": python_examples_sorted.append(x) python_examples_sorted.sort() examples_sorted.extend(python_examples_sorted) for item in examples_sorted: print("example ", item) print() if args.kinds or args.list: return # # We communicate results in two ways. First, a simple message relating # PASS, FAIL, CRASH or SKIP is always written to the standard output. It # is expected that this will be one of the main use cases. A developer can # just run test.py with no options and see that all of the tests still # pass. # # The second main use case is when detailed status is requested (with the # --text or --html options). Typically this will be text if a developer # finds a problem, or HTML for nightly builds. In these cases, an # XML file is written containing the status messages from the test suites. # This file is then read and translated into text or HTML. It is expected # that nobody will really be interested in the XML, so we write it somewhere # with a unique name (time) to avoid collisions. In case an error happens, we # provide a runtime option to retain the temporary files. # # When we run examples as smoke tests, they are going to want to create # lots and lots of trace files. We aren't really interested in the contents # of the trace files, so we also just stash them off in the temporary dir. # The retain option also causes these unchecked trace files to be kept. # date_and_time = time.strftime("%Y-%m-%d-%H-%M-%S-CUT", time.gmtime()) if not os.path.exists(TMP_OUTPUT_DIR): os.makedirs(TMP_OUTPUT_DIR) testpy_output_dir = os.path.join(TMP_OUTPUT_DIR, date_and_time) if not os.path.exists(testpy_output_dir): os.makedirs(testpy_output_dir) # # Load results from the latest results.xml, then use the list of # failed tests to filter out (SKIP) successful tests # previously_run_tests_to_skip = {"test": [], "example": []} if args.rerun_failed: previously_run_tests_to_skip = load_previously_successful_tests() # # Create the main output file and start filling it with XML. We need to # do this since the tests will just append individual results to this file. # The file is created outside the directory that gets automatically deleted. # xml_results_file = os.path.join(TMP_OUTPUT_DIR, f"{date_and_time}-results.xml") with open(xml_results_file, "w", encoding="utf-8") as f: f.write('\n') f.write("\n") # # We need to figure out what test suites to execute. We are either given one # suite or example explicitly via the --suite or --example/--pyexample option, # or we need to call into the test runner and ask it to list all of the available # test suites. Further, we need to provide the constraint information if it # has been given to us. # # This translates into allowing the following options with respect to the # suites # # ./test.py: run all of the suites and examples # ./test.py --constrain=core: run all of the suites of all kinds # ./test.py --constrain=unit: run all unit suites # ./test.py --suite=some-test-suite: run a single suite # ./test.py --example=examples/udp/udp-echo: run single example # ./test.py --pyexample=examples/wireless/mixed-wireless.py: run python example # ./test.py --suite=some-suite --example=some-example: run the single suite # # We can also use the --constrain option to provide an ordering of test # execution quite easily. # # Flag indicating a specific suite was explicitly requested single_suite = False if len(args.suite): # See if this is a valid test suite. path_cmd = os.path.join("utils", test_runner_name + " --print-test-name-list") rc, suites, standard_err, et = run_job_synchronously(path_cmd, os.getcwd(), False, False) if isinstance(suites, bytes): suites = suites.decode() suites = suites.replace("\r\n", "\n") suites_found = fnmatch.filter(suites.split("\n"), args.suite) if not suites_found: print( "The test suite was not run because an unknown test suite name was requested.", file=sys.stderr, ) sys.exit(2) elif len(suites_found) == 1: single_suite = True suites = "\n".join(suites_found) elif ENABLE_TESTS and len(args.example) == 0 and len(args.pyexample) == 0: if len(args.constrain): path_cmd = os.path.join( "utils", test_runner_name + " --print-test-name-list --test-type=%s" % args.constrain, ) rc, suites, standard_err, et = run_job_synchronously( path_cmd, os.getcwd(), False, False ) else: path_cmd = os.path.join("utils", test_runner_name + " --print-test-name-list") rc, suites, standard_err, et = run_job_synchronously( path_cmd, os.getcwd(), False, False ) else: suites = "" # # suite_list will either a single test suite name that the user has # indicated she wants to run or a list of test suites provided by # the test-runner possibly according to user provided constraints. # We go through the trouble of setting up the parallel execution # even in the case of a single suite to avoid having to process the # results in two different places. # if isinstance(suites, bytes): suites = suites.decode() suite_list = suites.split("\n") # # Performance tests should only be run when they are requested, # i.e. they are not run by default in test.py. # If a specific suite was requested we run it, even if # it is a performance test. if not single_suite and args.constrain != "performance": # Get a list of all of the performance tests. path_cmd = os.path.join( "utils", test_runner_name + " --print-test-name-list --test-type=%s" % "performance" ) rc, performance_tests, standard_err, et = run_job_synchronously( path_cmd, os.getcwd(), False, False ) if isinstance(performance_tests, bytes): performance_tests = performance_tests.decode() performance_test_list = performance_tests.split("\n") # Remove any performance tests from the suites list. for performance_test in performance_test_list: if performance_test in suite_list: suite_list.remove(performance_test) # We now have a possibly large number of test suites to run, so we want to # run them in parallel. We're going to spin up a number of worker threads # that will run our test jobs for us. # input_queue = queue.Queue(0) output_queue = queue.Queue(0) jobs = 0 threads = [] # # In Python 2.6 you can just use multiprocessing module, but we don't want # to introduce that dependency yet; so we jump through a few hoops. # processors = 1 if sys.platform != "win32": if "SC_NPROCESSORS_ONLN" in os.sysconf_names: processors = os.sysconf("SC_NPROCESSORS_ONLN") else: proc = subprocess.Popen( "sysctl -n hw.ncpu", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) stdout_results, stderr_results = proc.communicate() stdout_results = stdout_results.decode() stderr_results = stderr_results.decode() if len(stderr_results) == 0: processors = int(stdout_results) else: processors = os.cpu_count() if args.process_limit: if processors < args.process_limit: print("Using all %s processors" % processors) else: processors = args.process_limit print("Limiting to %s worker processes" % processors) # # Now, spin up one thread per processor which will eventually mean one test # per processor running concurrently. # for i in range(processors): thread = worker_thread(input_queue, output_queue) threads.append(thread) thread.start() # # Keep track of some summary statistics # total_tests = 0 skipped_tests = 0 skipped_testnames = [] # # We now have worker threads spun up, and a list of work to do. So, run # through the list of test suites and dispatch a job to run each one. # # Dispatching will run with unlimited speed and the worker threads will # execute as fast as possible from the queue. # # Note that we actually dispatch tests to be skipped, so all the # PASS, FAIL, CRASH and SKIP processing is done in the same place. # for test in suite_list: test = test.strip() if len(test): job = Job() job.set_is_example(False) job.set_is_pyexample(False) job.set_display_name(test) job.set_tmp_file_name(os.path.join(testpy_output_dir, "%s.xml" % test)) job.set_cwd(os.getcwd()) job.set_basedir(os.getcwd()) job.set_tempdir(testpy_output_dir) if args.multiple: multiple = "" else: multiple = " --stop-on-failure" if args.fullness: if len(args.fullness): fullness = args.fullness.upper() fullness = " --fullness=%s" % fullness else: fullness = " --fullness=QUICK" elif args.only_fullness: fullness = " --only-fullness=%s" % args.only_fullness.upper() path_cmd = os.path.join( "utils", test_runner_name + " --test-name=%s%s%s" % (test, multiple, fullness) ) job.set_shell_command(path_cmd) if args.valgrind and test in core_valgrind_skip_tests: job.set_is_skip(True) job.set_skip_reason("crashes valgrind") if args.rerun_failed and test in previously_run_tests_to_skip["test"]: job.is_skip = True job.set_skip_reason("didn't fail in the previous run") if args.verbose: print("Queue %s" % test) input_queue.put(job) jobs = jobs + 1 total_tests = total_tests + 1 # # We've taken care of the discovered or specified test suites. Now we # have to deal with examples run as smoke tests. We have a list of all of # the example programs it makes sense to try and run. Each example will # have a condition associated with it that must evaluate to true for us # to try and execute it. This is used to determine if the example has # a dependency that is not satisfied. # # We don't care at all how the trace files come out, so we just write them # to a single temporary directory. # # XXX As it stands, all of the trace files have unique names, and so file # collisions can only happen if two instances of an example are running in # two versions of the test.py process concurrently. We may want to create # uniquely named temporary traces directories to avoid this problem. # # We need to figure out what examples to execute. We are either given one # suite or example explicitly via the --suite or --example option, or we # need to walk the list of examples looking for available example # conditions. # # This translates into allowing the following options with respect to the # suites # # ./test.py: run all of the examples # ./test.py --constrain=unit run no examples # ./test.py --constrain=example run all of the examples # ./test.py --suite=some-test-suite run no examples # ./test.py --example=some-example run the single example with no parameters # ./test.py --example="some-example --args=2" run the single example with custom parameters # ./test.py --example=some-example* run the all examples-to-run.py instances with said example # ./test.py --suite=some-suite --example=some-example run the single example # # if len(args.suite) == 0 and len(args.pyexample) == 0: if len(args.constrain) == 0 or args.constrain == "example": if ENABLE_EXAMPLES: if args.example: if args.example.endswith("*"): # If an example name is passed without arguments, we filter all examples containing said program example_tests = list( filter(lambda x: args.example[:-1] in x[0], example_tests) ) args.example_args = [] else: example_tests = list( filter( lambda x: " ".join([args.example, *args.example_args]) == x[0].split("/")[-1], example_tests, ) ) args.example_args = [] if not example_tests or args.example_args: # If an example name is passed with arguments, we create an example entry for said example example_name = " ".join([args.example, *args.example_args]) example_path = "%s%s-%s%s" % ( APPNAME, VERSION, args.example, BUILD_PROFILE_SUFFIX, ) if example_path in ns3_runnable_programs_dictionary: example_path = ns3_runnable_programs_dictionary[example_path] example_path += ".exe" if sys.platform == "win32" else "" example_path = " ".join([example_path, *args.example_args]) example_tests = [(example_name, example_path, "True", "True", "QUICK")] else: print("No example matching the name %s" % example_name) example_tests = [] for name, test, do_run, do_valgrind_run, fullness in example_tests: # Remove any arguments and directory names from test. test_name = re.findall(r'(?:".*?"|\S)+', test)[0].strip('"') test_name = os.path.basename(test_name) test_name = test_name[:-4] if sys.platform == "win32" else test_name # Don't try to run this example if it isn't runnable. if test_name in ns3_runnable_programs_dictionary: if eval(do_run): job = Job() job.set_is_example(True) job.set_is_pyexample(False) job.set_display_name(name) job.set_tmp_file_name("") job.set_cwd(testpy_output_dir) job.set_basedir(os.getcwd()) job.set_tempdir(testpy_output_dir) job.set_shell_command(test) job.set_build_path(args.buildpath) if args.valgrind and not eval(do_valgrind_run): job.set_is_skip(True) job.set_skip_reason("skip in valgrind runs") if ( args.rerun_failed and name in previously_run_tests_to_skip["example"] ): job.is_skip = True job.set_skip_reason("didn't fail in the previous run") if args.verbose: print("Queue %s" % test) if args.fullness == "QUICK" and fullness != "QUICK": job.set_is_skip(True) job.set_skip_reason( f"skip {fullness} examples when QUICK run selected" ) elif ( args.fullness == "EXTENSIVE" and fullness != "EXTENSIVE" and fullness != "QUICK" ): job.set_is_skip(True) job.set_skip_reason( f"skip {fullness} examples when EXTENSIVE run selected" ) # TAKES_FOREVER includes everything, so no need to exclude anything input_queue.put(job) jobs = jobs + 1 total_tests = total_tests + 1 # # Run some Python examples as smoke tests. We have a list of all of # the example programs it makes sense to try and run. Each example will # have a condition associated with it that must evaluate to true for us # to try and execute it. This is used to determine if the example has # a dependency that is not satisfied. # # We don't care at all how the trace files come out, so we just write them # to a single temporary directory. # # We need to figure out what python examples to execute. We are either # given one pyexample explicitly via the --pyexample option, or we # need to walk the list of python examples # # This translates into allowing the following options with respect to the # suites # # ./test.py --constrain=pyexample run all of the python examples # ./test.py --pyexample=some-example.py: run the single python example # if len(args.suite) == 0 and len(args.example) == 0 and len(args.pyexample) == 0: if len(args.constrain) == 0 or args.constrain == "pyexample": for test, do_run, fullness in python_tests: # Remove any arguments and directory names from test. test_name = test.split(" ", 1)[0] test_name = os.path.basename(test_name) # Don't try to run this example if it isn't runnable. if test_name in ns3_runnable_scripts: if eval(do_run): job = Job() job.set_is_example(False) job.set_is_pyexample(True) job.set_display_name(test) job.set_tmp_file_name("") job.set_cwd(testpy_output_dir) job.set_basedir(os.getcwd()) job.set_tempdir(testpy_output_dir) job.set_shell_command(test) job.set_build_path("") # # Python programs and valgrind do not work and play # well together, so we skip them under valgrind. # We go through the trouble of doing all of this # work to report the skipped tests in a consistent # way through the output formatter. # if args.valgrind: job.set_is_skip(True) job.set_skip_reason("skip in valgrind runs") # # The user can disable python bindings, so we need # to pay attention to that and give some feedback # that we're not testing them # if not ENABLE_PYTHON_BINDINGS: job.set_is_skip(True) job.set_skip_reason("requires Python bindings") if args.verbose: print("Queue %s" % test) if args.fullness == "QUICK" and fullness != "QUICK": job.set_is_skip(True) job.set_skip_reason(f"skip {fullness} examples when QUICK run selected") elif ( args.fullness == "EXTENSIVE" and fullness != "EXTENSIVE" and fullness != "QUICK" ): job.set_is_skip(True) job.set_skip_reason( f"skip {fullness} examples when EXTENSIVE run selected" ) # TAKES_FOREVER includes everything, so no need to exclude anything input_queue.put(job) jobs = jobs + 1 total_tests = total_tests + 1 elif len(args.pyexample): # Find the full relative path to file if only a partial path has been given. if not os.path.exists(args.pyexample): import glob files = glob.glob("./**/%s" % args.pyexample, recursive=True) if files: args.pyexample = files[0] # Don't try to run this example if it isn't runnable. example_name = os.path.basename(args.pyexample) if example_name not in ns3_runnable_scripts: print("Example %s is not runnable." % example_name) elif not os.path.exists(args.pyexample): print("Example %s does not exist." % example_name) else: # # If you tell me to run a python example, I will try and run the example # irrespective of any condition. # job = Job() job.set_is_pyexample(True) job.set_display_name(args.pyexample) job.set_tmp_file_name("") job.set_cwd(testpy_output_dir) job.set_basedir(os.getcwd()) job.set_tempdir(testpy_output_dir) job.set_shell_command(args.pyexample) job.set_build_path("") if args.verbose: print("Queue %s" % args.pyexample) input_queue.put(job) jobs = jobs + 1 total_tests = total_tests + 1 # # Tell the worker threads to pack up and go home for the day. Each one # will exit when they see their is_break task. # for i in range(processors): job = Job() job.set_is_break(True) input_queue.put(job) # # Now all of the tests have been dispatched, so all we have to do here # in the main thread is to wait for them to complete. Keyboard interrupt # handling is broken as mentioned above. We use a signal handler to catch # sigint and set a global variable. When the worker threads sense this # they stop doing real work and will just start throwing jobs back at us # with is_break set to True. In this case, there are no real results so we # ignore them. If there are real results, we always print PASS or FAIL to # standard out as a quick indication of what happened. # passed_tests = 0 failed_tests = 0 failed_testnames = [] crashed_tests = 0 crashed_testnames = [] valgrind_errors = 0 valgrind_testnames = [] failed_jobs = [] for i in range(jobs): job = output_queue.get() if job.is_break: continue if job.is_example or job.is_pyexample: kind = "Example" else: kind = "TestSuite" if job.is_skip: status = "SKIP" status_print = colors.GREY + status + colors.NORMAL skipped_tests = skipped_tests + 1 skipped_testnames.append(job.display_name + (" (%s)" % job.skip_reason)) else: failed_jobs.append(job) if job.returncode == 0: status = "PASS" status_print = colors.GREEN + status + colors.NORMAL passed_tests = passed_tests + 1 failed_jobs.pop() elif job.returncode == 1: failed_tests = failed_tests + 1 failed_testnames.append(job.display_name) status = "FAIL" status_print = colors.RED + status + colors.NORMAL elif job.returncode == 2: valgrind_errors = valgrind_errors + 1 valgrind_testnames.append(job.display_name) status = "VALGR" status_print = colors.CYAN + status + colors.NORMAL else: crashed_tests = crashed_tests + 1 crashed_testnames.append(job.display_name) status = "CRASH" status_print = colors.PINK + status + colors.NORMAL print("[%d/%d] %s" % (i + 1, total_tests, status_print), end="") if args.duration or args.constrain == "performance": print(" (%.3f)" % job.elapsed_time, end="") print(":", end="") if "NS_COMMANDLINE_INTROSPECTION" in os.environ: print(" Wrote example usage for", end="") print(" %s %s" % (kind, job.display_name)) if job.is_example or job.is_pyexample: # # Examples are the odd man out here. They are written without any # knowledge that they are going to be run as a test, so we need to # cook up some kind of output for them. We're writing an xml file, # so we do some simple XML that says we ran the example. # # XXX We could add some timing information to the examples, i.e. run # them through time and print the results here. # with open(xml_results_file, "a", encoding="utf-8") as f: f.write("\n") example_name = " %s\n" % job.display_name f.write(example_name) if status == "PASS": f.write(" PASS\n") elif status == "FAIL": f.write(" FAIL\n") elif status == "VALGR": f.write(" VALGR\n") elif status == "SKIP": f.write(" SKIP\n") f.write(" %s\n" % job.skip_reason) else: f.write(" CRASH\n") f.write(' \n") else: # # If we're not running an example, we're running a test suite. # These puppies are running concurrently and generating output # that was written to a temporary file to avoid collisions. # # Now that we are executing sequentially in the main thread, we can # concatenate the contents of the associated temp file to the main # results file and remove that temp file. # # One thing to consider is that a test suite can crash just as # well as any other program, so we need to deal with that # possibility as well. If it ran correctly it will return 0 # if it passed, or 1 if it failed. In this case, we can count # on the results file it saved being complete. If it crashed, it # will return some other code, and the file should be considered # corrupt and useless. If the suite didn't create any XML, then # we're going to have to do it ourselves. # # Another issue is how to deal with a valgrind error. If we run # a test suite under valgrind and it passes, we will get a return # code of 0 and there will be a valid xml results file since the code # ran to completion. If we get a return code of 1 under valgrind, # the test case failed, but valgrind did not find any problems so the # test case return code was passed through. We will have a valid xml # results file here as well since the test suite ran. If we see a # return code of 2, this means that valgrind found an error (we asked # it to return 2 if it found a problem in run_job_synchronously) but # the suite ran to completion so there is a valid xml results file. # If the suite crashes under valgrind we will see some other error # return code (like 139). If valgrind finds an illegal instruction or # some other strange problem, it will die with its own strange return # code (like 132). However, if the test crashes by itself, not under # valgrind we will also see some other return code. # # If the return code is 0, 1, or 2, we have a valid xml file. If we # get another return code, we have no xml and we can't really say what # happened -- maybe the TestSuite crashed, maybe valgrind crashed due # to an illegal instruction. If we get something beside 0-2, we assume # a crash and fake up an xml entry. After this is all done, we still # need to indicate a valgrind error somehow, so we fake up an xml entry # with a VALGR result. Thus, in the case of a working TestSuite that # fails valgrind, we'll see the PASS entry for the working TestSuite # followed by a VALGR failing test suite of the same name. # if job.is_skip: with open(xml_results_file, "a", encoding="utf-8") as f: f.write("\n") f.write(" %s\n" % job.display_name) f.write(" SKIP\n") f.write(" %s\n" % job.skip_reason) f.write("\n") else: failed_jobs.append(job) if job.returncode == 0 or job.returncode == 1 or job.returncode == 2: with open(xml_results_file, "a", encoding="utf-8") as f_to, open( job.tmp_file_name, encoding="utf-8" ) as f_from: contents = f_from.read() if status == "VALGR": pre = contents.find("") + len("") post = contents.find("") contents = contents[:pre] + "VALGR" + contents[post:] f_to.write(contents) # When running with sanitizers, the program may # crash before ever writing the expected xml # output file try: et = ET.parse(job.tmp_file_name) if et.find("Result").text in ["PASS", "SKIP"]: failed_jobs.pop() except: pass else: with open(xml_results_file, "a", encoding="utf-8") as f: f.write("\n") f.write(" %s\n" % job.display_name) f.write(" CRASH\n") f.write("\n") # # We have all of the tests run and the results written out. One final # bit of housekeeping is to wait for all of the threads to close down # so we can exit gracefully. # for thread in threads: thread.join() # # Back at the beginning of time, we started the body of an XML document # since the test suites and examples were going to just write their # individual pieces. So, we need to finish off and close out the XML # document # with open(xml_results_file, "a", encoding="utf-8") as f: f.write("\n") # # Print a quick summary of events # print( "%d of %d tests passed (%d passed, %d skipped, %d failed, %d crashed, %d valgrind errors)" % ( passed_tests, total_tests, passed_tests, skipped_tests, failed_tests, crashed_tests, valgrind_errors, ) ) # # Repeat summary of skipped, failed, crashed, valgrind events # if skipped_testnames: skipped_testnames.sort() print("List of SKIPped tests:\n %s" % "\n ".join(map(str, skipped_testnames))) if failed_testnames: failed_testnames.sort() print("List of FAILed tests:\n %s" % "\n ".join(map(str, failed_testnames))) if crashed_testnames: crashed_testnames.sort() print("List of CRASHed tests:\n %s" % "\n ".join(map(str, crashed_testnames))) if valgrind_testnames: valgrind_testnames.sort() print("List of VALGR failures:\n %s" % "\n ".join(map(str, valgrind_testnames))) if failed_jobs and args.verbose_failed: for job in failed_jobs: if job.standard_out or job.standard_err: job_type = "example" if (job.is_example or job.is_pyexample) else "test suite" print( f"===================== Begin of {job_type} '{job.display_name}' stdout =====================" ) print(job.standard_out) print( f"===================== Begin of {job_type} '{job.display_name}' stderr =====================" ) print(job.standard_err) print( f"===================== End of {job_type} '{job.display_name}' ==============================" ) # # The last things to do are to translate the XML results file to "human- # readable form" if the user asked for it (or make an XML file somewhere) # if len(args.html) + len(args.text) + len(args.xml): print() if len(args.html): translate_to_html(xml_results_file, args.html) if len(args.text): translate_to_text(xml_results_file, args.text) if len(args.xml): xml_file = args.xml + (".xml" if ".xml" not in args.xml else "") print("Writing results to xml file %s..." % xml_file, end="") shutil.copyfile(xml_results_file, xml_file) print("done.") # # Let the user know if they need to turn on tests or examples. # if not ENABLE_TESTS or not ENABLE_EXAMPLES: print() if not ENABLE_TESTS: print("*** Note: ns-3 tests are currently disabled. Enable them by adding") print('*** "--enable-tests" to ./ns3 configure or modifying your .ns3rc file.') print() if not ENABLE_EXAMPLES: print("*** Note: ns-3 examples are currently disabled. Enable them by adding") print('*** "--enable-examples" to ./ns3 configure or modifying your .ns3rc file.') print() # # Let the user know if they tried to use valgrind but it was not # present on their machine. # if args.valgrind and not VALGRIND_FOUND: print() print("*** Note: you are trying to use valgrind, but valgrind could not be found") print("*** on your machine. All tests and examples will crash or be skipped.") print() # # If we have been asked to retain all of the little temporary files, we # don't delete tm. If we do delete the temporary files, delete only the # directory we just created. We don't want to happily delete any retained # directories, which will probably surprise the user. # if not args.retain: shutil.rmtree(TMP_OUTPUT_DIR) if passed_tests + skipped_tests == total_tests: return 0 # success else: return 1 # catchall for general errors def split_program_and_arguments(argv): split_argv = re.findall(r'(?:".*[|*]?"|\S)+', argv) program = "" program_args = [] if split_argv: program = split_argv[0] if len(split_argv) > 1: program_args = split_argv[1:] return program, program_args def main(argv): parser = argparse.ArgumentParser() parser.add_argument( "-b", "--buildpath", action="store", type=str, default="", help="specify the path where ns-3 was built (defaults to the build directory for the current variant)", ) parser.add_argument( "-c", "--constrain", action="store", type=str, default="", help="constrain the test-runner by kind of test", ) parser.add_argument( "-d", "--duration", action="store_true", default=False, help="print the duration of each test suite and example", ) parser.add_argument( "-e", "--example", action="store", type=str, default="", help="specify a single example to run (no relative path is needed)", ) parser.add_argument( "-u", "--update-data", action="store_true", default=False, help="If examples use reference data files, get them to re-generate them", ) fullness_group = parser.add_mutually_exclusive_group(required=False) fullness_group.add_argument( "-f", "--fullness", action="store", type=str, default="QUICK", choices=["QUICK", "EXTENSIVE", "TAKES_FOREVER"], help="choose the duration of tests to run: QUICK, EXTENSIVE, or TAKES_FOREVER, where EXTENSIVE includes QUICK and TAKES_FOREVER includes QUICK and EXTENSIVE (only QUICK tests are run by default)", ) fullness_group.add_argument( "-of", "--only-fullness", action="store", type=str, default=None, choices=["QUICK", "EXTENSIVE", "TAKES_FOREVER"], help="choose the duration of tests to run: QUICK, EXTENSIVE, or TAKES_FOREVER (only tests marked with fullness will be executed)", ) parser.add_argument( "-g", "--grind", action="store_true", dest="valgrind", default=False, help="run the test suites and examples using valgrind", ) parser.add_argument( "-k", "--kinds", action="store_true", default=False, help="print the kinds of tests available", ) parser.add_argument( "-l", "--list", action="store_true", default=False, help="print the list of known tests" ) parser.add_argument( "-m", "--multiple", action="store_true", default=False, help="report multiple failures from test suites and test cases", ) parser.add_argument( "-n", "--no-build", action="store_true", default=False, help="do not build before starting testing", ) parser.add_argument( "-p", "--pyexample", action="store", type=str, default="", help="specify a single python example to run (with relative path)", ) parser.add_argument( "-r", "--retain", action="store_true", default=False, help="retain all temporary files (which are normally deleted)", ) parser.add_argument( "-s", "--suite", action="store", type=str, default="", help="specify a single test suite to run", ) parser.add_argument( "-t", "--text", action="store", type=str, default="", metavar="TEXT-FILE", help="write detailed test results into TEXT-FILE.txt", ) parser.add_argument( "-v", "--verbose", action="store_true", default=False, help="print progress and informational messages", ) parser.add_argument( "--verbose-failed", action="store_true", default=False, help="print progress and informational messages for failed jobs", ) parser.add_argument( "-w", "--web", "--html", action="store", type=str, dest="html", default="", metavar="HTML-FILE", help="write detailed test results into HTML-FILE.html", ) parser.add_argument( "-x", "--xml", action="store", type=str, default="", metavar="XML-FILE", help="write detailed test results into XML-FILE.xml", ) parser.add_argument( "--nocolor", action="store_true", default=False, help="do not use colors in the standard output", ) parser.add_argument( "--jobs", action="store", type=int, dest="process_limit", default=0, help="limit number of worker threads", ) parser.add_argument( "--rerun-failed", action="store_true", dest="rerun_failed", default=False, help="rerun failed tests", ) global args args = parser.parse_args() args.example, exargs = split_program_and_arguments(args.example) setattr(args, "example_args", exargs) signal.signal(signal.SIGINT, sigint_hook) # From waf/waflib/Options.py envcolor = os.environ.get("NOCOLOR", "") and "no" or "auto" or "yes" if args.nocolor or envcolor == "no": colors_lst["USE"] = False return run_tests() if __name__ == "__main__": sys.exit(main(sys.argv))