[issue18643] implement socketpair() on Windows

[Charles-François Natali]

Charles-François Natali added the comment:

Here's a patch adding socketpair to test.support.

This version has been used in test_selectors for quite some time now,
and would probably be useful for other tests as well.

----------
Added file: http://bugs.python.org/file32909/socketpair-1.diff

_______________________________________
Python tracker <report at bugs.python.org>
<http://bugs.python.org/issue18643>
_______________________________________
-------------- next part --------------
diff -r a1a936a3b2f6 Lib/test/support/__init__.py
--- a/Lib/test/support/__init__.py	Fri Nov 29 18:57:47 2013 +0100
+++ b/Lib/test/support/__init__.py	Sat Nov 30 16:35:41 2013 +0100
@@ -90,6 +90,7 @@
     "is_jython", "check_impl_detail",
     # network
     "HOST", "IPV6_ENABLED", "find_unused_port", "bind_port", "open_urlresource",
+    "socketpair",
     # processes
     'temp_umask', "reap_children",
     # logging
@@ -610,6 +611,31 @@
     port = sock.getsockname()[1]
     return port
 
+if hasattr(socket, 'socketpair'):
+    socketpair = socket.socketpair
+else:
+    def socketpair(family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0):
+        """Ad-hoc socketpair implementation."""
+        if family != socket.AF_INET:
+            raise ValueError("Invalid family: {r}".format(family))
+
+        with socket.socket(family, type, proto) as l:
+            l.bind((HOST, 0))
+            l.listen(16)
+            c = socket.socket(family, type, proto)
+            try:
+                c.connect(l.getsockname())
+                caddr = c.getsockname()
+                while True:
+                    a, addr = l.accept()
+                    # check that we've got the correct client
+                    if addr == caddr:
+                        return c, a
+                    a.close()
+            except error:
+                c.close()
+                raise
+ 
 def _is_ipv6_enabled():
     """Check whether IPv6 is enabled on this host."""
     if socket.has_ipv6:
diff -r a1a936a3b2f6 Lib/test/test_support.py
--- a/Lib/test/test_support.py	Fri Nov 29 18:57:47 2013 +0100
+++ b/Lib/test/test_support.py	Sat Nov 30 16:35:41 2013 +0100
@@ -89,6 +89,17 @@
         s.listen(1)
         s.close()
 
+    def test_socketpair(self):
+        c, s = support.socketpair()
+        self.addCleanup(c.close)
+        self.addCleanup(s.close)
+        c.send(b'spam')
+        self.assertEqual(b'spam', s.recv(1024))
+        s.send(b'foo')
+        self.assertEqual(b'foo', c.recv(1024))
+        c.close()
+        self.assertFalse(s.recv(1024))
+
     # Tests for temp_dir()
 
     def test_temp_dir(self):