forked from requests/toolbelt
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_socket_options_adapter.py
129 lines (110 loc) · 4.18 KB
/
test_socket_options_adapter.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
# -*- 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