Patch #1003700: Add socketpair function to socket module.

Lib/test/test_socket.py

@@ -187,6 +187,28 @@ class SocketConnectedTest(ThreadedTCPSocketTest):
         self.serv_conn = None
         ThreadedTCPSocketTest.clientTearDown(self)
 
+class SocketPairTest(unittest.TestCase, ThreadableTest):
+
+    def __init__(self, methodName='runTest'):
+        unittest.TestCase.__init__(self, methodName=methodName)
+        ThreadableTest.__init__(self)
+
+    def setUp(self):
+        self.serv, self.cli = socket.socketpair()
+
+    def tearDown(self):
+        self.serv.close()
+        self.serv = None
+
+    def clientSetUp(self):
+        pass
+
+    def clientTearDown(self):
+        self.cli.close()
+        self.cli = None
+        ThreadableTest.clientTearDown(self)
+
+    
 #######################################################################
 ## Begin Tests
 
@@ -541,6 +563,25 @@ class BasicUDPTest(ThreadedUDPSocketTest):
     def _testRecvFrom(self):
         self.cli.sendto(MSG, 0, (HOST, PORT))
 
+class BasicSocketPairTest(SocketPairTest):
+
+    def __init__(self, methodName='runTest'):
+        SocketPairTest.__init__(self, methodName=methodName)
+
+    def testRecv(self):
+        msg = self.serv.recv(1024)
+        self.assertEqual(msg, MSG)
+
+    def _testRecv(self):
+        self.cli.send(MSG)
+
+    def testSend(self):
+        self.serv.send(MSG)
+
+    def _testSend(self):
+        msg = self.cli.recv(1024)
+        self.assertEqual(msg, MSG)
+
 class NonBlockingTCPTests(ThreadedTCPSocketTest):
 
     def __init__(self, methodName='runTest'):
@@ -786,6 +827,8 @@ def test_main():
         LineBufferedFileObjectClassTestCase,
         SmallBufferedFileObjectClassTestCase
     ])
+    if hasattr(socket, "socketpair"):
+        tests.append(BasicSocketPairTest)
     test_support.run_unittest(*tests)
 
 if __name__ == "__main__":