Autoformat py files using Black (#105)
* Autoformat all py files using Black
* Fix lint errors
* Fix indentation errors (https://travis-ci.org/httplib2/httplib2/jobs/408136309)
* Refactor three test cases and exclude them on on Travis py27/pypy
diff --git a/tests/test_http.py b/tests/test_http.py
index 29d67af..2bee6ca 100644
--- a/tests/test_http.py
+++ b/tests/test_http.py
@@ -1,22 +1,32 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
import email.utils
+import errno
import httplib2
import mock
import os
import pytest
+from six.moves import http_client, urllib
import socket
import tests
-from six.moves import http_client, urllib
+
+DUMMY_URL = "http://127.0.0.1:1"
-dummy_url = 'http://127.0.0.1:1'
+def _raise_connection_refused_exception(*args, **kwargs):
+ raise socket.error(errno.ECONNREFUSED, "Connection refused.")
def test_connection_type():
http = httplib2.Http()
http.force_exception_to_status_code = False
- response, content = http.request(dummy_url, connection_type=tests.MockHTTPConnection)
- assert response['content-location'] == dummy_url
- assert content == b'the body'
+ response, content = http.request(
+ DUMMY_URL, connection_type=tests.MockHTTPConnection
+ )
+ assert response["content-location"] == DUMMY_URL
+ assert content == b"the body"
def test_bad_status_line_retry():
@@ -25,7 +35,9 @@
httplib2.RETRIES = 1
http.force_exception_to_status_code = False
try:
- response, content = http.request(dummy_url, connection_type=tests.MockHTTPBadStatusConnection)
+ response, content = http.request(
+ DUMMY_URL, connection_type=tests.MockHTTPBadStatusConnection
+ )
except http_client.BadStatusLine:
assert tests.MockHTTPBadStatusConnection.num_calls == 2
httplib2.RETRIES = old_retries
@@ -35,39 +47,60 @@
http = httplib2.Http()
http.force_exception_to_status_code = False
with tests.assert_raises(httplib2.ServerNotFoundError):
- with mock.patch('socket.socket.connect', side_effect=socket.gaierror):
+ with mock.patch("socket.socket.connect", side_effect=socket.gaierror):
http.request("http://no-such-hostname./")
# Now test with exceptions turned off
http.force_exception_to_status_code = True
response, content = http.request("http://no-such-hostname./")
- assert response['content-type'] == 'text/plain'
+ assert response["content-type"] == "text/plain"
assert content.startswith(b"Unable to find")
assert response.status == 400
-def test_connection_refused():
+@pytest.mark.skipif(
+ os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"),
+ reason="Fails on Travis py27/pypy, works elsewhere. "
+ "See https://travis-ci.org/httplib2/httplib2/jobs/408769880.",
+)
+@mock.patch("socket.socket.connect", spec=True)
+def test_connection_refused_raises_exception(mock_socket_connect):
+ mock_socket_connect.side_effect = _raise_connection_refused_exception
http = httplib2.Http()
http.force_exception_to_status_code = False
with tests.assert_raises(socket.error):
- http.request(dummy_url)
+ http.request(DUMMY_URL)
- # Now test with exceptions turned off
+
+@pytest.mark.skipif(
+ os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"),
+ reason="Fails on Travis py27/pypy, works elsewhere. "
+ "See https://travis-ci.org/httplib2/httplib2/jobs/408769880.",
+)
+@mock.patch("socket.socket.connect", spec=True)
+def test_connection_refused_returns_response(mock_socket_connect):
+ mock_socket_connect.side_effect = _raise_connection_refused_exception
+ http = httplib2.Http()
http.force_exception_to_status_code = True
- response, content = http.request(dummy_url)
- assert response['content-type'] == 'text/plain'
- assert (b"Connection refused" in content or b"actively refused" in content)
+ response, content = http.request(DUMMY_URL)
+ content = content.lower()
+ assert response["content-type"] == "text/plain"
+ assert (
+ b"connection refused" in content
+ or b"actively refused" in content
+ or b"socket is not connected" in content
+ )
assert response.status == 400
def test_get_iri():
http = httplib2.Http()
- query = u'?a=\N{CYRILLIC CAPITAL LETTER DJE}'
+ query = u"?a=\N{CYRILLIC CAPITAL LETTER DJE}"
with tests.server_reflect() as uri:
- response, content = http.request(uri + query, 'GET')
+ response, content = http.request(uri + query, "GET")
assert response.status == 200
reflected = tests.HttpRequest.from_bytes(content)
- assert reflected.uri == '/?a=%D0%82'
+ assert reflected.uri == "/?a=%D0%82"
def test_get_is_default_method():
@@ -77,13 +110,13 @@
response, content = http.request(uri)
assert response.status == 200
reflected = tests.HttpRequest.from_bytes(content)
- assert reflected.method == 'GET'
+ assert reflected.method == "GET"
def test_different_methods():
# Test that all methods can be used
http = httplib2.Http()
- methods = ['GET', 'PUT', 'DELETE', 'POST', 'unknown']
+ methods = ["GET", "PUT", "DELETE", "POST", "unknown"]
with tests.server_reflect(request_count=len(methods)) as uri:
for method in methods:
response, content = http.request(uri, method, body=b" ")
@@ -96,9 +129,9 @@
# Test that we don't try to read the response of a HEAD request
# since httplib blocks response.read() for HEAD requests.
http = httplib2.Http()
- respond_with = b'HTTP/1.0 200 OK\r\ncontent-length: 14\r\n\r\nnon-empty-body'
+ respond_with = b"HTTP/1.0 200 OK\r\ncontent-length: " b"14\r\n\r\nnon-empty-body"
with tests.server_const_bytes(respond_with) as uri:
- response, content = http.request(uri, 'HEAD')
+ response, content = http.request(uri, "HEAD")
assert response.status == 200
assert content == b""
@@ -107,7 +140,7 @@
# Test that can do a GET w/o the cache turned on.
http = httplib2.Http()
with tests.server_const_http() as uri:
- response, content = http.request(uri, 'GET')
+ response, content = http.request(uri, "GET")
assert response.status == 200
assert response.previous is None
@@ -116,32 +149,34 @@
# Test that we provide a default user-agent
http = httplib2.Http()
with tests.server_reflect() as uri:
- response, content = http.request(uri, 'GET')
+ response, content = http.request(uri, "GET")
assert response.status == 200
reflected = tests.HttpRequest.from_bytes(content)
- assert reflected.headers.get('user-agent', '').startswith('Python-httplib2/')
+ assert reflected.headers.get("user-agent", "").startswith("Python-httplib2/")
def test_user_agent_non_default():
# Test that the default user-agent can be over-ridden
http = httplib2.Http()
with tests.server_reflect() as uri:
- response, content = http.request(uri, 'GET', headers={'User-Agent': 'fred/1.0'})
+ response, content = http.request(uri, "GET", headers={"User-Agent": "fred/1.0"})
assert response.status == 200
reflected = tests.HttpRequest.from_bytes(content)
- assert reflected.headers.get('user-agent') == 'fred/1.0'
+ assert reflected.headers.get("user-agent") == "fred/1.0"
def test_get_300_with_location():
# Test the we automatically follow 300 redirects if a Location: header is provided
http = httplib2.Http()
- final_content = b'This is the final destination.\n'
+ final_content = b"This is the final destination.\n"
routes = {
- '/final': tests.http_response_bytes(body=final_content),
- '': tests.http_response_bytes(status='300 Multiple Choices', headers={'location': '/final'}),
+ "/final": tests.http_response_bytes(body=final_content),
+ "": tests.http_response_bytes(
+ status="300 Multiple Choices", headers={"location": "/final"}
+ ),
}
with tests.server_route(routes, request_count=2) as uri:
- response, content = http.request(uri, 'GET')
+ response, content = http.request(uri, "GET")
assert response.status == 200
assert content == final_content
assert response.previous.status == 300
@@ -149,7 +184,7 @@
# Confirm that the intermediate 300 is not cached
with tests.server_route(routes, request_count=2) as uri:
- response, content = http.request(uri, 'GET')
+ response, content = http.request(uri, "GET")
assert response.status == 200
assert content == final_content
assert response.previous.status == 300
@@ -161,11 +196,12 @@
http = httplib2.Http()
http.follow_redirects = False
response = tests.http_response_bytes(
- status='300 Multiple Choices',
- headers={'location': '/final'},
- body=b'redirect body')
+ status="300 Multiple Choices",
+ headers={"location": "/final"},
+ body=b"redirect body",
+ )
with tests.server_const_bytes(response) as uri:
- response, content = http.request(uri, 'GET')
+ response, content = http.request(uri, "GET")
assert response.status == 300
@@ -173,80 +209,92 @@
# Not giving a Location: header in a 300 response is acceptable
# In which case we just return the 300 response
http = httplib2.Http()
- with tests.server_const_http(status='300 Multiple Choices', body=b'redirect body') as uri:
- response, content = http.request(uri, 'GET')
+ with tests.server_const_http(
+ status="300 Multiple Choices", body=b"redirect body"
+ ) as uri:
+ response, content = http.request(uri, "GET")
assert response.status == 300
assert response.previous is None
- assert content == b'redirect body'
+ assert content == b"redirect body"
def test_get_301():
# Test that we automatically follow 301 redirects
# and that we cache the 301 response
http = httplib2.Http(cache=tests.get_cache_path())
- destination = ''
+ destination = ""
routes = {
- '/final': tests.http_response_bytes(body=b'This is the final destination.\n'),
- '': tests.http_response_bytes(
- status='301 Now where did I leave that URL', headers={'location': '/final'}, body=b'redirect body'),
+ "/final": tests.http_response_bytes(body=b"This is the final destination.\n"),
+ "": tests.http_response_bytes(
+ status="301 Now where did I leave that URL",
+ headers={"location": "/final"},
+ body=b"redirect body",
+ ),
}
with tests.server_route(routes, request_count=3) as uri:
- destination = urllib.parse.urljoin(uri, '/final')
- response1, content1 = http.request(uri, 'GET')
- response2, content2 = http.request(uri, 'GET')
+ destination = urllib.parse.urljoin(uri, "/final")
+ response1, content1 = http.request(uri, "GET")
+ response2, content2 = http.request(uri, "GET")
assert response1.status == 200
- assert 'content-location' in response2
- assert response1['content-location'] == destination
- assert content1 == b'This is the final destination.\n'
+ assert "content-location" in response2
+ assert response1["content-location"] == destination
+ assert content1 == b"This is the final destination.\n"
assert response1.previous.status == 301
assert not response1.previous.fromcache
assert response2.status == 200
- assert response2['content-location'] == destination
- assert content2 == b'This is the final destination.\n'
+ assert response2["content-location"] == destination
+ assert content2 == b"This is the final destination.\n"
assert response2.previous.status == 301
assert response2.previous.fromcache
@pytest.mark.skip(
- not os.environ.get('httplib2_test_still_run_skipped') and
- os.environ.get('TRAVIS_PYTHON_VERSION') in ('2.7', 'pypy'),
- reason='FIXME: timeout on Travis py27 and pypy, works elsewhere',
+ not os.environ.get("httplib2_test_still_run_skipped")
+ and os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"),
+ reason="FIXME: timeout on Travis py27 and pypy, works elsewhere",
)
def test_head_301():
# Test that we automatically follow 301 redirects
http = httplib2.Http()
- destination = ''
+ destination = ""
routes = {
- '/final': tests.http_response_bytes(body=b'This is the final destination.\n'),
- '': tests.http_response_bytes(
- status='301 Now where did I leave that URL', headers={'location': '/final'}, body=b'redirect body'),
+ "/final": tests.http_response_bytes(body=b"This is the final destination.\n"),
+ "": tests.http_response_bytes(
+ status="301 Now where did I leave that URL",
+ headers={"location": "/final"},
+ body=b"redirect body",
+ ),
}
with tests.server_route(routes, request_count=2) as uri:
- destination = urllib.parse.urljoin(uri, '/final')
- response, content = http.request(uri, 'HEAD')
+ destination = urllib.parse.urljoin(uri, "/final")
+ response, content = http.request(uri, "HEAD")
assert response.status == 200
- assert response['content-location'] == destination
+ assert response["content-location"] == destination
assert response.previous.status == 301
assert not response.previous.fromcache
-@pytest.mark.xfail(reason='FIXME: 301 cache works only with follow_redirects, should work regardless')
+@pytest.mark.xfail(
+ reason=(
+ "FIXME: 301 cache works only with follow_redirects, should work " "regardless"
+ )
+)
def test_get_301_no_redirect():
# Test that we cache the 301 response
http = httplib2.Http(cache=tests.get_cache_path(), timeout=0.5)
http.follow_redirects = False
response = tests.http_response_bytes(
- status='301 Now where did I leave that URL',
- headers={'location': '/final', 'cache-control': 'max-age=300'},
- body=b'redirect body',
+ status="301 Now where did I leave that URL",
+ headers={"location": "/final", "cache-control": "max-age=300"},
+ body=b"redirect body",
add_date=True,
)
with tests.server_const_bytes(response) as uri:
- response, _ = http.request(uri, 'GET')
+ response, _ = http.request(uri, "GET")
assert response.status == 301
assert not response.fromcache
- response, _ = http.request(uri, 'GET')
+ response, _ = http.request(uri, "GET")
assert response.status == 301
assert response.fromcache
@@ -255,39 +303,41 @@
# Test that we automatically follow 302 redirects
# and that we DO NOT cache the 302 response
http = httplib2.Http(cache=tests.get_cache_path())
- second_url, final_url = '', ''
+ second_url, final_url = "", ""
routes = {
- '/final': tests.http_response_bytes(body=b'This is the final destination.\n'),
- '/second': tests.http_response_bytes(
- status='302 Found', headers={'location': '/final'}, body=b'second redirect'),
- '': tests.http_response_bytes(
- status='302 Found', headers={'location': '/second'}, body=b'redirect body'),
+ "/final": tests.http_response_bytes(body=b"This is the final destination.\n"),
+ "/second": tests.http_response_bytes(
+ status="302 Found", headers={"location": "/final"}, body=b"second redirect"
+ ),
+ "": tests.http_response_bytes(
+ status="302 Found", headers={"location": "/second"}, body=b"redirect body"
+ ),
}
with tests.server_route(routes, request_count=7) as uri:
- second_url = urllib.parse.urljoin(uri, '/second')
- final_url = urllib.parse.urljoin(uri, '/final')
- response1, content1 = http.request(second_url, 'GET')
- response2, content2 = http.request(second_url, 'GET')
- response3, content3 = http.request(uri, 'GET')
+ second_url = urllib.parse.urljoin(uri, "/second")
+ final_url = urllib.parse.urljoin(uri, "/final")
+ response1, content1 = http.request(second_url, "GET")
+ response2, content2 = http.request(second_url, "GET")
+ response3, content3 = http.request(uri, "GET")
assert response1.status == 200
- assert response1['content-location'] == final_url
- assert content1 == b'This is the final destination.\n'
+ assert response1["content-location"] == final_url
+ assert content1 == b"This is the final destination.\n"
assert response1.previous.status == 302
assert not response1.previous.fromcache
assert response2.status == 200
# FIXME:
# assert response2.fromcache
- assert response2['content-location'] == final_url
- assert content2 == b'This is the final destination.\n'
+ assert response2["content-location"] == final_url
+ assert content2 == b"This is the final destination.\n"
assert response2.previous.status == 302
assert not response2.previous.fromcache
- assert response2.previous['content-location'] == second_url
+ assert response2.previous["content-location"] == second_url
assert response3.status == 200
# FIXME:
# assert response3.fromcache
- assert content3 == b'This is the final destination.\n'
+ assert content3 == b"This is the final destination.\n"
assert response3.previous.status == 302
assert not response3.previous.fromcache
@@ -299,28 +349,30 @@
http = httplib2.Http()
http.force_exception_to_status_code = False
routes = {
- '/second': tests.http_response_bytes(
- status='302 Found', headers={'location': '/final'}, body=b'second redirect'),
- '': tests.http_response_bytes(
- status='302 Found', headers={'location': '/second'}, body=b'redirect body'),
+ "/second": tests.http_response_bytes(
+ status="302 Found", headers={"location": "/final"}, body=b"second redirect"
+ ),
+ "": tests.http_response_bytes(
+ status="302 Found", headers={"location": "/second"}, body=b"redirect body"
+ ),
}
with tests.server_route(routes, request_count=4) as uri:
try:
- http.request(uri, 'GET', redirections=1)
- assert False, 'This should not happen'
+ http.request(uri, "GET", redirections=1)
+ assert False, "This should not happen"
except httplib2.RedirectLimit:
pass
except Exception:
- assert False, 'Threw wrong kind of exception '
+ assert False, "Threw wrong kind of exception "
# Re-run the test with out the exceptions
http.force_exception_to_status_code = True
- response, content = http.request(uri, 'GET', redirections=1)
+ response, content = http.request(uri, "GET", redirections=1)
assert response.status == 500
- assert response.reason.startswith('Redirected more')
- assert response['status'] == '302'
- assert content == b'second redirect'
+ assert response.reason.startswith("Redirected more")
+ assert response["status"] == "302"
+ assert content == b"second redirect"
assert response.previous is not None
@@ -329,61 +381,63 @@
# a 302 with no Location: header.
http = httplib2.Http()
http.force_exception_to_status_code = False
- with tests.server_const_http(status='302 Found', request_count=2) as uri:
+ with tests.server_const_http(status="302 Found", request_count=2) as uri:
try:
- http.request(uri, 'GET')
- assert False, 'Should never reach here'
+ http.request(uri, "GET")
+ assert False, "Should never reach here"
except httplib2.RedirectMissingLocation:
pass
except Exception:
- assert False, 'Threw wrong kind of exception '
+ assert False, "Threw wrong kind of exception "
# Re-run the test with out the exceptions
http.force_exception_to_status_code = True
- response, content = http.request(uri, 'GET')
+ response, content = http.request(uri, "GET")
assert response.status == 500
- assert response.reason.startswith('Redirected but')
- assert '302' == response['status']
- assert content == b''
+ assert response.reason.startswith("Redirected but")
+ assert "302" == response["status"]
+ assert content == b""
@pytest.mark.skip(
- not os.environ.get('httplib2_test_still_run_skipped') and
- os.environ.get('TRAVIS_PYTHON_VERSION') in ('2.7', 'pypy'),
- reason='FIXME: timeout on Travis py27 and pypy, works elsewhere',
+ not os.environ.get("httplib2_test_still_run_skipped")
+ and os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"),
+ reason="FIXME: timeout on Travis py27 and pypy, works elsewhere",
)
def test_303():
# Do a follow-up GET on a Location: header
# returned from a POST that gave a 303.
http = httplib2.Http()
routes = {
- '/final': tests.make_http_reflect(),
- '': tests.make_http_reflect(status='303 See Other', headers={'location': '/final'}),
+ "/final": tests.make_http_reflect(),
+ "": tests.make_http_reflect(
+ status="303 See Other", headers={"location": "/final"}
+ ),
}
with tests.server_route(routes, request_count=2) as uri:
- response, content = http.request(uri, 'POST', " ")
+ response, content = http.request(uri, "POST", " ")
assert response.status == 200
reflected = tests.HttpRequest.from_bytes(content)
- assert reflected.uri == '/final'
+ assert reflected.uri == "/final"
assert response.previous.status == 303
# Skip follow-up GET
http = httplib2.Http()
http.follow_redirects = False
with tests.server_route(routes, request_count=1) as uri:
- response, content = http.request(uri, 'POST', " ")
+ response, content = http.request(uri, "POST", " ")
assert response.status == 303
# All methods can be used
http = httplib2.Http()
- cases = 'DELETE GET HEAD POST PUT EVEN_NEW_ONES'.split(' ')
+ cases = "DELETE GET HEAD POST PUT EVEN_NEW_ONES".split(" ")
with tests.server_route(routes, request_count=len(cases) * 2) as uri:
for method in cases:
- response, content = http.request(uri, method, body=b'q q')
+ response, content = http.request(uri, method, body=b"q q")
assert response.status == 200
reflected = tests.HttpRequest.from_bytes(content)
- assert reflected.method == 'GET'
+ assert reflected.method == "GET"
def test_etag_used():
@@ -393,40 +447,45 @@
response_kwargs = dict(
add_date=True,
add_etag=True,
- body=b'something',
- headers={
- 'cache-control': 'public,max-age=300',
- },
+ body=b"something",
+ headers={"cache-control": "public,max-age=300"},
)
def handler(request):
- if request.headers.get('range'):
+ if request.headers.get("range"):
return tests.http_response_bytes(status=206, **response_kwargs)
return tests.http_response_bytes(**response_kwargs)
with tests.server_request(handler, request_count=2) as uri:
- response, _ = http.request(uri, 'GET', headers={'accept-encoding': 'identity'})
- assert response['etag'] == '"437b930db84b8079c2dd804a71936b5f"'
+ response, _ = http.request(uri, "GET", headers={"accept-encoding": "identity"})
+ assert response["etag"] == '"437b930db84b8079c2dd804a71936b5f"'
- http.request(uri, 'GET', headers={'accept-encoding': 'identity'})
+ http.request(uri, "GET", headers={"accept-encoding": "identity"})
response, _ = http.request(
- uri, 'GET',
- headers={'accept-encoding': 'identity', 'cache-control': 'must-revalidate'},
+ uri,
+ "GET",
+ headers={"accept-encoding": "identity", "cache-control": "must-revalidate"},
)
assert response.status == 200
assert response.fromcache
# TODO: API to read cache item, at least internal to tests
- cache_file_name = os.path.join(cache_path, httplib2.safename(httplib2.urlnorm(uri)[-1]))
- with open(cache_file_name, 'r') as f:
+ cache_file_name = os.path.join(
+ cache_path, httplib2.safename(httplib2.urlnorm(uri)[-1])
+ )
+ with open(cache_file_name, "r") as f:
status_line = f.readline()
assert status_line.startswith("status:")
- response, content = http.request(uri, 'HEAD', headers={'accept-encoding': 'identity'})
+ response, content = http.request(
+ uri, "HEAD", headers={"accept-encoding": "identity"}
+ )
assert response.status == 200
assert response.fromcache
- response, content = http.request(uri, 'GET', headers={'accept-encoding': 'identity', 'range': 'bytes=0-0'})
+ response, content = http.request(
+ uri, "GET", headers={"accept-encoding": "identity", "range": "bytes=0-0"}
+ )
assert response.status == 206
assert not response.fromcache
@@ -434,63 +493,67 @@
def test_etag_ignore():
# Test that we can forcibly ignore ETags
http = httplib2.Http(cache=tests.get_cache_path())
- response_kwargs = dict(
- add_date=True,
- add_etag=True,
- )
+ response_kwargs = dict(add_date=True, add_etag=True)
with tests.server_reflect(request_count=3, **response_kwargs) as uri:
- response, content = http.request(uri, 'GET', headers={'accept-encoding': 'identity'})
+ response, content = http.request(
+ uri, "GET", headers={"accept-encoding": "identity"}
+ )
assert response.status == 200
- assert response['etag'] != ""
+ assert response["etag"] != ""
response, content = http.request(
- uri, 'GET',
- headers={'accept-encoding': 'identity', 'cache-control': 'max-age=0'},
+ uri,
+ "GET",
+ headers={"accept-encoding": "identity", "cache-control": "max-age=0"},
)
reflected = tests.HttpRequest.from_bytes(content)
- assert reflected.headers.get('if-none-match')
+ assert reflected.headers.get("if-none-match")
http.ignore_etag = True
response, content = http.request(
- uri, 'GET',
- headers={'accept-encoding': 'identity', 'cache-control': 'max-age=0'},
+ uri,
+ "GET",
+ headers={"accept-encoding": "identity", "cache-control": "max-age=0"},
)
assert not response.fromcache
reflected = tests.HttpRequest.from_bytes(content)
- assert not reflected.headers.get('if-none-match')
+ assert not reflected.headers.get("if-none-match")
def test_etag_override():
# Test that we can forcibly ignore ETags
http = httplib2.Http(cache=tests.get_cache_path())
- response_kwargs = dict(
- add_date=True,
- add_etag=True,
- )
+ response_kwargs = dict(add_date=True, add_etag=True)
with tests.server_reflect(request_count=3, **response_kwargs) as uri:
- response, _ = http.request(uri, 'GET', headers={'accept-encoding': 'identity'})
+ response, _ = http.request(uri, "GET", headers={"accept-encoding": "identity"})
assert response.status == 200
- assert response['etag'] != ''
+ assert response["etag"] != ""
response, content = http.request(
- uri, 'GET',
- headers={'accept-encoding': 'identity', 'cache-control': 'max-age=0'},
+ uri,
+ "GET",
+ headers={"accept-encoding": "identity", "cache-control": "max-age=0"},
)
assert response.status == 200
reflected = tests.HttpRequest.from_bytes(content)
- assert reflected.headers.get('if-none-match')
- assert reflected.headers.get('if-none-match') != 'fred'
+ assert reflected.headers.get("if-none-match")
+ assert reflected.headers.get("if-none-match") != "fred"
response, content = http.request(
- uri, 'GET',
- headers={'accept-encoding': 'identity', 'cache-control': 'max-age=0', 'if-none-match': 'fred'},
+ uri,
+ "GET",
+ headers={
+ "accept-encoding": "identity",
+ "cache-control": "max-age=0",
+ "if-none-match": "fred",
+ },
)
assert response.status == 200
reflected = tests.HttpRequest.from_bytes(content)
- assert reflected.headers.get('if-none-match') == 'fred'
+ assert reflected.headers.get("if-none-match") == "fred"
-@pytest.mark.skip(reason='was commented in legacy code')
+@pytest.mark.skip(reason="was commented in legacy code")
def test_get_304_end_to_end():
pass
# Test that end to end headers get overwritten in the cache
@@ -517,23 +580,18 @@
def handler(read):
read()
yield tests.http_response_bytes(
- status=200,
- body=b'something',
- headers={
- 'date': date,
- 'last-modified': date,
- },
+ status=200, body=b"something", headers={"date": date, "last-modified": date}
)
request2 = read()
- assert request2.headers['if-modified-since'] == date
+ assert request2.headers["if-modified-since"] == date
yield tests.http_response_bytes(status=304)
with tests.server_yield(handler, request_count=2) as uri:
- response, content = http.request(uri, 'GET')
- assert response.get('last-modified') == date
+ response, content = http.request(uri, "GET")
+ assert response.get("last-modified") == date
- response, content = http.request(uri, 'GET')
+ response, content = http.request(uri, "GET")
assert response.status == 200
assert response.fromcache
@@ -542,51 +600,48 @@
# Test that we do follow 307 redirects but
# do not cache the 307
http = httplib2.Http(cache=tests.get_cache_path(), timeout=1)
- r307 = tests.http_response_bytes(
- status=307,
- headers={'location': '/final'},
- )
+ r307 = tests.http_response_bytes(status=307, headers={"location": "/final"})
r200 = tests.http_response_bytes(
status=200,
add_date=True,
- body=b'final content\n',
- headers={'cache-control': 'max-age=300'},
+ body=b"final content\n",
+ headers={"cache-control": "max-age=300"},
)
with tests.server_list_http([r307, r200, r307]) as uri:
- response, content = http.request(uri, 'GET')
+ response, content = http.request(uri, "GET")
assert response.previous.status == 307
assert not response.previous.fromcache
assert response.status == 200
assert not response.fromcache
- assert content == b'final content\n'
+ assert content == b"final content\n"
- response, content = http.request(uri, 'GET')
+ response, content = http.request(uri, "GET")
assert response.previous.status == 307
assert not response.previous.fromcache
assert response.status == 200
assert response.fromcache
- assert content == b'final content\n'
+ assert content == b"final content\n"
def test_get_410():
# Test that we pass 410's through
http = httplib2.Http()
with tests.server_const_http(status=410) as uri:
- response, content = http.request(uri, 'GET')
+ response, content = http.request(uri, "GET")
assert response.status == 410
def test_get_duplicate_headers():
# Test that duplicate headers get concatenated via ','
http = httplib2.Http()
- response = b'''HTTP/1.0 200 OK\r\n\
+ response = b"""HTTP/1.0 200 OK\r\n\
Link: link1\r\n\
Content-Length: 7\r\n\
Link: link2\r\n\r\n\
-content'''
+content"""
with tests.server_const_bytes(response) as uri:
- response, content = http.request(uri, 'GET')
+ response, content = http.request(uri, "GET")
assert response.status == 200
assert content == b"content"
- assert response['link'], 'link1, link2'
+ assert response["link"], "link1, link2"