streamline multiprocessing tests

consolidated checking of multiprocessing python module availability into
package setup

created a base class for all multiprocessing plugin tests to do cleanups

6 files changed, 50 insertions, 92 deletions

diff --git a/functional_tests/test_multiprocessing/__init__.py b/functional_tests/test_multiprocessing/__init__.py
new file mode 100644
index 0000000..856b0ad
--- /dev/null
+++ b/functional_tests/test_multiprocessing/__init__.py
@@ -0,0 +1,31 @@
+import os
+from unittest import TestCase
+
+from nose.plugins import PluginTester
+from nose.plugins.skip import SkipTest
+from nose.plugins.multiprocess import MultiProcess
+
+support = os.path.join(os.path.dirname(__file__), 'support')
+
+def setup():
+    try:
+        import multiprocessing
+        if 'active' in MultiProcess.status:
+            raise SkipTest("Multiprocess plugin is active. Skipping tests of "
+                           "plugin itself.")
+    except ImportError:
+        raise SkipTest("multiprocessing module not available")
+
+class MPTestBase(PluginTester, TestCase):
+    processes = 1
+    activate = '--processes=1'
+    plugins = [MultiProcess()]
+    suitepath = os.path.join(support, 'timeout.py')
+
+    def __init__(self, *args, **kwargs):
+        self.activate = '--processes=%d' % self.processes
+        PluginTester.__init__(self)
+        TestCase.__init__(self, *args, **kwargs)
+
+    def tearDown(self):
+        MultiProcess.status.pop('active')
diff --git a/functional_tests/test_multiprocessing/support/timeout.py b/functional_tests/test_multiprocessing/support/timeout.py
index 52dce12..480c859 100644
--- a/functional_tests/test_multiprocessing/support/timeout.py
+++ b/functional_tests/test_multiprocessing/support/timeout.py
@@ -1,6 +1,12 @@
+#make sure all tests in this file are dispatched to the same subprocess
+def setup():
+    pass
 
 def test_timeout():
     "this test *should* fail when process-timeout=1"
     from time import sleep
     sleep(2)
 
+# check timeout will not prevent remaining tests dispatched to the same subprocess to continue to run
+def test_pass():
+    pass
diff --git a/functional_tests/test_multiprocessing/test_class.py b/functional_tests/test_multiprocessing/test_class.py
index 1e36ba6..6f42ac3 100644
--- a/functional_tests/test_multiprocessing/test_class.py
+++ b/functional_tests/test_multiprocessing/test_class.py
@@ -1,33 +1,13 @@
 import os
-import unittest
 
-from nose.plugins import PluginTester
-from nose.plugins.skip import SkipTest
-from nose.plugins.multiprocess import MultiProcess
-
-
-support = os.path.join(os.path.dirname(__file__), 'support')
-
-
-def setup():
-    try:
-        import multiprocessing
-        if 'active' in MultiProcess.status:
-            raise SkipTest("Multiprocess plugin is active. Skipping tests of "
-                           "plugin itself.")
-    except ImportError:
-        raise SkipTest("multiprocessing module not available")
+from . import support, MPTestBase
 
 
 #test case for #462
-class TestClassFixture(PluginTester, unittest.TestCase):
-    activate = '--processes=1'
-    plugins = [MultiProcess()]
+class TestClassFixture(MPTestBase):
     suitepath = os.path.join(support, 'class.py')
 
     def runTest(self):
         assert str(self.output).strip().endswith('OK')
         assert 'Ran 2 tests' in self.output
-    def tearDown(self):
-        MultiProcess.status.pop('active')
 
diff --git a/functional_tests/test_multiprocessing/test_concurrent_shared.py b/functional_tests/test_multiprocessing/test_concurrent_shared.py
index cfb36c4..9974bfd 100644
--- a/functional_tests/test_multiprocessing/test_concurrent_shared.py
+++ b/functional_tests/test_multiprocessing/test_concurrent_shared.py
@@ -1,32 +1,12 @@
 import os
-import unittest
 
-from nose.plugins import PluginTester
-from nose.plugins.skip import SkipTest
-from nose.plugins.multiprocess import MultiProcess
+from . import support, MPTestBase
 
