1import httplib2
2import mock
3import os
4import pickle
5import pytest
6import socket
7import sys
8import tests
9import time
10from six.moves import urllib
11
12
13@pytest.mark.skipif(
14    sys.version_info <= (3,),
15    reason=(
16        "TODO: httplib2._convert_byte_str was defined only in python3 code " "version"
17    ),
18)
19def test_convert_byte_str():
20    with tests.assert_raises(TypeError):
21        httplib2._convert_byte_str(4)
22    assert httplib2._convert_byte_str(b"Hello") == "Hello"
23    assert httplib2._convert_byte_str("World") == "World"
24
25
26def test_reflect():
27    http = httplib2.Http()
28    with tests.server_reflect() as uri:
29        response, content = http.request(uri + "?query", "METHOD")
30    assert response.status == 200
31    host = urllib.parse.urlparse(uri).netloc
32    assert content.startswith(
33        """\
34METHOD /?query HTTP/1.1\r\n\
35Host: {host}\r\n""".format(
36            host=host
37        ).encode()
38    ), content
39
40
41def test_pickle_http():
42    http = httplib2.Http(cache=tests.get_cache_path())
43    new_http = pickle.loads(pickle.dumps(http))
44
45    assert tuple(sorted(new_http.__dict__)) == tuple(sorted(http.__dict__))
46    assert new_http.credentials.credentials == http.credentials.credentials
47    assert new_http.certificates.credentials == http.certificates.credentials
48    assert new_http.cache.cache == http.cache.cache
49    for key in new_http.__dict__:
50        if key not in ("cache", "certificates", "credentials"):
51            assert getattr(new_http, key) == getattr(http, key)
52
53
54def test_pickle_http_with_connection():
55    http = httplib2.Http()
56    http.request("http://random-domain:81/", connection_type=tests.MockHTTPConnection)
57    new_http = pickle.loads(pickle.dumps(http))
58    assert tuple(http.connections) == ("http:random-domain:81",)
59    assert new_http.connections == {}
60
61
62def test_pickle_custom_request_http():
63    http = httplib2.Http()
64    http.request = lambda: None
65    http.request.dummy_attr = "dummy_value"
66    new_http = pickle.loads(pickle.dumps(http))
67    assert getattr(new_http.request, "dummy_attr", None) is None
68
69
70@pytest.mark.xfail(
71    sys.version_info >= (3,),
72    reason=(
73        "FIXME: for unknown reason global timeout test fails in Python3 "
74        "with response 200"
75    ),
76)
77def test_timeout_global():
78    def handler(request):
79        time.sleep(0.5)
80        return tests.http_response_bytes()
81
82    try:
83        socket.setdefaulttimeout(0.1)
84    except Exception:
85        pytest.skip("cannot set global socket timeout")
86    try:
87        http = httplib2.Http()
88        http.force_exception_to_status_code = True
89        with tests.server_request(handler) as uri:
90            response, content = http.request(uri)
91            assert response.status == 408
92            assert response.reason.startswith("Request Timeout")
93    finally:
94        socket.setdefaulttimeout(None)
95
96
97def test_timeout_individual():
98    def handler(request):
99        time.sleep(0.5)
100        return tests.http_response_bytes()
101
102    http = httplib2.Http(timeout=0.1)
103    http.force_exception_to_status_code = True
104
105    with tests.server_request(handler) as uri:
106        response, content = http.request(uri)
107        assert response.status == 408
108        assert response.reason.startswith("Request Timeout")
109
110
111def test_timeout_subsequent():
112    class Handler(object):
113        number = 0
114
115        @classmethod
116        def handle(cls, request):
117            # request.number is always 1 because of
118            # the new socket connection each time
119            cls.number += 1
120            if cls.number % 2 != 0:
121                time.sleep(0.6)
122                return tests.http_response_bytes(status=500)
123            return tests.http_response_bytes(status=200)
124
125    http = httplib2.Http(timeout=0.5)
126    http.force_exception_to_status_code = True
127
128    with tests.server_request(Handler.handle, request_count=2) as uri:
129        response, _ = http.request(uri)
130        assert response.status == 408
131        assert response.reason.startswith("Request Timeout")
132
133        response, _ = http.request(uri)
134        assert response.status == 200
135
136
137def test_timeout_https():
138    c = httplib2.HTTPSConnectionWithTimeout("localhost", 80, timeout=47)
139    assert 47 == c.timeout
140
141
142# @pytest.mark.xfail(
143#     sys.version_info >= (3,),
144#     reason='[py3] last request should open new connection, but client does not realize socket was closed by server',
145# )
146def test_connection_close():
147    http = httplib2.Http()
148    g = []
149
150    def handler(request):
151        g.append(request.number)
152        return tests.http_response_bytes(proto="HTTP/1.1")
153
154    with tests.server_request(handler, request_count=3) as uri:
155        http.request(uri, "GET")  # conn1 req1
156        for c in http.connections.values():
157            assert c.sock is not None
158        http.request(uri, "GET", headers={"connection": "close"})
159        time.sleep(0.7)
160        http.request(uri, "GET")  # conn2 req1
161    assert g == [1, 2, 1]
162
163
164def test_get_end2end_headers():
165    # one end to end header
166    response = {"content-type": "application/atom+xml", "te": "deflate"}
167    end2end = httplib2._get_end2end_headers(response)
168    assert "content-type" in end2end
169    assert "te" not in end2end
170    assert "connection" not in end2end
171
172    # one end to end header that gets eliminated
173    response = {
174        "connection": "content-type",
175        "content-type": "application/atom+xml",
176        "te": "deflate",
177    }
178    end2end = httplib2._get_end2end_headers(response)
179    assert "content-type" not in end2end
180    assert "te" not in end2end
181    assert "connection" not in end2end
182
183    # Degenerate case of no headers
184    response = {}
185    end2end = httplib2._get_end2end_headers(response)
186    assert len(end2end) == 0
187
188    # Degenerate case of connection referrring to a header not passed in
189    response = {"connection": "content-type"}
190    end2end = httplib2._get_end2end_headers(response)
191    assert len(end2end) == 0
192
193
194@pytest.mark.xfail(
195    os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"),
196    reason="FIXME: fail on Travis py27 and pypy, works elsewhere",
197)
198@pytest.mark.parametrize("scheme", ("http", "https"))
199def test_ipv6(scheme):
200    # Even if IPv6 isn't installed on a machine it should just raise socket.error
201    uri = "{scheme}://[::1]:1/".format(scheme=scheme)
202    try:
203        httplib2.Http(timeout=0.1).request(uri)
204    except socket.gaierror:
205        assert False, "should get the address family right for IPv6"
206    except socket.error:
207        pass
208
209
210@pytest.mark.parametrize(
211    "conn_type",
212    (httplib2.HTTPConnectionWithTimeout, httplib2.HTTPSConnectionWithTimeout),
213)
214def test_connection_proxy_info_attribute_error(conn_type):
215    # HTTPConnectionWithTimeout did not initialize its .proxy_info attribute
216    # https://github.com/httplib2/httplib2/pull/97
217    # Thanks to Joseph Ryan https://github.com/germanjoey
218    conn = conn_type("no-such-hostname.", 80)
219    # TODO: replace mock with dummy local server
220    with tests.assert_raises(socket.gaierror):
221        with mock.patch("socket.socket.connect", side_effect=socket.gaierror):
222            conn.request("GET", "/")
223
224
225def test_http_443_forced_https():
226    http = httplib2.Http()
227    http.force_exception_to_status_code = True
228    uri = "http://localhost:443/"
229    # sorry, using internal structure of Http to check chosen scheme
230    with mock.patch("httplib2.Http._request") as m:
231        http.request(uri)
232        assert len(m.call_args) > 0, "expected Http._request() call"
233        conn = m.call_args[0][0]
234        assert isinstance(conn, httplib2.HTTPConnectionWithTimeout)
235
236
237def test_close():
238    http = httplib2.Http()
239    assert len(http.connections) == 0
240    with tests.server_const_http() as uri:
241        http.request(uri)
242        assert len(http.connections) == 1
243        http.close()
244        assert len(http.connections) == 0
245
246
247def test_connect_exception_type():
248    # This autoformatting PR actually changed the behavior of error handling:
249    # https://github.com/httplib2/httplib2/pull/105/files#diff-c6669c781a2dee1b2d2671cab4e21c66L985
250    # potentially changing the type of the error raised by connect()
251    # https://github.com/httplib2/httplib2/pull/150
252    http = httplib2.Http()
253    with mock.patch("httplib2.socket.socket.connect", side_effect=socket.timeout("foo")):
254        with tests.assert_raises(socket.timeout):
255            http.request(tests.DUMMY_URL)
256