Commit acd0fda1 authored by Antoine Pitrou's avatar Antoine Pitrou

Fix SOCK_CLOEXEC and SOCK_NONBLOCK tests on recent glibcs with old Linux kernels.

parent 8035bc5c
...@@ -13,6 +13,7 @@ import queue ...@@ -13,6 +13,7 @@ import queue
import sys import sys
import os import os
import array import array
import platform
import contextlib import contextlib
from weakref import proxy from weakref import proxy
import signal import signal
...@@ -1827,11 +1828,24 @@ class ContextManagersTest(ThreadedTCPSocketTest): ...@@ -1827,11 +1828,24 @@ class ContextManagersTest(ThreadedTCPSocketTest):
self.assertTrue(sock._closed) self.assertTrue(sock._closed)
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")
class CloexecConstantTest(unittest.TestCase): class CloexecConstantTest(unittest.TestCase):
def test_SOCK_CLOEXEC(self): def test_SOCK_CLOEXEC(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)))
s = socket.socket(socket.AF_INET, s = socket.socket(socket.AF_INET,
socket.SOCK_STREAM | socket.SOCK_CLOEXEC) socket.SOCK_STREAM | socket.SOCK_CLOEXEC)
self.assertTrue(s.type & socket.SOCK_CLOEXEC) self.assertTrue(s.type & socket.SOCK_CLOEXEC)
...@@ -1850,6 +1864,10 @@ class NonblockConstantTest(unittest.TestCase): ...@@ -1850,6 +1864,10 @@ class NonblockConstantTest(unittest.TestCase):
self.assertEqual(s.gettimeout(), None) self.assertEqual(s.gettimeout(), None)
def test_SOCK_NONBLOCK(self): def test_SOCK_NONBLOCK(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)))
# a lot of it seems silly and redundant, but I wanted to test that # a lot of it seems silly and redundant, but I wanted to test that
# changing back and forth worked ok # changing back and forth worked ok
s = socket.socket(socket.AF_INET, s = socket.socket(socket.AF_INET,
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment