runtests.py 6.3 KB
Newer Older
1 2
#!/usr/bin/env python3

3
import argparse, importlib, logging, os, re, sys, tempfile, time, traceback
4 5
current_dir = os.path.dirname(os.path.realpath(__file__))
sys.path.append(current_dir + "/tools")
6 7
import dnstest.params as params
import dnstest.utils
8

9 10
TESTS_DIR = "tests"
COMMON_DATA_DIR = "data"
11
LAST_WORKING_DIR = params.outs_dir + "/knottest-last"
12

13
def save_traceback(outdir):
14
    file = open(params.out_dir + "/traceback.log", mode="a")
15 16 17
    traceback.print_exc(file=file)
    file.close()

18 19 20 21 22 23 24 25 26 27 28
def create_log(logger, filename="", level=logging.NOTSET):
    if filename:
        handler = logging.FileHandler(filename)
    else:
        handler = logging.StreamHandler()
    handler.setLevel(level)
    formatter = logging.Formatter('%(asctime)s# %(message)s', "%H:%M:%S")
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    return handler

29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48
def parse_args(cmd_args):
    parser = argparse.ArgumentParser()
    parser.add_argument("-d", dest="debug", action="store_true", \
                        help="enable exception traceback on stdout")
    parser.add_argument("tests", metavar="[:]test[/case]", nargs="*", \
                        help="([exclude] | run) specific (test set | [test case])")
    args = parser.parse_args(cmd_args)

    params.debug = True if args.debug else False
    params.common_data_dir = current_dir + '/' + COMMON_DATA_DIR

    # Process tests/cases arguments.
    excluded = dict()
    included = dict()
    for item in args.tests:
        if re.match(":", item):
            item = item[1:]
            storage = excluded
        else:
            storage = included
49

50 51 52 53 54 55 56 57 58 59 60
        parts = item.split("/")
        if len(parts) == 1:
            case = list()
        elif len(parts) == 2:
            case = [parts[1]]
        else:
            print("Invalid argument %s" % item)
            sys.exit(1)
        test = parts[0]

        if test in storage:
61
            storage[test].extend(case)
62 63 64 65 66 67 68 69 70 71 72 73 74 75 76
        else:
            storage[test] = case

    # List all tests if nothing was specified.
    if not included:
        for i in sorted(os.listdir("./" + TESTS_DIR)):
            included[i] = list()

    return included, excluded

def main(args):
    included, excluded = parse_args(args)

    timestamp = int(time.time())
    today = time.strftime("%Y-%m-%d", time.localtime(timestamp))
77 78
    outs_dir = tempfile.mkdtemp(prefix="knottest-%s-" % timestamp,
                                dir=params.outs_dir)
79

80 81 82 83 84 85 86 87
    # Try to create symlink to the latest result.
    try:
        if os.path.exists(LAST_WORKING_DIR):
            os.remove(LAST_WORKING_DIR)
        os.symlink(outs_dir, LAST_WORKING_DIR)
    except:
        pass

88 89 90 91 92
    # Set up logging.
    log = logging.getLogger()
    log.setLevel(logging.NOTSET)
    create_log(log)
    create_log(log, outs_dir + "/summary.log", logging.NOTSET)
93

94 95 96 97 98 99 100 101 102
    log.info("KNOT TESTING SUITE %s" % today)
    log.info("Working directory %s" % outs_dir)

    case_cnt = 0
    fail_cnt = 0
    skip_cnt = 0
    for test in sorted(included):
        # Skip excluded test set.
        if test in excluded and not excluded[test]:
103 104
            continue

105 106 107 108
        # Check test directory.
        test_dir = "%s/%s/%s" % (current_dir, TESTS_DIR, test)
        if not os.path.isdir(test_dir):
            log.error("Test \'%s\':\tIGNORED (invalid folder)" % test)
109 110
            continue

111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136
        log.info("Test \'%s\'" % test)

        # Set test cases to run.
        if not included[test]:
            # List all test cases.
            cases = sorted(os.listdir(test_dir))
        else:
            cases = included[test]

        for case in cases:
            # Skip excluded cases.
            if test in excluded and case in excluded[test]:
                continue

            case_cnt += 1

            case_dir = test_dir + "/" + case
            test_file = case_dir + "/test.py"
            if not os.path.isfile(test_file):
                log.error(" * case \'%s\':\tMISSING" % case)
                fail_cnt += 1
                continue

            try:
                out_dir = outs_dir + "/" + test + "/" + case
                os.makedirs(out_dir, exist_ok=True)
137
                params.module = TESTS_DIR + "." + test + "." + case
138 139 140 141 142 143 144 145 146 147 148 149 150 151
                params.test_dir = case_dir
                params.out_dir = out_dir
                params.case_log = open(out_dir + "/case.log", mode="a")
                params.test = None
                params.err = False
                params.err_msg = ""
            except OsError:
                log.error(" * case \'%s\':\tEXCEPTION (no dir \'%s\')" %
                          (case, out_dir))
                fail_cnt += 1
                continue

            try:
                importlib.import_module("%s.%s.%s.test" % (TESTS_DIR, test, case))
152
            except dnstest.utils.Skip as exc:
153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172
                log.error(" * case \'%s\':\tSKIPPED (%s)" % (case, format(exc)))
                skip_cnt += 1
            except Exception as exc:
                save_traceback(params.out_dir)
                if params.debug:
                    traceback.print_exc()
                else:
                    log.error(" * case \'%s\':\tEXCEPTION (%s)" %
                              (case, format(exc)))
                fail_cnt += 1
            except BaseException as exc:
                save_traceback(params.out_dir)
                if params.debug:
                    traceback.print_exc()
                else:
                    log.info("INTERRUPTED")
                # Stop servers if still running.
                if params.test:
                    params.test.end()
                sys.exit(1)
173
            else:
174 175 176 177 178 179 180
                if params.err:
                    msg = " (%s)" % params.err_msg if params.err_msg else ""
                    log.info(" * case \'%s\':\tFAILED%s" % (case, msg))
                    fail_cnt += 1
                else:
                    log.info(" * case \'%s\':\tOK" % case)

181 182
            # Stop servers if still running.
            if params.test:
183
                params.test.end()
184

185
            params.case_log.close()
186

187 188 189 190
    msg_cases = "TEST CASES: %i" % case_cnt
    msg_skips = ", SKIPPED: %i" % skip_cnt if skip_cnt > 0 else ""
    msg_res = ", FAILED: %i" % fail_cnt if fail_cnt > 0 else ", SUCCESS"
    log.info(msg_cases + msg_skips + msg_res)
191

192 193 194 195
    if fail_cnt:
        sys.exit(1)
    else:
        sys.exit(0)
196

197 198
if __name__ == "__main__":
    main(sys.argv[1:])