[pypy-svn] r77809 - pypy/branch/fast-forward/pypy/module/_multiprocessing/test

afa at codespeak.net afa at codespeak.net
Mon Oct 11 23:39:32 CEST 2010


Author: afa
Date: Mon Oct 11 23:39:30 2010
New Revision: 77809

Modified:
   pypy/branch/fast-forward/pypy/module/_multiprocessing/test/test_connection.py
Log:
Refactor: one test for two implementations.


Modified: pypy/branch/fast-forward/pypy/module/_multiprocessing/test/test_connection.py
==============================================================================
--- pypy/branch/fast-forward/pypy/module/_multiprocessing/test/test_connection.py	(original)
+++ pypy/branch/fast-forward/pypy/module/_multiprocessing/test/test_connection.py	Mon Oct 11 23:39:30 2010
@@ -7,43 +7,48 @@
         from pypy.module._multiprocessing import interp_connection
         from pypy.module._multiprocessing import interp_semaphore
 
-class AppTestConnection:
-    def setup_class(cls):
-        space = gettestobjspace(usemodules=('_multiprocessing', 'thread'))
-        cls.space = space
-        if sys.platform == "win32":
-            # stubs for some modules,
-            # just for multiprocessing to import correctly.
-            w_modules = space.sys.get('modules')
-            space.setitem(w_modules, space.wrap('msvcrt'), space.sys)
-            space.setitem(w_modules, space.wrap('_subprocess'), space.sys)
-
-        # import multiprocessing once
-        space.appexec([], """(): import multiprocessing""")
-
-    def test_winpipe_connection(self):
-        import sys
-        if sys.platform != "win32":
-            skip("win32 only")
-
-        import multiprocessing
-        rhandle, whandle = multiprocessing.Pipe()
+class BaseConnectionTest(object):
+    def test_connection(self):
+        rhandle, whandle = self.make_pair()
 
         obj = [1, 2.0, "hello"]
         whandle.send(obj)
         obj2 = rhandle.recv()
         assert obj == obj2
 
-    def test_ospipe_connection(self):
-        import _multiprocessing
-        import os
-        fd1, fd2 = os.pipe()
-        rhandle = _multiprocessing.Connection(fd1, writable=False)
-        whandle = _multiprocessing.Connection(fd2, readable=False)
+class AppTestWinpipeConnection(BaseConnectionTest):
+    def setup_class(cls):
+        if sys.platform != "win32":
+            py.test.skip("win32 only")
 
-        obj = [1, 2.0, "hello"]
-        whandle.send(obj)
-        obj2 = rhandle.recv()
-        assert obj == obj2
+        space = gettestobjspace(usemodules=('_multiprocessing', 'thread'))
+        cls.space = space
 
+        # stubs for some modules,
+        # just for multiprocessing to import correctly on Windows
+        w_modules = space.sys.get('modules')
+        space.setitem(w_modules, space.wrap('msvcrt'), space.sys)
+        space.setitem(w_modules, space.wrap('_subprocess'), space.sys)
+
+        cls.w_make_pair = space.appexec([], """():
+            import multiprocessing
+            def make_pair():
+                rhandle, whandle = multiprocessing.Pipe()
+                return rhandle, whandle
+            return make_pair
+        """)
 
+class AppTestSocketConnection(BaseConnectionTest):
+    def setup_class(cls):
+        space = gettestobjspace(usemodules=('_multiprocessing', 'thread'))
+        cls.space = space
+        cls.w_make_pair = space.appexec([], """():
+            import _multiprocessing
+            import os
+            def make_pair():
+                fd1, fd2 = os.pipe()
+                rhandle = _multiprocessing.Connection(fd1, writable=False)
+                whandle = _multiprocessing.Connection(fd2, readable=False)
+                return rhandle, whandle
+            return make_pair
+        """)



More information about the Pypy-commit mailing list