mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-09-26 18:29:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Also guard other SOCK_NONBLOCK test

Browse source
This commit is contained in:
Antoine Pitrou 2010-10-23 17:05:13 +00:00
parent 75d8c5cea2
commit 1cae8b58eb
1 changed files with 13 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

22
Lib/test/test_socket.py
View file

@ -35,6 +35,14 @@ def try_address(host, port=0, family=socket.AF_INET):
sock.close() sock.close()
return True return True
def linux_version():
try:
# platform.release() is something like '2.6.33.7-desktop-2mnb'
version_string = platform.release().split('-')[0]
return tuple(map(int, version_string.split('.')))
except ValueError:
return 0, 0, 0
HOST = support.HOST HOST = support.HOST
MSG = 'Michael Gilfix was here\u1234\r\n'.encode('utf8') ## test unicode string and carriage return MSG = 'Michael Gilfix was here\u1234\r\n'.encode('utf8') ## test unicode string and carriage return
SUPPORTS_IPV6 = socket.has_ipv6 and try_address('::1', family=socket.AF_INET6) SUPPORTS_IPV6 = socket.has_ipv6 and try_address('::1', family=socket.AF_INET6)
@ -911,10 +919,14 @@ class NonBlockingTCPTests(ThreadedTCPSocketTest):
if hasattr(socket, "SOCK_NONBLOCK"): if hasattr(socket, "SOCK_NONBLOCK"):
def testInitNonBlocking(self): def testInitNonBlocking(self):
v = linux_version()
if v < (2, 6, 28):
self.skipTest("Linux kernel 2.6.28 or higher required, not %s"
% ".".join(map(str, v)))
# reinit server socket # reinit server socket
self.serv.close() self.serv.close()
self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM | self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM |
socket.SOCK_NONBLOCK) socket.SOCK_NONBLOCK)
self.port = support.bind_port(self.serv) self.port = support.bind_port(self.serv)
self.serv.listen(1) self.serv.listen(1)
# actual testing # actual testing
@ -1829,14 +1841,6 @@ class ContextManagersTest(ThreadedTCPSocketTest):
self.assertRaises(socket.error, sock.sendall, b'foo') self.assertRaises(socket.error, sock.sendall, b'foo')
def linux_version():
try:
# platform.release() is something like '2.6.33.7-desktop-2mnb'
version_string = platform.release().split('-')[0]
return tuple(map(int, version_string.split('.')))
except ValueError:
return 0, 0, 0
@unittest.skipUnless(hasattr(socket, "SOCK_CLOEXEC"), @unittest.skipUnless(hasattr(socket, "SOCK_CLOEXEC"),
"SOCK_CLOEXEC not defined") "SOCK_CLOEXEC not defined")
@unittest.skipUnless(fcntl, "module fcntl not available") @unittest.skipUnless(fcntl, "module fcntl not available")