test_selectors2.py 文件源码

def test_selector_raises_timeout_error_on_interrupt_over_time(self):
        selectors2._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, selectors2.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')