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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
|
import httplib2
import mock
import os
import pickle
import pytest
import socket
import sys
import tests
import time
from six.moves import urllib
@pytest.mark.skipif(
sys.version_info <= (3,),
reason=(
"TODO: httplib2._convert_byte_str was defined only in python3 code " "version"
),
)
def test_convert_byte_str():
with tests.assert_raises(TypeError):
httplib2._convert_byte_str(4)
assert httplib2._convert_byte_str(b"Hello") == "Hello"
assert httplib2._convert_byte_str("World") == "World"
def test_reflect():
http = httplib2.Http()
with tests.server_reflect() as uri:
response, content = http.request(uri + "?query", "METHOD")
assert response.status == 200
host = urllib.parse.urlparse(uri).netloc
assert content.startswith(
"""\
METHOD /?query HTTP/1.1\r\n\
Host: {host}\r\n""".format(
host=host
).encode()
), content
def test_pickle_http():
http = httplib2.Http(cache=tests.get_cache_path())
new_http = pickle.loads(pickle.dumps(http))
assert tuple(sorted(new_http.__dict__)) == tuple(sorted(http.__dict__))
assert new_http.credentials.credentials == http.credentials.credentials
assert new_http.certificates.credentials == http.certificates.credentials
assert new_http.cache.cache == http.cache.cache
for key in new_http.__dict__:
if key not in ("cache", "certificates", "credentials"):
assert getattr(new_http, key) == getattr(http, key)
def test_pickle_http_with_connection():
http = httplib2.Http()
http.request("http://random-domain:81/", connection_type=tests.MockHTTPConnection)
new_http = pickle.loads(pickle.dumps(http))
assert tuple(http.connections) == ("http:random-domain:81",)
assert new_http.connections == {}
def test_pickle_custom_request_http():
http = httplib2.Http()
http.request = lambda: None
http.request.dummy_attr = "dummy_value"
new_http = pickle.loads(pickle.dumps(http))
assert getattr(new_http.request, "dummy_attr", None) is None
@pytest.mark.xfail(
sys.version_info >= (3,),
reason=(
"FIXME: for unknown reason global timeout test fails in Python3 "
"with response 200"
),
)
def test_timeout_global():
def handler(request):
time.sleep(0.5)
return tests.http_response_bytes()
try:
socket.setdefaulttimeout(0.1)
except Exception:
pytest.skip("cannot set global socket timeout")
try:
http = httplib2.Http()
http.force_exception_to_status_code = True
with tests.server_request(handler) as uri:
response, content = http.request(uri)
assert response.status == 408
assert response.reason.startswith("Request Timeout")
finally:
socket.setdefaulttimeout(None)
def test_timeout_individual():
def handler(request):
time.sleep(0.5)
return tests.http_response_bytes()
http = httplib2.Http(timeout=0.1)
http.force_exception_to_status_code = True
with tests.server_request(handler) as uri:
response, content = http.request(uri)
assert response.status == 408
assert response.reason.startswith("Request Timeout")
def test_timeout_subsequent():
class Handler(object):
number = 0
@classmethod
def handle(cls, request):
# request.number is always 1 because of
# the new socket connection each time
cls.number += 1
if cls.number % 2 != 0:
time.sleep(0.6)
return tests.http_response_bytes(status=500)
return tests.http_response_bytes(status=200)
http = httplib2.Http(timeout=0.5)
http.force_exception_to_status_code = True
with tests.server_request(Handler.handle, request_count=2) as uri:
response, _ = http.request(uri)
assert response.status == 408
assert response.reason.startswith("Request Timeout")
response, _ = http.request(uri)
assert response.status == 200
def test_timeout_https():
c = httplib2.HTTPSConnectionWithTimeout("localhost", 80, timeout=47)
assert 47 == c.timeout
# @pytest.mark.xfail(
# sys.version_info >= (3,),
# reason='[py3] last request should open new connection, but client does not realize socket was closed by server',
# )
def test_connection_close():
http = httplib2.Http()
g = []
def handler(request):
g.append(request.number)
return tests.http_response_bytes(proto="HTTP/1.1")
with tests.server_request(handler, request_count=3) as uri:
http.request(uri, "GET") # conn1 req1
for c in http.connections.values():
assert c.sock is not None
http.request(uri, "GET", headers={"connection": "close"})
time.sleep(0.7)
http.request(uri, "GET") # conn2 req1
assert g == [1, 2, 1]
def test_get_end2end_headers():
# one end to end header
response = {"content-type": "application/atom+xml", "te": "deflate"}
end2end = httplib2._get_end2end_headers(response)
assert "content-type" in end2end
assert "te" not in end2end
assert "connection" not in end2end
# one end to end header that gets eliminated
response = {
"connection": "content-type",
"content-type": "application/atom+xml",
"te": "deflate",
}
end2end = httplib2._get_end2end_headers(response)
assert "content-type" not in end2end
assert "te" not in end2end
assert "connection" not in end2end
# Degenerate case of no headers
response = {}
end2end = httplib2._get_end2end_headers(response)
assert len(end2end) == 0
# Degenerate case of connection referrring to a header not passed in
response = {"connection": "content-type"}
end2end = httplib2._get_end2end_headers(response)
assert len(end2end) == 0
@pytest.mark.xfail(
os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"),
reason="FIXME: fail on Travis py27 and pypy, works elsewhere",
)
@pytest.mark.parametrize("scheme", ("http", "https"))
def test_ipv6(scheme):
# Even if IPv6 isn't installed on a machine it should just raise socket.error
uri = "{scheme}://[::1]:1/".format(scheme=scheme)
try:
httplib2.Http(timeout=0.1).request(uri)
except socket.gaierror:
assert False, "should get the address family right for IPv6"
except socket.error:
pass
@pytest.mark.parametrize(
"conn_type",
(httplib2.HTTPConnectionWithTimeout, httplib2.HTTPSConnectionWithTimeout),
)
def test_connection_proxy_info_attribute_error(conn_type):
# HTTPConnectionWithTimeout did not initialize its .proxy_info attribute
# https://github.com/httplib2/httplib2/pull/97
# Thanks to Joseph Ryan https://github.com/germanjoey
conn = conn_type("no-such-hostname.", 80)
# TODO: replace mock with dummy local server
with tests.assert_raises(socket.gaierror):
with mock.patch("socket.socket.connect", side_effect=socket.gaierror):
conn.request("GET", "/")
def test_http_443_forced_https():
http = httplib2.Http()
http.force_exception_to_status_code = True
uri = "http://localhost:443/"
# sorry, using internal structure of Http to check chosen scheme
with mock.patch("httplib2.Http._request") as m:
http.request(uri)
assert len(m.call_args) > 0, "expected Http._request() call"
conn = m.call_args[0][0]
assert isinstance(conn, httplib2.HTTPConnectionWithTimeout)
|
