130 lines
4.2 KiB
Python
130 lines
4.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Tests for the SocketOptionsAdapter and TCPKeepAliveAdapter."""
|
|
import contextlib
|
|
import platform
|
|
import socket
|
|
import sys
|
|
|
|
import pytest
|
|
try:
|
|
from unittest import mock
|
|
except ImportError:
|
|
import mock
|
|
import requests
|
|
from requests_toolbelt._compat import poolmanager
|
|
|
|
from requests_toolbelt.adapters import socket_options
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def remove_keepidle():
|
|
"""A context manager to remove TCP_KEEPIDLE from socket."""
|
|
TCP_KEEPIDLE = getattr(socket, 'TCP_KEEPIDLE', None)
|
|
if TCP_KEEPIDLE is not None:
|
|
del socket.TCP_KEEPIDLE
|
|
|
|
yield
|
|
|
|
if TCP_KEEPIDLE is not None:
|
|
socket.TCP_KEEPIDLE = TCP_KEEPIDLE
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def set_keepidle(value):
|
|
"""A context manager to set TCP_KEEPALIVE on socket always."""
|
|
TCP_KEEPIDLE = getattr(socket, 'TCP_KEEPIDLE', None)
|
|
socket.TCP_KEEPIDLE = value
|
|
|
|
yield
|
|
|
|
if TCP_KEEPIDLE is not None:
|
|
socket.TCP_KEEPIDLE = TCP_KEEPIDLE
|
|
else:
|
|
del socket.TCP_KEEPIDLE
|
|
|
|
|
|
@mock.patch.object(requests, '__build__', 0x020500)
|
|
@mock.patch.object(poolmanager, 'PoolManager')
|
|
def test_options_passing_on_newer_requests(PoolManager):
|
|
"""Show that options are passed for a new enough version of requests."""
|
|
fake_opts = [('test', 'options', 'fake')]
|
|
adapter = socket_options.SocketOptionsAdapter(
|
|
socket_options=fake_opts,
|
|
pool_connections=10,
|
|
pool_maxsize=5,
|
|
pool_block=True,
|
|
)
|
|
PoolManager.assert_called_once_with(
|
|
num_pools=10, maxsize=5, block=True,
|
|
socket_options=fake_opts
|
|
)
|
|
assert adapter.socket_options == fake_opts
|
|
|
|
|
|
@mock.patch.object(requests, '__build__', 0x020300)
|
|
@mock.patch.object(poolmanager, 'PoolManager')
|
|
def test_options_not_passed_on_older_requests(PoolManager):
|
|
"""Show that options are not passed for older versions of requests."""
|
|
fake_opts = [('test', 'options', 'fake')]
|
|
socket_options.SocketOptionsAdapter(
|
|
socket_options=fake_opts,
|
|
pool_connections=10,
|
|
pool_maxsize=5,
|
|
pool_block=True,
|
|
)
|
|
assert PoolManager.called is False
|
|
|
|
|
|
@pytest.mark.xfail(sys.version_info.major == 2 and platform.system() == "Windows",
|
|
reason="Windows does not have TCP_KEEPINTVL in Python 2")
|
|
@mock.patch.object(requests, '__build__', 0x020500)
|
|
@mock.patch.object(poolmanager, 'PoolManager')
|
|
def test_keep_alive_on_newer_requests_no_idle(PoolManager):
|
|
"""Show that options are generated correctly from kwargs."""
|
|
socket_opts = [
|
|
(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1),
|
|
(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
|
|
(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 10),
|
|
(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 10),
|
|
]
|
|
with remove_keepidle():
|
|
adapter = socket_options.TCPKeepAliveAdapter(
|
|
idle=30, interval=10, count=10,
|
|
pool_connections=10,
|
|
pool_maxsize=5,
|
|
pool_block=True,
|
|
)
|
|
PoolManager.assert_called_once_with(
|
|
num_pools=10, maxsize=5, block=True,
|
|
socket_options=socket_opts
|
|
)
|
|
assert adapter.socket_options == socket_opts
|
|
|
|
|
|
@pytest.mark.xfail(sys.version_info.major == 2 and platform.system() == "Windows",
|
|
reason="Windows does not have TCP_KEEPINTVL in Python 2")
|
|
@mock.patch.object(requests, '__build__', 0x020500)
|
|
@mock.patch.object(poolmanager, 'PoolManager')
|
|
def test_keep_alive_on_newer_requests_with_idle(PoolManager):
|
|
"""Show that options are generated correctly from kwargs with KEEPIDLE."""
|
|
with set_keepidle(3000):
|
|
socket_opts = [
|
|
(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1),
|
|
(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
|
|
(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 10),
|
|
(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 10),
|
|
(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 30),
|
|
]
|
|
adapter = socket_options.TCPKeepAliveAdapter(
|
|
idle=30, interval=10, count=10,
|
|
pool_connections=10,
|
|
pool_maxsize=5,
|
|
pool_block=True,
|
|
)
|
|
|
|
PoolManager.assert_called_once_with(
|
|
num_pools=10, maxsize=5, block=True,
|
|
socket_options=socket_opts
|
|
)
|
|
assert adapter.socket_options == socket_opts
|