Test: run_all is now in Python and defaults to parallelizing tests

In addition to the conversion to Python, run_all now defaults to running
all Python tests in parallel, using the same approach previously
available via '--fast'. If desired one can force all tests to run
serially by calling run_all with --serial

A debugging flag was now also included in run_all.  Pass one or more -l
(-l, -ll or -lll) for different levels of debugging information.

test/run_all

@@ -0,0 +1,206 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+from __future__ import print_function
+import os
+import sys
+import glob
+import argparse
+import logging
+import time
+from multiprocessing import cpu_count
+from Queue import Queue, Empty
+from threading import Thread
+from subprocess import call, Popen, PIPE
+
+# Look for taskd in $PATH instead of task/src/
+os.environ["TASKD_USE_PATH"] = "1"
+
+TIMEOUT = .2
+
+
+def run_test(testqueue, outqueue):
+    while True:
+        try:
+            test = testqueue.get(block=True, timeout=TIMEOUT)
+        except Empty:
+            break
+
+        log.info("Running test %s", test)
+
+        try:
+            p = Popen(os.path.abspath(test), stdout=PIPE, stderr=PIPE,
+                      env=os.environ)
+            out, err = p.communicate()
+        except Exception as e:
+            log.exception(e)
+            # Premature end
+            break
+
+        output = ("# {0}\n".format(os.path.basename(test)), out, err)
+        log.debug("Collected output %s", output)
+        outqueue.put(output)
+
+        testqueue.task_done()
+
+
+class TestRunner(object):
+    def __init__(self):
+        self.threads = []
+        self.tap = open(cmd_args.tapfile, 'w')
+
+        self._parallelq = Queue()
+        self._serialq = Queue()
+        self._outputq = Queue()
+
+    def _find_tests(self):
+        for test in glob.glob("*.t") + glob.glob("*.t.exe"):
+            if os.access(test, os.X_OK):
+                # Executables only
+                if not cmd_args.serial:
+                    with open(test) as fh:
+                        if "/usr/bin/env python" in fh.readline():
+                            log.debug("Treating as parallel: %s", test)
+                            self._parallelq.put(test)
+                        else:
+                            log.debug("Treating as serial: %s", test)
+                            self._serialq.put(test)
+                else:
+                    log.debug("Treating %s as serial", test)
+                    self._serialq.put(test)
+            else:
+                log.debug("Ignored test %s as it is not executable", test)
+
+        log.info("Parallel tests: %s", self._parallelq.qsize())
+        log.info("Serial tests: %s", self._serialq.qsize())
+
+    def _prepare_threads(self):
+        # Serial thread
+        self.threads.append(
+            Thread(target=run_test, args=(self._serialq, self._outputq))
+        )
+        # Parallel threads
+        self.threads.extend([
+            Thread(target=run_test, args=(self._parallelq, self._outputq))
+            for i in range(cpu_count())
+        ])
+        log.info("Spawned %s threads to run tests", len(self.threads))
+
+    def _start_threads(self):
+        for thread in self.threads:
+            # Threads die when main thread dies
+            log.debug("Starting thread %s", thread)
+            thread.daemon = True
+            thread.start()
+
+    def _print_timestamp_to_tap(self):
+        now = time.time()
+        timestamp = "# {0} ==> {1}\n".format(
+            now,
+            time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(now)),
+        )
+
+        log.debug("Adding timestamp %s to TAP file", timestamp)
+        self.tap.write(timestamp)
+
+    def _get_remaining_tests(self):
+        return self._parallelq.qsize() + self._serialq.qsize()
+
+    def is_running(self):
+        for thread in self.threads:
+            if thread.is_alive():
+                return True
+
+        return False
+
+    def start(self):
+        self._find_tests()
+        self._prepare_threads()
+
+        self._print_timestamp_to_tap()
+
+        finished = 0
+        total = self._get_remaining_tests()
+
+        self._start_threads()
+
+        while self.is_running() or not self._outputq.empty():
+            try:
+                outputs = self._outputq.get(block=True, timeout=TIMEOUT)
+            except Empty:
+                continue
+
+            log.debug("Outputting to TAP: %s", outputs)
+
+            for output in outputs:
+                self.tap.write(output)
+
+                if cmd_args.verbose:
+                    sys.stdout.write(output)
+
+            self._outputq.task_done()
+            finished += 1
+
+            log.warning("Finished %s out of %s tests", finished, total)
+
+        self._print_timestamp_to_tap()
+
+        if not self._parallelq.empty() or not self._serialq.empty():
+            raise RuntimeError(
+                "Something went wrong, not all tests were ran. {0} "
+                "remaining.".format(self._get_remaining_tests()))
+
+    def show_report(self):
+        self.tap.flush()
+
+        log.debug("Calling 'problems --summary' for report")
+        call([os.path.abspath("problems"), "--summary", cmd_args.tapfile])
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description="Run Taskwarrior tests")
+    parser.add_argument('--verbose', '-v', action="store_true",
+                        help="Also send TAP output to stdout")
+    parser.add_argument('--logging', '-l', action="count",
+                        help="Logging level. -lll is the highest level")
+    parser.add_argument('--serial', action="store_true",
+                        help="Do not run tests in parallel")
+    parser.add_argument('--tapfile', default="all.log",
+                        help="File to use for TAP output")
+    return parser.parse_args()
+
+
+def main():
+    runner = TestRunner()
+    runner.start()
+    if not cmd_args.verbose:
+        runner.show_report()
+
+
+if __name__ == "__main__":
+    cmd_args = parse_args()
+
+    if cmd_args.logging == 1:
+        level = logging.WARN
+    elif cmd_args.logging == 2:
+        level = logging.INFO
+    elif cmd_args.logging >= 3:
+        level = logging.DEBUG
+    else:
+        level = logging.ERROR
+
+    logging.basicConfig(
+        format="# %(asctime)s - %(levelname)s - %(message)s",
+        level=level,
+    )
+    log = logging.getLogger(__name__)
+
+    log.debug("Parsed commandline arguments: %s", cmd_args)
+
+    try:
+        main()
+    except Exception as e:
+        log.exception(e)
+        sys.exit(1)
+
+# vim: ai sts=4 et sw=4