Add tests for _syscall_wrapper on Python <3.5
MODIFY
test/__init__.py
test/__init__.py
@@ -49,6 +49,18 @@ def onlyPy26OrOlder(test):
return wrapper
+def onlyPy34OrOlder(test):
+ """Skips this test unless you are on Python2.6.x or earlier."""
+
+ @functools.wraps(test)
+ def wrapper(*args, **kwargs):
+ msg = "{name} only runs on Python2.6.x or older".format(name=test.__name__)
+ if sys.version_info >= (3, 4):
+ pytest.skip(msg)
+ return test(*args, **kwargs)
+ return wrapper
+
+
def onlyPy27OrNewer(test):
"""Skips this test unless you are on Python 2.7.x or later."""

MODIFY
test/test_selectors.py
test/test_selectors.py
@@ -2,11 +2,13 @@ from __future__ import with_statement
import errno
import os
import psutil
+import mock
import select
import signal
import sys
import time
import threading
+from . import onlyPy34OrOlder
try: # Python 2.6 unittest module doesn't have skip decorators.
from unittest import skipIf, skipUnless
@@ -746,6 +748,78 @@ class TestUniqueSelectScenarios(BaseSelectorTestCase):
selector = self.make_selector()
self.assertIsInstance(selector, selectors.SelectSelector)
+ @onlyPy34OrOlder
+ def test_selector_raises_timeout_error_on_interrupt_over_time(self):
+ selectors._DEFAULT_SELECTOR = None
+
+ mock_socket = mock.Mock()
+ mock_socket.fileno.return_value = 1
+
+ def slow_interrupting_select(*args, **kwargs):
+ time.sleep(0.2)
+ error = OSError()
+ error.errno = errno.EINTR
+ raise error
+
+ patch_select_module(self, select=slow_interrupting_select)
+
+ selector = self.make_selector()
+ selector.register(mock_socket, selectors.EVENT_READ)
+
+ try:
+ selector.select(timeout=0.1)
+ except OSError as e:
+ self.assertEqual(e.errno, errno.ETIMEDOUT)
+ else:
+ self.fail('Didn\'t raise an OSError')
+
+ @onlyPy34OrOlder
+ def test_timeout_is_recalculated_after_interrupt(self):
+ selectors._DEFAULT_SELECTOR = None
+
+ mock_socket = mock.Mock()
+ mock_socket.fileno.return_value = 1
+
+ class InterruptingSelect(object):
+ """ Helper object that imitates a select that interrupts
+ after sleeping some time then returns a result. """
+ def __init__(self):
+ self.call_count = 0
+ self.calls = []
+
+ def select(self, *args, **kwargs):
+ self.calls.append((args, kwargs))
+ self.call_count += 1
+ if self.call_count == 1:
+ time.sleep(0.1)
+ error = OSError()
+ error.errno = errno.EINTR
+ raise error
+ else:
+ return [1], [], []
+
+ mock_select = InterruptingSelect()
+
+ patch_select_module(self, select=mock_select.select)
+
+ selector = self.make_selector()
+ selector.register(mock_socket, selectors.EVENT_READ)
+
+ result = selector.select(timeout=1.0)
+
+ # Make sure the mocked call actually completed correctly.
+ self.assertEqual(len(result), 1)
+ self.assertEqual(result[0][0].fileobj, mock_socket)
+ self.assertEqual(result[0][1], selectors.EVENT_READ)
+
+ # There should be two calls to the mock_select.select() function
+ self.assertEqual(mock_select.call_count, 2)
+
+ # Timeout should be less in the second call.
+ # The structure of mock_select.calls is [(args, kwargs), (args, kwargs)] where
+ # args is ([r], [w], [x], timeout).
+ self.assertLess(mock_select.calls[1][0][3], mock_select.calls[0][0][3])
+
@skipUnless(hasattr(selectors, "SelectSelector"), "Platform doesn't have a SelectSelector")
class SelectSelectorTestCase(BaseSelectorTestCase):

MODIFY
urllib3/util/selectors.py
urllib3/util/selectors.py
@@ -11,9 +11,14 @@ import select
import socket
import sys
import time
-from collections import namedtuple, Mapping
+from collections import namedtuple
from ..packages.six import integer_types
+try:
+ from collections.abc import Mapping
+except ImportError:
+ from collections import Mapping
+
try:
monotonic = time.monotonic
except (AttributeError, ImportError): # Python 3.3<

Add tests for _syscall_wrapper on Python <3.5