diff options
| -rw-r--r-- | test/test_all.py | 23 | ||||
| -rw-r--r-- | test/utils/monothread.py | 20 | ||||
| -rw-r--r-- | test/utils/multithread.py | 24 | ||||
| -rw-r--r-- | test/utils/screendisplay.py | 115 | ||||
| -rw-r--r-- | test/utils/testset.py | 4 |
5 files changed, 42 insertions, 144 deletions
diff --git a/test/test_all.py b/test/test_all.py index 195be412..34bb0a55 100644 --- a/test/test_all.py +++ b/test/test_all.py @@ -5,7 +5,8 @@ import tempfile from utils.test import Test from utils.testset import TestSet -from utils import cosmetics, monothread, screendisplay +from utils import cosmetics, multithread +from multiprocessing import Queue testset = TestSet("../") TAGS = {"regression": "REGRESSION", # Regression tests @@ -667,16 +668,15 @@ By default, no tag is omitted." % ", ".join(TAGS.keys()), default="") "Z3 and its python binding are necessary for TranslatorZ3." if TAGS["z3"] not in exclude_tags: exclude_tags.append(TAGS["z3"]) + test_ko = [] + test_ok = [] # Set callbacks if multiproc is False: - testset.set_callback(task_done=monothread.task_done, - task_new=monothread.task_new) testset.set_cpu_numbers(1) - else: - screendisplay.init(testset.cpu_c) - testset.set_callback(task_done=screendisplay.task_done, - task_new=screendisplay.task_new) + testset.set_callback(task_done=lambda test, error:multithread.task_done(test, error, test_ok, test_ko), + task_new=multithread.task_new) + # Filter testset according to tags testset.filter_tags(exclude_tags=exclude_tags) @@ -686,6 +686,13 @@ By default, no tag is omitted." % ", ".join(TAGS.keys()), default="") # Finalize testset.end(clean=not args.do_not_clean) - + print + print (cosmetics.colors["green"] + + "Result: %d/%d pass" % (len(test_ok), len(test_ok) + len(test_ko)) + + cosmetics.colors["end"]) + for test, error in test_ko: + command_line = " ".join(test.command_line) + print cosmetics.colors["red"] + 'ERROR', cosmetics.colors["lightcyan"] + command_line + cosmetics.colors["end"] + print error # Exit with an error if at least a test failed exit(testset.tests_passed()) diff --git a/test/utils/monothread.py b/test/utils/monothread.py deleted file mode 100644 index ae64f3c5..00000000 --- a/test/utils/monothread.py +++ /dev/null @@ -1,20 +0,0 @@ -import sys -import cosmetics - - -def task_done(test, error): - s = "[%s] Running tests on %s ..." % (test.base_dir.upper(), - " ".join(test.command_line)) - already_printed = len(s) - if error is not None: - cosmetics.write_colored("ERROR", "red", already_printed) - print error - else: - cosmetics.write_colored("OK", "green", already_printed) - - -def task_new(test): - s = "[%s] Running tests on %s ..." % (test.base_dir.upper(), - " ".join(test.command_line)) - sys.stdout.write(s) - sys.stdout.flush() diff --git a/test/utils/multithread.py b/test/utils/multithread.py new file mode 100644 index 00000000..287b5ebd --- /dev/null +++ b/test/utils/multithread.py @@ -0,0 +1,24 @@ +import sys +import cosmetics +import time + + +def task_done(test, error, test_ok, test_ko): + command_line = " ".join(test.command_line) + if error is not None: + print cosmetics.colors["red"] + 'ERROR', + print cosmetics.colors["lightcyan"] + command_line + cosmetics.colors["end"] + print error + test_ko.append((test, error)) + else: + print cosmetics.colors["green"] + 'DONE', + print cosmetics.colors["lightcyan"] + command_line + cosmetics.colors["end"], + print "%ds" % (time.time() - test.start_time) + test_ok.append((test, error)) + + +def task_new(test): + command_line = " ".join(test.command_line) + print cosmetics.colors["lightcyan"], + print test.base_dir.upper(), command_line, + print cosmetics.colors["end"] diff --git a/test/utils/screendisplay.py b/test/utils/screendisplay.py deleted file mode 100644 index 7c7bfde1..00000000 --- a/test/utils/screendisplay.py +++ /dev/null @@ -1,115 +0,0 @@ -import time -import signal -from cosmetics import getTerminalSize, colors - - -global_state = {"termSize": getTerminalSize(), - "message": "", - "pstate": []} - - -def print_conf(conf, value): - "Print a configuration line" - return colors["green"] + conf + ": " + colors["end"] + str(value) - - -def clr_screen(): - "Update the screen to display some information" - - # Header - to_print = [] - to_print.append(" " * (global_state["termSize"][0] / 2 - 10) + colors[ - "blue"] + "Miasm2 Regression tests" + colors["end"]) - to_print.append("") - to_print.append("=" * global_state["termSize"][0]) - to_print.append("") - to_print.append(print_conf("Current mode", "Multiprocessing")) - to_print.append(print_conf("Nb CPU detected", global_state["cpu_c"])) - to_print.append("") - to_print.append("=" * global_state["termSize"][0]) - to_print.append("") - test_done = 0 - test_failed = 0 - message = global_state["message"] + "\n" - for v in global_state["pstate"]: - if v["status"] != "running": - test_done += 1 - if v["status"] != 0: - test_failed += 1 - cmd_line = " ".join(v["test"].command_line) - message += colors["red"] + "FAIL:" + colors["end"] + cmd_line - message += "\n" + v["message"] + "\n" - - to_print.append(print_conf("Success rate", "%d/%d" % - (test_done - test_failed, test_done))) - printed_time = time.strftime( - "%M:%S", time.gmtime(time.time() - global_state["init_time"])) - to_print.append(print_conf("Cumulated time", printed_time)) - to_print.append("") - to_print.append("=" * global_state["termSize"][0]) - - cur = "\n".join(to_print) - cur += "\n" - - # Message - cur += message - print cur - already_printed = cur.count("\n") - - # Current state - current_job = [] - for process in global_state["pstate"]: - if process["status"] == "running": - current_job.append(process) - print "\n" * (global_state["termSize"][1] - already_printed - 3 - len(current_job)) - - for job in current_job: - command_line = " ".join(job["test"].command_line) - base_dir = job["test"].base_dir.upper() - s = "[" + colors["lightcyan"] + command_line + colors["end"] - s_end = base_dir - cur_time = time.strftime( - "%M:%Ss", time.gmtime(time.time() - job["init_time"])) - l = len(command_line) + len(s_end) + 4 + len(str(cur_time)) + 2 - s_end += " " + colors["blue"] + cur_time + colors["end"] + "]" - print "%s%s%s" % (s, " " * (global_state["termSize"][0] - l), s_end) - - -def on_signal(sig1, sig2): - "Update view every second" - clr_screen() - signal.alarm(1) - - -def init(cpu_c): - """Initialize global state - @cpu_c: number of cpu (for conf displaying) - """ - # Init global_state - global_state["cpu_c"] = cpu_c - global_state["init_time"] = time.time() - - # Launch view updater - signal.signal(signal.SIGALRM, on_signal) - signal.alarm(1) - - -def task_done(test, error): - "Report a test done" - for task in global_state["pstate"]: - if task["test"] == test: - if error is not None: - task["status"] = -1 - task["message"] = error - else: - task["status"] = 0 - break - clr_screen() - - -def task_new(test): - "Report a new test" - global_state["pstate"].append({"status": "running", - "test": test, - "init_time": time.time()}) - clr_screen() diff --git a/test/utils/testset.py b/test/utils/testset.py index 54df732c..e510ff6e 100644 --- a/test/utils/testset.py +++ b/test/utils/testset.py @@ -1,9 +1,10 @@ import os import subprocess import sys +import time from multiprocessing import cpu_count, Queue, Process -from test import Test +from test import Test class Message(object): "Message exchanged in the TestSet message queue" @@ -136,6 +137,7 @@ class TestSet(object): test = todo_queue.get() if test is None: break + test.start_time = time.time() message_queue.put(MessageTaskNew(test)) # Go to the expected directory |