made tests parallel and time measurements more precise

Author: koss-null

py3/easy/roman_to_integer/__pycache__/roman_to_integer.cpython-38.pyc

@@ -18,10 +18,13 @@ def romanToInt(self, s):
         return num + cache
 
 
-def main():
-    a = input()
-    print(Solution().romanToInt(a))
+def get_input(inp):
+    return inp
 
 
-if __name__ == "__main__":
-    main()
+def format_output(a):
+    return str(a)
+
+
+def main(a):
+    return Solution().romanToInt(a)

py3/easy/roman_to_integer/roman_to_integer.py

@@ -33,11 +33,13 @@ def isMatch(self, s, p):
         return self.check_match(s, rg, 0)
 
 
-def main():
-    a = input()
-    b = input()
-    print(Solution().isMatch(a, b))
+def main(a, b):
+    return Solution().isMatch(a, b)
 
 
-if __name__ == "__main__":
-    main()
+def get_input(inp):
+    return inp.split("\n")
+
+
+def format_output(a):
+    return str(a)

py3/hard/is_match/__pycache__/is_match.cpython-38.pyc

@@ -84,11 +84,17 @@ def findMedianSortedArrays(self, nums1, nums2, desired_place=None, need_two=None
         return -1
 
 
-def main():
-    a = list(map(int, input().split()))
-    b = list(map(int, input().split()))
-    print(Solution().findMedianSortedArrays(a, b))
+def main(a, b):
+    return Solution().findMedianSortedArrays(a, b)
 
 
-if __name__ == "__main__":
-    main()
+def get_input(inp):
+    inp = inp.split("\n")
+    a = list(map(int, inp[0].split()))
+    b = list(map(int, inp[1].split()))
+    return a, b
+
+
+def format_output(a):
+    return str(a)
+

py3/hard/is_match/is_match.py

@@ -1,7 +1,19 @@
-def main():
-    a, b = input().split()
-    print(int(a) + abs(int(b)))
+def solve(a, b):
+    return a + b
+
+
+def get_input(inp):
+    a, b = inp.split()
+    return int(a), int(b)
+
+
+def format_output(a):
+    return str(a)
+
+
+def main(a, b):
+    return solve(a, b)
 
 
 if __name__ == "__main__":
-    main()
+    main(*get_input("1 2"))

py3/hard/median/__pycache__/median.cpython-38.pyc

@@ -5,7 +5,9 @@
 import sys
 import time
 import json
-import subprocess
+import importlib
+from collections import abc
+import multiprocessing as mp
 
 from collections import namedtuple as nt
 
@@ -26,34 +28,63 @@ def print_red(text):
     print(f"{bcolors.FAIL}{text}{bcolors.ENDC}")  
 
 
-Test = nt("Test", "process output expected name duration")
+Test = nt("Test", "solution result test_case name duration")
+lock = mp.Lock()
 
 
 def check_result(test):
-    out = test.output.decode("ascii").strip()
-    if test.expected == out: 
+    out = test.solution.format_output(*listify(test.result))
+    lock.acquire()
+    if test.test_case["output"] == out:
         print_green(f"Test: {test.name} is [OK]\tduration: {test.duration}")
         print_green(out)
+        print("-" * 10)
     else:
         print_red(f"Test: {test.name} is [FAIL]\tduration: {test.duration}")
-        print_green(test.expected)
+        print_red(f"Input:\n{test.test_case['input']}\n---")
+        print_green(test.test_case["output"])
         print_red(out)
         print("-" * 10)
+    lock.release()
+
+
+def listify(a):
+    if isinstance(a, str):
+        return [a.strip()]
+    if isinstance(a, abc.Iterable):
+        return a
+    return [a]
 
 
 def run_test(program_path, test_path):
     with open(test_path, "r") as test:
         test_case = json.load(test)
-        started = time.time()
-        p = subprocess.Popen(["python3", program_path + "/" + program_path.split("/")[-1] + ".py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+        solution_module_name = ".".join(program_path.split("/")) + "." + program_path.split("/")[-1]
+        solution = importlib.import_module(solution_module_name)
+        try:
+            # each solution should have main() function which returns the result solution
+            # and get_input() function which accepts the input as str, returns the input in prepared for main format
+            # and format_output() function which returns string to compare
+            inp = solution.get_input(test_case["input"])
+            started = time.time()
+            res = solution.main(*listify(inp))
+        except Exception as ex:
+            lock.acquire()
+            print_red("{} [FAIL]".format(test_path))
+            lock.release()
+            raise ex
         finished = time.time()
-        p = subprocess.Popen(["python3", program_path + "/" + program_path.split("/")[-1] + ".py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
-        outp, errs = p.communicate(str(test_case["input"]).encode("ascii"))
-        if errs:
-            print_red("[FAIL]")
-            print(errs)
-            return
-        check_result(Test(p, outp, test_case["output"], test_path.split("/")[-1], finished - started))
+        check_result(Test(solution, res, test_case, test_path.split("/")[-1], finished - started))
+
+
+SYSTEM_CORES_AMOUNT = 16
+
+
+def flush_processes(procs):
+    for p in procs:
+        p.start()
+    for p in procs:
+        p.join()
 
 
 def main(argv):
@@ -63,10 +94,15 @@ def main(argv):
     """
     cwd = os.getcwd()
     program_path = os.path.join(cwd, argv[1])
+    test_processes = []
     for test_path in os.listdir(os.path.join(program_path, "test")):
         if not re.match(test_path, "_res$") and not re.match(test_path, "_test_res$"):
             full_path = os.path.join(program_path, "test", test_path)
-            run_test(program_path, full_path)
+            test_processes.append(mp.Process(target=run_test, args=(argv[1], full_path)))
+            if len(test_processes) == SYSTEM_CORES_AMOUNT:
+                flush_processes(test_processes)
+                test_processes = []
+    flush_processes(test_processes)
 
 
 if __name__ == "__main__":