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
141
142
143
144
145
|
from __future__ import annotations
import typing
from http.client import ResponseNotReady
import pytest
from dummyserver.testcase import HTTPDummyServerTestCase as server
from urllib3 import HTTPConnectionPool
from urllib3.response import HTTPResponse
@pytest.fixture()
def pool() -> typing.Generator[HTTPConnectionPool, None, None]:
server.setup_class()
with HTTPConnectionPool(server.host, server.port) as pool:
yield pool
server.teardown_class()
def test_returns_urllib3_HTTPResponse(pool: HTTPConnectionPool) -> None:
conn = pool._get_conn()
method = "GET"
path = "/"
conn.request(method, path)
response = conn.getresponse()
assert isinstance(response, HTTPResponse)
def test_does_not_release_conn(pool: HTTPConnectionPool) -> None:
conn = pool._get_conn()
method = "GET"
path = "/"
conn.request(method, path)
response = conn.getresponse()
response.release_conn()
assert pool.pool.qsize() == 0 # type: ignore[union-attr]
def test_releases_conn(pool: HTTPConnectionPool) -> None:
conn = pool._get_conn()
assert conn is not None
method = "GET"
path = "/"
conn.request(method, path)
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:
conn = pool._get_conn()
method = "GET"
path = "/"
conn.request(method, path)
_ = 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'"
)
|
