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 <rep...@bugs.python.org>
<http://bugs.python.org/issue18643>
_______________________________________
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):
_______________________________________________
Python-bugs-list mailing list
Unsubscribe:
https://mail.python.org/mailman/options/python-bugs-list/archive%40mail-archive.com
