1 from __future__ import absolute_import 2 from __future__ import division 3 from __future__ import print_function 4 5 import email.utils 6 import errno 7 import httplib2 8 import mock 9 import os 10 import pytest 11 from six.moves import http_client, urllib 12 import socket 13 import tests 14 15 DUMMY_URL = "http://127.0.0.1:1" 16 17 18 def _raise_connection_refused_exception(*args, **kwargs): 19 raise socket.error(errno.ECONNREFUSED, "Connection refused.") 20 21 22 def test_connection_type(): 23 http = httplib2.Http() 24 http.force_exception_to_status_code = False 25 response, content = http.request( 26 DUMMY_URL, connection_type=tests.MockHTTPConnection 27 ) 28 assert response["content-location"] == DUMMY_URL 29 assert content == b"the body" 30 31 32 def test_bad_status_line_retry(): 33 http = httplib2.Http() 34 old_retries = httplib2.RETRIES 35 httplib2.RETRIES = 1 36 http.force_exception_to_status_code = False 37 try: 38 response, content = http.request( 39 DUMMY_URL, connection_type=tests.MockHTTPBadStatusConnection 40 ) 41 except http_client.BadStatusLine: 42 assert tests.MockHTTPBadStatusConnection.num_calls == 2 43 httplib2.RETRIES = old_retries 44 45 46 def test_unknown_server(): 47 http = httplib2.Http() 48 http.force_exception_to_status_code = False 49 with tests.assert_raises(httplib2.ServerNotFoundError): 50 with mock.patch("socket.socket.connect", side_effect=socket.gaierror): 51 http.request("http://no-such-hostname./") 52 53 # Now test with exceptions turned off 54 http.force_exception_to_status_code = True 55 response, content = http.request("http://no-such-hostname./") 56 assert response["content-type"] == "text/plain" 57 assert content.startswith(b"Unable to find") 58 assert response.status == 400 59 60 61 @pytest.mark.skipif( 62 os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"), 63 reason="Fails on Travis py27/pypy, works elsewhere. " 64 "See https://travis-ci.org/httplib2/httplib2/jobs/408769880.", 65 ) 66 @mock.patch("socket.socket.connect", spec=True) 67 def test_connection_refused_raises_exception(mock_socket_connect): 68 mock_socket_connect.side_effect = _raise_connection_refused_exception 69 http = httplib2.Http() 70 http.force_exception_to_status_code = False 71 with tests.assert_raises(socket.error): 72 http.request(DUMMY_URL) 73 74 75 @pytest.mark.skipif( 76 os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"), 77 reason="Fails on Travis py27/pypy, works elsewhere. " 78 "See https://travis-ci.org/httplib2/httplib2/jobs/408769880.", 79 ) 80 @mock.patch("socket.socket.connect", spec=True) 81 def test_connection_refused_returns_response(mock_socket_connect): 82 mock_socket_connect.side_effect = _raise_connection_refused_exception 83 http = httplib2.Http() 84 http.force_exception_to_status_code = True 85 response, content = http.request(DUMMY_URL) 86 content = content.lower() 87 assert response["content-type"] == "text/plain" 88 assert ( 89 b"connection refused" in content 90 or b"actively refused" in content 91 or b"socket is not connected" in content 92 ) 93 assert response.status == 400 94 95 96 def test_get_iri(): 97 http = httplib2.Http() 98 query = u"?a=\N{CYRILLIC CAPITAL LETTER DJE}" 99 with tests.server_reflect() as uri: 100 response, content = http.request(uri + query, "GET") 101 assert response.status == 200 102 reflected = tests.HttpRequest.from_bytes(content) 103 assert reflected.uri == "/?a=%D0%82" 104 105 106 def test_get_is_default_method(): 107 # Test that GET is the default method 108 http = httplib2.Http() 109 with tests.server_reflect() as uri: 110 response, content = http.request(uri) 111 assert response.status == 200 112 reflected = tests.HttpRequest.from_bytes(content) 113 assert reflected.method == "GET" 114 115 116 def test_different_methods(): 117 # Test that all methods can be used 118 http = httplib2.Http() 119 methods = ["GET", "PUT", "DELETE", "POST", "unknown"] 120 with tests.server_reflect(request_count=len(methods)) as uri: 121 for method in methods: 122 response, content = http.request(uri, method, body=b" ") 123 assert response.status == 200 124 reflected = tests.HttpRequest.from_bytes(content) 125 assert reflected.method == method 126 127 128 def test_head_read(): 129 # Test that we don't try to read the response of a HEAD request 130 # since httplib blocks response.read() for HEAD requests. 131 http = httplib2.Http() 132 respond_with = b"HTTP/1.0 200 OK\r\ncontent-length: " b"14\r\n\r\nnon-empty-body" 133 with tests.server_const_bytes(respond_with) as uri: 134 response, content = http.request(uri, "HEAD") 135 assert response.status == 200 136 assert content == b"" 137 138 139 def test_get_no_cache(): 140 # Test that can do a GET w/o the cache turned on. 141 http = httplib2.Http() 142 with tests.server_const_http() as uri: 143 response, content = http.request(uri, "GET") 144 assert response.status == 200 145 assert response.previous is None 146 147 148 def test_user_agent(): 149 # Test that we provide a default user-agent 150 http = httplib2.Http() 151 with tests.server_reflect() as uri: 152 response, content = http.request(uri, "GET") 153 assert response.status == 200 154 reflected = tests.HttpRequest.from_bytes(content) 155 assert reflected.headers.get("user-agent", "").startswith("Python-httplib2/") 156 157 158 def test_user_agent_non_default(): 159 # Test that the default user-agent can be over-ridden 160 http = httplib2.Http() 161 with tests.server_reflect() as uri: 162 response, content = http.request(uri, "GET", headers={"User-Agent": "fred/1.0"}) 163 assert response.status == 200 164 reflected = tests.HttpRequest.from_bytes(content) 165 assert reflected.headers.get("user-agent") == "fred/1.0" 166 167 168 def test_get_300_with_location(): 169 # Test the we automatically follow 300 redirects if a Location: header is provided 170 http = httplib2.Http() 171 final_content = b"This is the final destination.\n" 172 routes = { 173 "/final": tests.http_response_bytes(body=final_content), 174 "": tests.http_response_bytes( 175 status="300 Multiple Choices", headers={"location": "/final"} 176 ), 177 } 178 with tests.server_route(routes, request_count=2) as uri: 179 response, content = http.request(uri, "GET") 180 assert response.status == 200 181 assert content == final_content 182 assert response.previous.status == 300 183 assert not response.previous.fromcache 184 185 # Confirm that the intermediate 300 is not cached 186 with tests.server_route(routes, request_count=2) as uri: 187 response, content = http.request(uri, "GET") 188 assert response.status == 200 189 assert content == final_content 190 assert response.previous.status == 300 191 assert not response.previous.fromcache 192 193 194 def test_get_300_with_location_noredirect(): 195 # Test the we automatically follow 300 redirects if a Location: header is provided 196 http = httplib2.Http() 197 http.follow_redirects = False 198 response = tests.http_response_bytes( 199 status="300 Multiple Choices", 200 headers={"location": "/final"}, 201 body=b"redirect body", 202 ) 203 with tests.server_const_bytes(response) as uri: 204 response, content = http.request(uri, "GET") 205 assert response.status == 300 206 207 208 def test_get_300_without_location(): 209 # Not giving a Location: header in a 300 response is acceptable 210 # In which case we just return the 300 response 211 http = httplib2.Http() 212 with tests.server_const_http( 213 status="300 Multiple Choices", body=b"redirect body" 214 ) as uri: 215 response, content = http.request(uri, "GET") 216 assert response.status == 300 217 assert response.previous is None 218 assert content == b"redirect body" 219 220 221 def test_get_301(): 222 # Test that we automatically follow 301 redirects 223 # and that we cache the 301 response 224 http = httplib2.Http(cache=tests.get_cache_path()) 225 destination = "" 226 routes = { 227 "/final": tests.http_response_bytes(body=b"This is the final destination.\n"), 228 "": tests.http_response_bytes( 229 status="301 Now where did I leave that URL", 230 headers={"location": "/final"}, 231 body=b"redirect body", 232 ), 233 } 234 with tests.server_route(routes, request_count=3) as uri: 235 destination = urllib.parse.urljoin(uri, "/final") 236 response1, content1 = http.request(uri, "GET") 237 response2, content2 = http.request(uri, "GET") 238 assert response1.status == 200 239 assert "content-location" in response2 240 assert response1["content-location"] == destination 241 assert content1 == b"This is the final destination.\n" 242 assert response1.previous.status == 301 243 assert not response1.previous.fromcache 244 245 assert response2.status == 200 246 assert response2["content-location"] == destination 247 assert content2 == b"This is the final destination.\n" 248 assert response2.previous.status == 301 249 assert response2.previous.fromcache 250 251 252 @pytest.mark.skip( 253 not os.environ.get("httplib2_test_still_run_skipped") 254 and os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"), 255 reason="FIXME: timeout on Travis py27 and pypy, works elsewhere", 256 ) 257 def test_head_301(): 258 # Test that we automatically follow 301 redirects 259 http = httplib2.Http() 260 destination = "" 261 routes = { 262 "/final": tests.http_response_bytes(body=b"This is the final destination.\n"), 263 "": tests.http_response_bytes( 264 status="301 Now where did I leave that URL", 265 headers={"location": "/final"}, 266 body=b"redirect body", 267 ), 268 } 269 with tests.server_route(routes, request_count=2) as uri: 270 destination = urllib.parse.urljoin(uri, "/final") 271 response, content = http.request(uri, "HEAD") 272 assert response.status == 200 273 assert response["content-location"] == destination 274 assert response.previous.status == 301 275 assert not response.previous.fromcache 276 277 278 @pytest.mark.xfail( 279 reason=( 280 "FIXME: 301 cache works only with follow_redirects, should work " "regardless" 281 ) 282 ) 283 def test_get_301_no_redirect(): 284 # Test that we cache the 301 response 285 http = httplib2.Http(cache=tests.get_cache_path(), timeout=0.5) 286 http.follow_redirects = False 287 response = tests.http_response_bytes( 288 status="301 Now where did I leave that URL", 289 headers={"location": "/final", "cache-control": "max-age=300"}, 290 body=b"redirect body", 291 add_date=True, 292 ) 293 with tests.server_const_bytes(response) as uri: 294 response, _ = http.request(uri, "GET") 295 assert response.status == 301 296 assert not response.fromcache 297 response, _ = http.request(uri, "GET") 298 assert response.status == 301 299 assert response.fromcache 300 301 302 def test_get_302(): 303 # Test that we automatically follow 302 redirects 304 # and that we DO NOT cache the 302 response 305 http = httplib2.Http(cache=tests.get_cache_path()) 306 second_url, final_url = "", "" 307 routes = { 308 "/final": tests.http_response_bytes(body=b"This is the final destination.\n"), 309 "/second": tests.http_response_bytes( 310 status="302 Found", headers={"location": "/final"}, body=b"second redirect" 311 ), 312 "": tests.http_response_bytes( 313 status="302 Found", headers={"location": "/second"}, body=b"redirect body" 314 ), 315 } 316 with tests.server_route(routes, request_count=7) as uri: 317 second_url = urllib.parse.urljoin(uri, "/second") 318 final_url = urllib.parse.urljoin(uri, "/final") 319 response1, content1 = http.request(second_url, "GET") 320 response2, content2 = http.request(second_url, "GET") 321 response3, content3 = http.request(uri, "GET") 322 assert response1.status == 200 323 assert response1["content-location"] == final_url 324 assert content1 == b"This is the final destination.\n" 325 assert response1.previous.status == 302 326 assert not response1.previous.fromcache 327 328 assert response2.status == 200 329 # FIXME: 330 # assert response2.fromcache 331 assert response2["content-location"] == final_url 332 assert content2 == b"This is the final destination.\n" 333 assert response2.previous.status == 302 334 assert not response2.previous.fromcache 335 assert response2.previous["content-location"] == second_url 336 337 assert response3.status == 200 338 # FIXME: 339 # assert response3.fromcache 340 assert content3 == b"This is the final destination.\n" 341 assert response3.previous.status == 302 342 assert not response3.previous.fromcache 343 344 345 def test_get_302_redirection_limit(): 346 # Test that we can set a lower redirection limit 347 # and that we raise an exception when we exceed 348 # that limit. 349 http = httplib2.Http() 350 http.force_exception_to_status_code = False 351 routes = { 352 "/second": tests.http_response_bytes( 353 status="302 Found", headers={"location": "/final"}, body=b"second redirect" 354 ), 355 "": tests.http_response_bytes( 356 status="302 Found", headers={"location": "/second"}, body=b"redirect body" 357 ), 358 } 359 with tests.server_route(routes, request_count=4) as uri: 360 try: 361 http.request(uri, "GET", redirections=1) 362 assert False, "This should not happen" 363 except httplib2.RedirectLimit: 364 pass 365 except Exception: 366 assert False, "Threw wrong kind of exception " 367 368 # Re-run the test with out the exceptions 369 http.force_exception_to_status_code = True 370 response, content = http.request(uri, "GET", redirections=1) 371 372 assert response.status == 500 373 assert response.reason.startswith("Redirected more") 374 assert response["status"] == "302" 375 assert content == b"second redirect" 376 assert response.previous is not None 377 378 379 def test_get_302_no_location(): 380 # Test that we throw an exception when we get 381 # a 302 with no Location: header. 382 http = httplib2.Http() 383 http.force_exception_to_status_code = False 384 with tests.server_const_http(status="302 Found", request_count=2) as uri: 385 try: 386 http.request(uri, "GET") 387 assert False, "Should never reach here" 388 except httplib2.RedirectMissingLocation: 389 pass 390 except Exception: 391 assert False, "Threw wrong kind of exception " 392 393 # Re-run the test with out the exceptions 394 http.force_exception_to_status_code = True 395 response, content = http.request(uri, "GET") 396 397 assert response.status == 500 398 assert response.reason.startswith("Redirected but") 399 assert "302" == response["status"] 400 assert content == b"" 401 402 403 @pytest.mark.skip( 404 not os.environ.get("httplib2_test_still_run_skipped") 405 and os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"), 406 reason="FIXME: timeout on Travis py27 and pypy, works elsewhere", 407 ) 408 def test_303(): 409 # Do a follow-up GET on a Location: header 410 # returned from a POST that gave a 303. 411 http = httplib2.Http() 412 routes = { 413 "/final": tests.make_http_reflect(), 414 "": tests.make_http_reflect( 415 status="303 See Other", headers={"location": "/final"} 416 ), 417 } 418 with tests.server_route(routes, request_count=2) as uri: 419 response, content = http.request(uri, "POST", " ") 420 assert response.status == 200 421 reflected = tests.HttpRequest.from_bytes(content) 422 assert reflected.uri == "/final" 423 assert response.previous.status == 303 424 425 # Skip follow-up GET 426 http = httplib2.Http() 427 http.follow_redirects = False 428 with tests.server_route(routes, request_count=1) as uri: 429 response, content = http.request(uri, "POST", " ") 430 assert response.status == 303 431 432 # All methods can be used 433 http = httplib2.Http() 434 cases = "DELETE GET HEAD POST PUT EVEN_NEW_ONES".split(" ") 435 with tests.server_route(routes, request_count=len(cases) * 2) as uri: 436 for method in cases: 437 response, content = http.request(uri, method, body=b"q q") 438 assert response.status == 200 439 reflected = tests.HttpRequest.from_bytes(content) 440 assert reflected.method == "GET" 441 442 443 def test_etag_used(): 444 # Test that we use ETags properly to validate our cache 445 cache_path = tests.get_cache_path() 446 http = httplib2.Http(cache=cache_path) 447 response_kwargs = dict( 448 add_date=True, 449 add_etag=True, 450 body=b"something", 451 headers={"cache-control": "public,max-age=300"}, 452 ) 453 454 def handler(request): 455 if request.headers.get("range"): 456 return tests.http_response_bytes(status=206, **response_kwargs) 457 return tests.http_response_bytes(**response_kwargs) 458 459 with tests.server_request(handler, request_count=2) as uri: 460 response, _ = http.request(uri, "GET", headers={"accept-encoding": "identity"}) 461 assert response["etag"] == '"437b930db84b8079c2dd804a71936b5f"' 462 463 http.request(uri, "GET", headers={"accept-encoding": "identity"}) 464 response, _ = http.request( 465 uri, 466 "GET", 467 headers={"accept-encoding": "identity", "cache-control": "must-revalidate"}, 468 ) 469 assert response.status == 200 470 assert response.fromcache 471 472 # TODO: API to read cache item, at least internal to tests 473 cache_file_name = os.path.join( 474 cache_path, httplib2.safename(httplib2.urlnorm(uri)[-1]) 475 ) 476 with open(cache_file_name, "r") as f: 477 status_line = f.readline() 478 assert status_line.startswith("status:") 479 480 response, content = http.request( 481 uri, "HEAD", headers={"accept-encoding": "identity"} 482 ) 483 assert response.status == 200 484 assert response.fromcache 485 486 response, content = http.request( 487 uri, "GET", headers={"accept-encoding": "identity", "range": "bytes=0-0"} 488 ) 489 assert response.status == 206 490 assert not response.fromcache 491 492 493 def test_etag_ignore(): 494 # Test that we can forcibly ignore ETags 495 http = httplib2.Http(cache=tests.get_cache_path()) 496 response_kwargs = dict(add_date=True, add_etag=True) 497 with tests.server_reflect(request_count=3, **response_kwargs) as uri: 498 response, content = http.request( 499 uri, "GET", headers={"accept-encoding": "identity"} 500 ) 501 assert response.status == 200 502 assert response["etag"] != "" 503 504 response, content = http.request( 505 uri, 506 "GET", 507 headers={"accept-encoding": "identity", "cache-control": "max-age=0"}, 508 ) 509 reflected = tests.HttpRequest.from_bytes(content) 510 assert reflected.headers.get("if-none-match") 511 512 http.ignore_etag = True 513 response, content = http.request( 514 uri, 515 "GET", 516 headers={"accept-encoding": "identity", "cache-control": "max-age=0"}, 517 ) 518 assert not response.fromcache 519 reflected = tests.HttpRequest.from_bytes(content) 520 assert not reflected.headers.get("if-none-match") 521 522 523 def test_etag_override(): 524 # Test that we can forcibly ignore ETags 525 http = httplib2.Http(cache=tests.get_cache_path()) 526 response_kwargs = dict(add_date=True, add_etag=True) 527 with tests.server_reflect(request_count=3, **response_kwargs) as uri: 528 response, _ = http.request(uri, "GET", headers={"accept-encoding": "identity"}) 529 assert response.status == 200 530 assert response["etag"] != "" 531 532 response, content = http.request( 533 uri, 534 "GET", 535 headers={"accept-encoding": "identity", "cache-control": "max-age=0"}, 536 ) 537 assert response.status == 200 538 reflected = tests.HttpRequest.from_bytes(content) 539 assert reflected.headers.get("if-none-match") 540 assert reflected.headers.get("if-none-match") != "fred" 541 542 response, content = http.request( 543 uri, 544 "GET", 545 headers={ 546 "accept-encoding": "identity", 547 "cache-control": "max-age=0", 548 "if-none-match": "fred", 549 }, 550 ) 551 assert response.status == 200 552 reflected = tests.HttpRequest.from_bytes(content) 553 assert reflected.headers.get("if-none-match") == "fred" 554 555 556 @pytest.mark.skip(reason="was commented in legacy code") 557 def test_get_304_end_to_end(): 558 pass 559 # Test that end to end headers get overwritten in the cache 560 # uri = urllib.parse.urljoin(base, "304/end2end.cgi") 561 # response, content = http.request(uri, 'GET') 562 # assertNotEqual(response['etag'], "") 563 # old_date = response['date'] 564 # time.sleep(2) 565 566 # response, content = http.request(uri, 'GET', headers = {'Cache-Control': 'max-age=0'}) 567 # # The response should be from the cache, but the Date: header should be updated. 568 # new_date = response['date'] 569 # assert new_date != old_date 570 # assert response.status == 200 571 # assert response.fromcache == True 572 573 574 def test_get_304_last_modified(): 575 # Test that we can still handle a 304 576 # by only using the last-modified cache validator. 577 http = httplib2.Http(cache=tests.get_cache_path()) 578 date = email.utils.formatdate() 579 580 def handler(read): 581 read() 582 yield tests.http_response_bytes( 583 status=200, body=b"something", headers={"date": date, "last-modified": date} 584 ) 585 586 request2 = read() 587 assert request2.headers["if-modified-since"] == date 588 yield tests.http_response_bytes(status=304) 589 590 with tests.server_yield(handler, request_count=2) as uri: 591 response, content = http.request(uri, "GET") 592 assert response.get("last-modified") == date 593 594 response, content = http.request(uri, "GET") 595 assert response.status == 200 596 assert response.fromcache 597 598 599 def test_get_307(): 600 # Test that we do follow 307 redirects but 601 # do not cache the 307 602 http = httplib2.Http(cache=tests.get_cache_path(), timeout=1) 603 r307 = tests.http_response_bytes(status=307, headers={"location": "/final"}) 604 r200 = tests.http_response_bytes( 605 status=200, 606 add_date=True, 607 body=b"final content\n", 608 headers={"cache-control": "max-age=300"}, 609 ) 610 611 with tests.server_list_http([r307, r200, r307]) as uri: 612 response, content = http.request(uri, "GET") 613 assert response.previous.status == 307 614 assert not response.previous.fromcache 615 assert response.status == 200 616 assert not response.fromcache 617 assert content == b"final content\n" 618 619 response, content = http.request(uri, "GET") 620 assert response.previous.status == 307 621 assert not response.previous.fromcache 622 assert response.status == 200 623 assert response.fromcache 624 assert content == b"final content\n" 625 626 627 def test_get_410(): 628 # Test that we pass 410's through 629 http = httplib2.Http() 630 with tests.server_const_http(status=410) as uri: 631 response, content = http.request(uri, "GET") 632 assert response.status == 410 633 634 635 def test_get_duplicate_headers(): 636 # Test that duplicate headers get concatenated via ',' 637 http = httplib2.Http() 638 response = b"""HTTP/1.0 200 OK\r\n\ 639 Link: link1\r\n\ 640 Content-Length: 7\r\n\ 641 Link: link2\r\n\r\n\ 642 content""" 643 with tests.server_const_bytes(response) as uri: 644 response, content = http.request(uri, "GET") 645 assert response.status == 200 646 assert content == b"content" 647 assert response["link"], "link1, link2" 648