-
-support = os.path.join(os.path.dirname(__file__), 'support')
-
-
-def setup():
-    try:
-        import multiprocessing
-        if 'active' in MultiProcess.status:
-            raise SkipTest("Multiprocess plugin is active. Skipping tests of "
-                           "plugin itself.")
-    except ImportError:
-        raise SkipTest("multiprocessing module not available")
-
-
-class TestConcurrentShared(PluginTester, unittest.TestCase):
-    activate = '--processes=2'
-    plugins = [MultiProcess()]
+class TestConcurrentShared(MPTestBase):
+    processes = 2
     suitepath = os.path.join(support, 'concurrent_shared')
 
     def runTest(self):
         assert 'Ran 2 tests in 1.' in self.output, "make sure two tests use 1.x seconds (no more than 2 seconsd)"
         assert str(self.output).strip().endswith('OK')
-    def tearDown(self):
-        MultiProcess.status.pop('active')
 
diff --git a/functional_tests/test_multiprocessing/test_nameerror.py b/functional_tests/test_multiprocessing/test_nameerror.py
index d3d7210..4837aca 100644
--- a/functional_tests/test_multiprocessing/test_nameerror.py
+++ b/functional_tests/test_multiprocessing/test_nameerror.py
@@ -1,33 +1,13 @@
 import os
-import unittest
 
-from nose.plugins import PluginTester
-from nose.plugins.skip import SkipTest
-from nose.plugins.multiprocess import MultiProcess
+from . import support, MPTestBase
 
-
-support = os.path.join(os.path.dirname(__file__), 'support')
-
-
-def setup():
-    try:
-        import multiprocessing
-        if 'active' in MultiProcess.status:
-            raise SkipTest("Multiprocess plugin is active. Skipping tests of "
-                           "plugin itself.")
-    except ImportError:
-        raise SkipTest("multiprocessing module not available")
-
-
-class TestMPNameError(PluginTester, unittest.TestCase):
-    activate = '--processes=2'
-    plugins = [MultiProcess()]
+class TestMPNameError(MPTestBase):
+    processes = 2
     suitepath = os.path.join(support, 'nameerror.py')
 
     def runTest(self):
         print str(self.output)
         assert 'NameError' in self.output
         assert "'undefined_variable' is not defined" in self.output
-    def tearDown(self):
-        MultiProcess.status.pop('active')
 
diff --git a/functional_tests/test_multiprocessing/test_process_timeout.py b/functional_tests/test_multiprocessing/test_process_timeout.py
index 27e0584..afef674 100644
--- a/functional_tests/test_multiprocessing/test_process_timeout.py
+++ b/functional_tests/test_multiprocessing/test_process_timeout.py
@@ -1,40 +1,21 @@
 import os
-import unittest
 
-from nose.plugins import PluginTester
-from nose.plugins.skip import SkipTest
-from nose.plugins.multiprocess import MultiProcess
+from . import support, MPTestBase
 
-support = os.path.join(os.path.dirname(__file__), 'support')
-
-
-def setup():
-    try:
-        import multiprocessing
-        if 'active' in MultiProcess.status:
-            raise SkipTest("Multiprocess plugin is active. Skipping tests of "
-                           "plugin itself.")
-    except ImportError:
-        raise SkipTest("multiprocessing module not available")
-
-
-
-class TestMPTimeout(PluginTester, unittest.TestCase):
-    activate = '--processes=2'
+class TestMPTimeout(MPTestBase):
     args = ['--process-timeout=1']
-    plugins = [MultiProcess()]
     suitepath = os.path.join(support, 'timeout.py')
 
     def runTest(self):
         assert "TimedOutException: 'timeout.test_timeout'" in self.output
-
+        assert "Ran 2 tests in" in self.output
+        assert "FAILED (errors=1)" in self.output
 
 class TestMPTimeoutPass(TestMPTimeout):
     args = ['--process-timeout=3']
 
     def runTest(self):
         assert "TimedOutException: 'timeout.test_timeout'" not in self.output
+        assert "Ran 2 tests in" in self.output
         assert str(self.output).strip().endswith('OK')
-    def tearDown(self):
-        MultiProcess.status.pop('active')