Working on overhauling the process handling. Making it so you can spe…

…cify per-process initialization and cleanup.
ikollipara · Jun 16, 2015 · 695110f · 695110f
1 parent 6c72401
commit 695110f
Show file tree

Hide file tree

Showing 10 changed files with 133 additions and 71 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -18,7 +18,7 @@
 
 - Switched to cyan instead of blue on Windows only.
 - Stubbed in the beginnings of support for designating initialization to run in
-  each subprocess to obtain whatever resources might be needed by a single
+  each process to obtain whatever resources might be needed by a single
   process (like its own database, for example).
 
 
@@ -168,7 +168,7 @@ Issue #47.
 - Fixed a crash that could occur if an exception was raised during a test
   case's setUpClass() or tearDownClass()
 - We now explicitly terminate the thread pool and join() it.  This makes self
-  unit tests much easier to clean up on Windows, where the subprocesses would
+  unit tests much easier to clean up on Windows, where the processes would
   block deletion of temporary directories.
 - Set up continuous integration for Windows using AppVeyor.  Thanks to ionelmc
   for the tip!  Issue #11.
@@ -215,7 +215,7 @@ Issue #47.
   auto-detect all versions of python (in the form of pythonX.Y) in $PATH and
   run many permutations of self tests on each version.
 - Fixed a crash that could occur if discovery did not find any tests and
-  subprocesses was set higher than one.
+  processes was set higher than one.
 
 - Fixed lots of tests so that they would succeed in all environments.
 
@@ -311,7 +311,7 @@ Issue #47.
 - Clean - Low redundancy in output. Result stats for each test is lined up in a
   vertical column.
 
-- Fast - Can run tests in independent subprocesses.
+- Fast - Can run tests in independent processes.
 
 - Powerful - Multi-target + auto-discovery.
 

diff --git a/README-pypi.rst b/README-pypi.rst
@@ -18,7 +18,7 @@ Features
 
 - **Colorful** - Terminal output makes good use of color when the terminal supports it.
 - **Clean** - Low redundancy in output. Result stats for each test is lined up in a vertical column.
-- **Fast** - Can run tests in independent subprocesses.
+- **Fast** - Can run tests in independent processes.
 - **Powerful** - Multi-target + auto-discovery.
 - **Traditional** - Use the normal ``unittest`` classes and methods for your unit tests.
 - **Descriptive** - Four verbosity levels, from just dots to full docstring output.

diff --git a/README.md b/README.md
@@ -31,7 +31,7 @@ Features
 
 - **Colorful** - Terminal output makes good use of color when the terminal supports it.
 - **Clean** - Low redundancy in output. Result statistics for each test is vertically aligned.
-- **Fast** - Tests can run in independent subprocesses.
+- **Fast** - Tests can run in independent processes.
 - **Powerful** - Multi-target + auto-discovery.
 - **Traditional** - Use the normal `unittest` classes and methods for your unit tests.
 - **Descriptive** - Four verbosity levels, from just dots to full docstring output.

