run_demos.py

#!/bin/env python
"""
Program to manage and run ATS demo problems.

With inspiration, and in some places just stolen, from Ben Andre's
PFloTran regression test suite.

Author: Ethan Coon (ecoon@lanl.gov)
"""
from __future__ import print_function

import sys,os
import argparse
import textwrap
import time

try:
    import test_manager
except ImportError:
    sys.path.append(os.path.join(os.environ['ATS_SRC_DIR'], 'tools', 'testing'))
    import test_manager

def commandline_options(args=None):
    """
    Process the command line arguments and return them as a dict.
    """
    parser = argparse.ArgumentParser(description='Run an collection of ATS demo problems.')

    parser.add_argument('--backtrace', action='store_true',
                        help='show exception backtraces as extra debugging '
                        'output')

    parser.add_argument('--debug', action='store_true',
                        help='extra debugging output')

    parser.add_argument('-d', '--dry-run',
                        default=False, action='store_true',
                        help='perform a dry run, setup the test commands but '
                        'don\'t run them')

    parser.add_argument('-e', '--executable', nargs=1, default=None,
                        help='path to executable to use for testing')

    parser.add_argument('--list-suites', default=False, action='store_true',
                        help='print the list of test suites from the config '
                        'file and exit')

    parser.add_argument('--list-tests', default=False, action='store_true',
                        help='print the list of tests from the config file '
                        'and exit')

    parser.add_argument('-m', '--mpiexec', nargs=1, default=None,
                        help="path to the executable for mpiexec (mpirun, etc)"
                        "on the current machine.")

    parser.add_argument('-s', '--suites', nargs="+", default=[],
                        help='space separated list of test suite names')

    parser.add_argument('-t', '--tests', nargs="+", default=[],
                        help='space separated list of test names')

    parser.add_argument('--exclude', nargs="+", default=[],
                        help='space separated list of test names to exclude')

    parser.add_argument('--timeout', nargs=1, default=None,
                        help="test timeout (for assuming a job has hung and "
                        "needs to be killed)")

    parser.add_argument('configs', metavar='CONFIG_LOCATION', type=str,
                        nargs='+', help='list of directories and/or configuration '
                        'files to parse for suites and tests')
    
    options = parser.parse_args(args)
    return options


def main(options):
    txtwrap = textwrap.TextWrapper(width=78, subsequent_indent=4*" ")
    testlog = test_manager.setup_testlog(txtwrap)

    root_dir = os.getcwd()

    test_manager.check_options(options)
    executable = test_manager.check_for_executable(options, testlog)
    mpiexec = test_manager.check_for_mpiexec(options, testlog)
    config_file_list = test_manager.generate_config_file_list(options)

    print("Running ATS demo problems :")

    # loop through config files, cd into the appropriate directory,
    # read the appropriate config file and run the various tests.
    start = time.time()
    report = {}
    for config_file in sorted(config_file_list):
        # try:
            # NOTE(bja): the try block is inside this loop so that if
            # a single test throws an exception in a large batch of
            # tests, we can recover and at least try running the other
            # config files.
            print(80 * '=', file=testlog)
            print(f'Running {config_file}', file=testlog)

            d,f = os.path.split(config_file)
            d1,header = os.path.split(d)
            if len(header) > 20:
                header = header[:20]
            else:
                header = header + ' '*(20-len(header))
            print(f'{header} | ', end='')

            # get the absolute path of the directory
            test_dir = os.path.dirname(config_file)
            # cd into the test directory so that the relative paths in
            # test files are correct
            os.chdir(test_dir)
            if options.debug:
                print("Changed to working directory: {0}".format(test_dir))

            tm = test_manager.RegressionTestManager(executable, mpiexec, suffix='demo')

            if options.debug:
                tm.debug(True)

            # get the relative file name
            filename = os.path.basename(config_file)

            tm.generate_tests(filename,
                              options.suites,
                              options.tests,
                              options.exclude,
                              options.timeout,
                              False,
                              testlog)

            if options.debug:
                print(70 * '-')
                print(tm)

            if options.list_suites:
                tm.display_available_suites()

            if options.list_tests:
                tm.display_available_tests()

            tm.run_tests(options.dry_run,
                         False,
                         False,
                         False,
                         True,
                         testlog)

            report[filename] = tm.run_status()
            os.chdir(root_dir)
        # except Exception as error:
        #     message = txtwrap.fill(
        #         "ERROR: a problem occured in file '{0}'.  This is "
        #         "probably an error with commandline options, the "
        #         "configuration file, or an internal error.  The "
        #         "error is:\n{1}".format(config_file, str(error)))
        #     print(''.join(['\n', message, '\n']), file=testlog)
        #     if options.backtrace:
        #         traceback.print_exc()
        #     print('F', end='', file=sys.stdout)
        #     report[filename] = TestStatus()
        #     report[filename].fail = 1

            
    stop = time.time()
    status = 0
    if not options.dry_run:
        print("")
        run_time = stop - start
        test_manager.summary_report_by_file(report, testlog)
        test_manager.summary_report(run_time, report, testlog)
        status = test_manager.summary_report(run_time, report, sys.stdout)

    testlog.close()

    return status


def _preserve_cwd(function):
   def decorator(*args, **kwargs):
      cwd = os.getcwd()
      result = function(*args, **kwargs)
      os.chdir(cwd)
      return result
   return decorator


@_preserve_cwd
def run_demo_local(testnames, config_dir=None, options_list=None, force=False):
    """This is a short wrapper to run a demo not from the commandline."""
    if isinstance(testnames, str):
        testnames = [testnames,]
    
    if config_dir is None:
        config_dir = os.path.split(os.getcwd())[-1]
    
    workdir = os.path.join(os.environ['ATS_SRC_DIR'], 'docs', 'documentation', 'source', 'ats_demos')

    demodirs = [os.path.join(workdir, config_dir, testname+'.demo') for testname in testnames]
    testnames = [tname for (dd, tname) in zip(demodirs, testnames) if (not os.path.isdir(dd) or force)]

    if len(testnames) > 0:
        os.chdir(workdir)
        if options_list is None:
            options_list = []
        options_list += [config_dir, '-t'] + testnames
        opts = commandline_options(options_list)
        main(opts)


if __name__ == "__main__":
    cmdl_options = commandline_options()
    suite_status = main(cmdl_options)
    sys.exit(suite_status)