1 import httplib2 2 import mock 3 import os 4 import pickle 5 import pytest 6 import socket 7 import sys 8 import tests 9 import time 10 from 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 ) 19 def 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 26 def 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 """\ 34 METHOD /?query HTTP/1.1\r\n\ 35 Host: {host}\r\n""".format( 36 host=host 37 ).encode() 38 ), content 39 40 41 def 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 54 def 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 62 def 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 ) 77 def 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 97 def 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 111 def 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 137 def 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 # ) 146 def 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 164 def 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")) 199 def 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 ) 214 def 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 225 def 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