diff --git a/green/config.py b/green/config.py
@@ -41,7 +41,9 @@
 # Set the defaults in a re-usable way
 default_args = argparse.Namespace( # pragma: no cover
         targets         = ['.'], # Not in configs
-        subprocesses    = 1,
+        processes       = 1,
+        initializer     = '',
+        finalizer       = '',
         html            = False,
         termcolor       = None,
         notermcolor     = None,
@@ -153,18 +155,33 @@ def parseArguments(): # pragma: no cover
 
     concurrency_args = parser.add_argument_group("Concurrency Options")
     store_opt(
-        concurrency_args.add_argument('-s', '--subprocesses', action='store',
+        concurrency_args.add_argument('-s', '--processes', action='store',
             type=int, metavar='NUM',
-            help="Number of subprocesses to use to run tests.  Note that your "
+            help="Number of processes to use to run tests.  Note that your "
             "tests need to be written to avoid using the same resources (temp "
             "files, sockets, ports, etc.) for the multi-process mode to work "
-            "well. Default is 1, meaning disable using subprocesses. 0 means "
-            "try to autodetect the number of CPUs in the system. Note that for "
-            "a small number of trivial tests, running everything in a single "
-            "process may be faster than the overhead of initializing all the "
-            "subprocesses.",
+            "well (--initializer and --finalizer can help provision "
+            "per-process resources). Default is 1, meaning just use a single "
+            "process (maximizes compatibility with standard unittest and other "
+            "test runners). 0 means try to autodetect the number of CPUs in "
+            "the system. Note that for a small number of trivial tests, "
+            "running everything in a single process may be faster than the "
+            "overhead of initializing all the processes.",
         default=argparse.SUPPRESS))
-
+    store_opt(
+        concurrency_args.add_argument('-i', '--initializer', action='store',
+            metavar='EXECUTABLE_FILE', default='',
+            help="Executable to run inside of a single worker process before "
+            "it starts running tests.  This is the way to provision external "
+            "resources that each concurrent worker process needs to have "
+            "exclusive access to.  Can be a relative or absolute path."))
+    store_opt(
+        concurrency_args.add_argument('-z', '--finalizer', action='store',
+            metavar='EXECUTABLE_FILE', default='',
+            help="Executable to run inside of a single worker process after "
+            "it completes running tests and the process is about to be "
+            "destroyed.  This is the way to reclaim resources provisioned with "
+            "--initializer.  Can be a relative or absolute path."))
     format_args = parser.add_argument_group("Format Options")
     store_opt(format_args.add_argument('-m', '--html', action='store_true',
         help="HTML5 format.  Overrides terminal color options if specified.",
@@ -351,9 +368,10 @@ def mergeConfig(args, testing=False, coverage_testing=False): # pragma: no cover
                 'logging', 'version', 'failfast', 'run_coverage', 'options',
                 'completions', 'completion_file']:
             config_getter = config.getboolean
-        elif name in ['subprocesses', 'debug', 'verbose']:
+        elif name in ['processes', 'debug', 'verbose']:
             config_getter = config.getint
-        elif name in ['omit_patterns', 'warnings', 'file_pattern', 'test_pattern']:
+        elif name in ['file_pattern', 'finalizer', 'initializer',
+                'omit_patterns', 'warnings', 'test_pattern']:
             config_getter = config.get
         elif name in ['targets', 'help', 'config']:
             pass # Some options only make sense coming on the command-line.
@@ -442,4 +460,13 @@ def mergeConfig(args, testing=False, coverage_testing=False): # pragma: no cover
             cov.start()
         new_args.cov = cov
 
+    # Initializer and finalizer need to be real, executable files
+    new_args.initializer = new_args.initializer and os.path.realpath(new_args.initializer)
+    new_args.finalizer = new_args.finalizer and os.path.realpath(new_args.finalizer)
+    for cmd in [new_args.initializer, new_args.finalizer]:
+        if cmd and (not os.access(cmd, os.X_OK)):
+            print("'{}' is not an executable file.".format(cmd))
+            new_args.shouldExit = True
+            new_args.exitCode = 4
+
     return new_args
diff --git a/green/examples.py b/green/examples.py
@@ -52,6 +52,6 @@ def test5UnexpectedPass(self):
 #
 #
 #    def test8AnotherDelay(self):
-#        "This test will also take 2 seconds to pass, but not if you use 3 or more subprocesses!"
+#        "This test will also take 2 seconds to pass.
 #        import time
 #        time.sleep(2)
diff --git a/green/subprocess.py → green/process.py b/green/subprocess.py → green/process.py
@@ -16,9 +16,10 @@
 from green.loader import loadTargets
 
 
-class SubprocessLogger(object):
+
+class ProcessLogger(object):
     """I am used by LoggingDaemonlessPool to get crash output out to the
-    logger, instead of having subprocess crashes be silent"""
+    logger, instead of having process crashes be silent"""
 
 
     def __init__(self, callable):
@@ -47,7 +48,7 @@ def __call__(self, *args, **kwargs):
 
 class DaemonlessProcess(multiprocessing.Process):
     """I am used by LoggingDaemonlessPool to make pool workers NOT run in
-    daemon mode (daemon mode subprocess can't launch their own subprocesses)"""
+    daemon mode (daemon mode process can't launch their own subprocesses)"""
 
 
     def _get_daemon(self):
@@ -63,18 +64,16 @@ def _set_daemon(self, value):
 
 
 
-
-
 class LoggingDaemonlessPool(Pool):
-    "I use SubprocessLogger and DaemonlessProcess to make a pool of workers."
+    "I use ProcessLogger and DaemonlessProcess to make a pool of workers."
 
 
     Process = DaemonlessProcess
 
 
     def apply_async(self, func, args=(), kwds={}, callback=None):
         return Pool.apply_async(
-                self, SubprocessLogger(func), args, kwds, callback)
+                self, ProcessLogger(func), args, kwds, callback)
 
 #-------------------------------------------------------------------------------
 # START of Worker Finalization Monkey Patching
@@ -90,10 +89,10 @@ def __init__(self, processes=None, initializer=None, initargs=(),
         self._finalizer = finalizer
         self._finalargs = finalargs
         # Python 2 and 3 have different method signatures
-        if platform.python_version_tuple()[0] == '2':
+        if platform.python_version_tuple()[0] == '2': # pragma: no cover
             super(LoggingDaemonlessPool, self).__init__(processes, initializer,
                     initargs, maxtasksperchild)
-        else:
+        else: # pragma: no cover
             super(LoggingDaemonlessPool, self).__init__(processes, initializer,
                     initargs, maxtasksperchild, context)
 
@@ -124,9 +123,9 @@ def _repopulate_pool(self):
 from multiprocessing.pool import MaybeEncodingError
 
 # Python 2 and 3 raise a different error when they exit
-if platform.python_version_tuple()[0] == '2':
+if platform.python_version_tuple()[0] == '2': # pragma: no cover
     PortableOSError = IOError
-else:
+else: # pragma: no cover
     PortableOSError = OSError
 
 
@@ -140,7 +139,12 @@ def worker(inqueue, outqueue, initializer=None, initargs=(), maxtasks=None,
         outqueue._reader.close()
 
     if initializer is not None:
-        initializer(*initargs)
+        try:
+            initializer(*initargs)
+        except Exception as e:
+            print("Warning, initializer command '{}' failed with:\n{}"
+                    .format(initializer.command,
+                        ''.join(traceback.format_tb(sys.exc_info()[2]))))
 
     completed = 0
     while maxtasks is None or (maxtasks and completed < maxtasks):
@@ -175,13 +179,16 @@ def worker(inqueue, outqueue, initializer=None, initargs=(), maxtasks=None,
 
     util.debug('worker exiting after %d tasks' % completed)
 
+
+
 # Unmodified (see above)
 class RemoteTraceback(Exception):
     def __init__(self, tb):
         self.tb = tb
     def __str__(self):
         return self.tb
 
+
 # Unmodified (see above)
 class ExceptionWithTraceback:
     def __init__(self, exc, tb):
@@ -192,6 +199,7 @@ def __init__(self, exc, tb):
     def __reduce__(self):
         return rebuild_exc, (self.exc, self.tb)
 
+
 # Unmodified (see above)
 def rebuild_exc(exc, tb):
     exc.__cause__ = RemoteTraceback(tb)
@@ -202,7 +210,7 @@ def rebuild_exc(exc, tb):
 #-------------------------------------------------------------------------------
 
 def poolRunner(test_name, coverage_number=None, omit_patterns=[]):
-    "I am the function that pool worker subprocesses run.  I run one unit test."
+    "I am the function that pool worker processes run.  I run one unit test."
     # Each pool worker gets his own temp directory, to avoid having tests that
     # are used to taking turns using the same temp file name from interfering
     # with eachother.  So long as the test doesn't use a hard-coded temp

diff --git a/green/result.py b/green/result.py
@@ -38,7 +38,7 @@ def proto_error(err):
 
 class ProtoTest():
     """I take a full-fledged TestCase and preserve just the information we need
-    and can pass between subprocesses.
+    and can pass between processes.
     """
 
 
@@ -87,7 +87,7 @@ def getDescription(self, verbose):
 
 class ProtoError():
     """I take a full-fledged test error and preserve just the information we
-    need and can bass between subprocesses.
+    need and can bass between processes.
     """
 
 
@@ -132,7 +132,7 @@ def displayStdout(self, test):
 
 class ProtoTestResult(BaseTestResult):
     """
-    I'm the TestResult object for a single unit test run in a subprocess.
+    I'm the TestResult object for a single unit test run in a process.
     """
 
 

diff --git a/green/runner.py b/green/runner.py
@@ -1,6 +1,8 @@
 from __future__ import unicode_literals
 from __future__ import print_function
 
+from subprocess import check_output
+import sys
 from unittest.signals import (
         registerResult, installHandler, removeResult)
 import warnings
@@ -13,7 +15,30 @@
 from green.loader import toProtoTestList
 from green.output import GreenStream
 from green.result import GreenTestResult
-from green.subprocess import LoggingDaemonlessPool, poolRunner
+from green.process import LoggingDaemonlessPool, poolRunner
+
+
+class InitializerOrFinalizer:
+    """
+    I represent a command that will be run as either the initializer or the
+    finalizer for a worker process.  The only reason I'm a class instead of a
+    function is so that I can be instantiated at the creation time of the Pool
+    (with the user's customized command to run), but actually run at the
+    appropriate time.
+    """
+    def __init__(self, command):
+        self.command = command
+
+
+    def __call__(self, *args):
+        if not self.command:
+            return
+        print(str(args), self.command)
+        try:
+            check_output(self.command)
+        except:
+            raise(Exception('aoeuaoeuaoeu'))
+
 
 
 
@@ -48,11 +73,13 @@ def run(suite, stream, args):
 
         result.startTestRun()
 
-        if args.subprocesses == 1:
+        if args.processes == 1:
             suite.run(result)
         else:
             tests = toProtoTestList(suite)
-            pool = LoggingDaemonlessPool(processes=args.subprocesses or None)
+            pool = LoggingDaemonlessPool(processes=args.processes or None,
+                    initializer=InitializerOrFinalizer(args.initializer),
+                    finalizer=InitializerOrFinalizer(args.finalizer))
             if tests:
                 async_responses = []
                 for index, test in enumerate(tests):