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
130
131
132
133
134
135
136
137
138
139
140
|
from __future__ import annotations
import contextlib
import typing
from http.client import ResponseNotReady
from unittest import mock
import pytest
from dummyserver.testcase import HypercornDummyServerTestCase as server
from urllib3 import HTTPConnectionPool
from urllib3.response import HTTPResponse
@pytest.fixture()
def pool() -> typing.Generator[HTTPConnectionPool]:
server.setup_class()
with HTTPConnectionPool(server.host, server.port) as pool:
yield pool
server.teardown_class()
def test_returns_urllib3_HTTPResponse(pool: HTTPConnectionPool) -> None:
with contextlib.closing(pool._get_conn()) as conn:
conn.request("GET", "/")
response = conn.getresponse()
assert isinstance(response, HTTPResponse)
@mock.patch("urllib3.connection.sys.audit")
def test_audit_event(audit_mock: mock.Mock, pool: HTTPConnectionPool) -> None:
with contextlib.closing(pool._get_conn()) as conn:
conn.request("GET", "/")
audit_mock.assert_any_call("http.client.connect", conn, conn.host, conn.port)
# Ensure the event is raised only once.
connect_events = [
call
for call in audit_mock.mock_calls
if call.args[0] == "http.client.connect"
]
assert len(connect_events) == 1
def test_does_not_release_conn(pool: HTTPConnectionPool) -> None:
with contextlib.closing(pool._get_conn()) as conn:
conn.request("GET", "/")
response = conn.getresponse()
response.release_conn()
assert pool.pool.qsize() == 0 # type: ignore[union-attr]
def test_releases_conn(pool: HTTPConnectionPool) -> None:
with contextlib.closing(pool._get_conn()) as conn:
conn.request("GET", "/")
response = conn.getresponse()
# If these variables are set by the pool
# then the response can release the connection
# back into the pool.
response._pool = pool # type: ignore[attr-defined]
response._connection = conn # type: ignore[attr-defined]
response.release_conn()
assert pool.pool.qsize() == 1 # type: ignore[union-attr]
def test_double_getresponse(pool: HTTPConnectionPool) -> None:
with contextlib.closing(pool._get_conn()) as conn:
conn.request("GET", "/")
_ = conn.getresponse()
# Calling getrepsonse() twice should cause an error
with pytest.raises(ResponseNotReady):
conn.getresponse()
def test_connection_state_properties(pool: HTTPConnectionPool) -> None:
conn = pool._get_conn()
assert conn.is_closed is True
assert conn.is_connected is False
assert conn.has_connected_to_proxy is False
assert conn.is_verified is False
assert conn.proxy_is_verified is None
conn.connect()
assert conn.is_closed is False
assert conn.is_connected is True
assert conn.has_connected_to_proxy is False
assert conn.is_verified is False
assert conn.proxy_is_verified is None
conn.request("GET", "/")
resp = conn.getresponse()
assert resp.status == 200
conn.close()
assert conn.is_closed is True
assert conn.is_connected is False
assert conn.has_connected_to_proxy is False
assert conn.is_verified is False
assert conn.proxy_is_verified is None
def test_set_tunnel_is_reset(pool: HTTPConnectionPool) -> None:
conn = pool._get_conn()
assert conn.is_closed is True
assert conn.is_connected is False
assert conn.has_connected_to_proxy is False
assert conn.is_verified is False
assert conn.proxy_is_verified is None
conn.set_tunnel(host="host", port=8080, scheme="http")
assert conn._tunnel_host == "host" # type: ignore[attr-defined]
assert conn._tunnel_port == 8080 # type: ignore[attr-defined]
assert conn._tunnel_scheme == "http" # type: ignore[attr-defined]
conn.close()
assert conn._tunnel_host is None # type: ignore[attr-defined]
assert conn._tunnel_port is None # type: ignore[attr-defined]
assert conn._tunnel_scheme is None # type: ignore[attr-defined]
def test_invalid_tunnel_scheme(pool: HTTPConnectionPool) -> None:
conn = pool._get_conn()
with pytest.raises(ValueError) as e:
conn.set_tunnel(host="host", port=8080, scheme="socks")
assert (
str(e.value)
== "Invalid proxy scheme for tunneling: 'socks', must be either 'http' or 'https'"
)
|
