encode/httpx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Drop Response(..., request=...) style in test cases. (#1243)

Browse Source 
* Drop Response(..., request=...) style in test cases except where required

* Lowercase variable name
This commit is contained in:
Tom Christie 2020-09-02 10:10:32 +01:00 committed by GitHub
parent 46b9282c6e
commit de502a44c6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 209 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

203
tests/models/test_responses.py
View File

@ -8,8 +8,6 @@ import pytest
import httpx
from httpx._content_streams import AsyncIteratorStream, IteratorStream
REQUEST = httpx.Request("GET", "https://example.org")
def streaming_body():
yield b"Hello, "
@ -22,28 +20,35 @@ async def async_streaming_body():
def test_response():
response = httpx.Response(200, content=b"Hello, world!", request=REQUEST)
response = httpx.Response(
200,
content=b"Hello, world!",
request=httpx.Request("GET", "https://example.org"),
)
assert response.status_code == 200
assert response.reason_phrase == "OK"
assert response.text == "Hello, world!"
assert response.request is REQUEST
assert response.request.method == "GET"
assert response.request.url == "https://example.org"
assert response.elapsed >= datetime.timedelta(0)
assert not response.is_error
def test_raise_for_status():
request = httpx.Request("GET", "https://example.org")
# 2xx status codes are not an error.
response = httpx.Response(200, request=REQUEST)
response = httpx.Response(200, request=request)
response.raise_for_status()
# 4xx status codes are a client error.
response = httpx.Response(403, request=REQUEST)
response = httpx.Response(403, request=request)
with pytest.raises(httpx.HTTPStatusError):
response.raise_for_status()
# 5xx status codes are a server error.
response = httpx.Response(500, request=REQUEST)
response = httpx.Response(500, request=request)
with pytest.raises(httpx.HTTPStatusError):
response.raise_for_status()
@ -55,7 +60,10 @@ def test_raise_for_status():
def test_response_repr():
response = httpx.Response(200, content=b"Hello, world!", request=REQUEST)
response = httpx.Response(
200,
content=b"Hello, world!",
)
assert repr(response) == "<Response [200 OK]>"
@ -65,7 +73,11 @@ def test_response_content_type_encoding():
"""
headers = {"Content-Type": "text-plain; charset=latin-1"}
content = "Latin 1: ÿ".encode("latin-1")
response = httpx.Response(200, content=content, headers=headers, request=REQUEST)
response = httpx.Response(
200,
content=content,
headers=headers,
)
assert response.text == "Latin 1: ÿ"
assert response.encoding == "latin-1"
@ -75,7 +87,10 @@ def test_response_autodetect_encoding():
Autodetect encoding if there is no charset info in a Content-Type header.
"""
content = "おはようございます。".encode("EUC-JP")
response = httpx.Response(200, content=content, request=REQUEST)
response = httpx.Response(
200,
content=content,
)
assert response.text == "おはようございます。"
assert response.encoding == "EUC-JP"
@ -86,7 +101,11 @@ def test_response_fallback_to_autodetect():
"""
headers = {"Content-Type": "text-plain; charset=invalid-codec-name"}
content = "おはようございます。".encode("EUC-JP")
response = httpx.Response(200, content=content, headers=headers, request=REQUEST)
response = httpx.Response(
200,
content=content,
headers=headers,
)
assert response.text == "おはようございます。"
assert response.encoding == "EUC-JP"
@ -98,7 +117,11 @@ def test_response_default_text_encoding():
"""
content = b"Hello, world!"
headers = {"Content-Type": "text/plain"}
response = httpx.Response(200, content=content, headers=headers, request=REQUEST)
response = httpx.Response(
200,
content=content,
headers=headers,
)
assert response.status_code == 200
assert response.encoding == "iso-8859-1"
assert response.text == "Hello, world!"
@ -108,7 +131,10 @@ def test_response_default_encoding():
"""
Default to utf-8 if all else fails.
"""
response = httpx.Response(200, content=b"", request=REQUEST)
response = httpx.Response(
200,
content=b"",
)
assert response.text == ""
assert response.encoding == "utf-8"
@ -118,7 +144,11 @@ def test_response_non_text_encoding():
Default to apparent encoding for non-text content-type headers.
"""
headers = {"Content-Type": "image/png"}
response = httpx.Response(200, content=b"xyz", headers=headers, request=REQUEST)
response = httpx.Response(
200,
content=b"xyz",
headers=headers,
)
assert response.text == "xyz"
assert response.encoding == "ascii"
@ -128,7 +158,9 @@ def test_response_set_explicit_encoding():
"Content-Type": "text-plain; charset=utf-8"
} # Deliberately incorrect charset
response = httpx.Response(
200, content="Latin 1: ÿ".encode("latin-1"), headers=headers, request=REQUEST
200,
content="Latin 1: ÿ".encode("latin-1"),
headers=headers,
)
response.encoding = "latin-1"
assert response.text == "Latin 1: ÿ"
@ -137,7 +169,8 @@ def test_response_set_explicit_encoding():
def test_response_force_encoding():
response = httpx.Response(
200, content="Snowman: ☃".encode("utf-8"), request=REQUEST
200,
content="Snowman: ☃".encode("utf-8"),
)
response.encoding = "iso-8859-1"
assert response.status_code == 200
@ -147,7 +180,10 @@ def test_response_force_encoding():
def test_read():
response = httpx.Response(200, content=b"Hello, world!", request=REQUEST)
response = httpx.Response(
200,
content=b"Hello, world!",
)
assert response.status_code == 200
assert response.text == "Hello, world!"
@ -163,7 +199,10 @@ def test_read():
@pytest.mark.asyncio
async def test_aread():
response = httpx.Response(200, content=b"Hello, world!", request=REQUEST)
response = httpx.Response(
200,
content=b"Hello, world!",
)
assert response.status_code == 200
assert response.text == "Hello, world!"
@ -179,7 +218,10 @@ async def test_aread():
def test_iter_raw():
stream = IteratorStream(iterator=streaming_body())
response = httpx.Response(200, stream=stream, request=REQUEST)
response = httpx.Response(
200,
stream=stream,
)
raw = b""
for part in response.iter_raw():
@ -190,7 +232,10 @@ def test_iter_raw():
@pytest.mark.asyncio
async def test_aiter_raw():
stream = AsyncIteratorStream(aiterator=async_streaming_body())
response = httpx.Response(200, stream=stream, request=REQUEST)
response = httpx.Response(
200,
stream=stream,
)
raw = b""
async for part in response.aiter_raw():
@ -199,7 +244,10 @@ async def test_aiter_raw():
def test_iter_bytes():
response = httpx.Response(200, content=b"Hello, world!", request=REQUEST)
response = httpx.Response(
200,
content=b"Hello, world!",
)
content = b""
for part in response.iter_bytes():
@ -209,7 +257,10 @@ def test_iter_bytes():
@pytest.mark.asyncio
async def test_aiter_bytes():
response = httpx.Response(200, content=b"Hello, world!", request=REQUEST)
response = httpx.Response(
200,
content=b"Hello, world!",
)
content = b""
async for part in response.aiter_bytes():
@ -218,7 +269,10 @@ async def test_aiter_bytes():
def test_iter_text():
response = httpx.Response(200, content=b"Hello, world!", request=REQUEST)
response = httpx.Response(
200,
content=b"Hello, world!",
)
content = ""
for part in response.iter_text():
@ -228,7 +282,10 @@ def test_iter_text():
@pytest.mark.asyncio
async def test_aiter_text():
response = httpx.Response(200, content=b"Hello, world!", request=REQUEST)
response = httpx.Response(
200,
content=b"Hello, world!",
)
content = ""
async for part in response.aiter_text():
@ -237,7 +294,10 @@ async def test_aiter_text():
def test_iter_lines():
response = httpx.Response(200, content=b"Hello,\nworld!", request=REQUEST)
response = httpx.Response(
200,
content=b"Hello,\nworld!",
)
content = []
for line in response.iter_lines():
@ -247,7 +307,10 @@ def test_iter_lines():
@pytest.mark.asyncio
async def test_aiter_lines():
response = httpx.Response(200, content=b"Hello,\nworld!", request=REQUEST)
response = httpx.Response(
200,
content=b"Hello,\nworld!",
)
content = []
async for line in response.aiter_lines():
@ -257,7 +320,10 @@ async def test_aiter_lines():
def test_sync_streaming_response():
stream = IteratorStream(iterator=streaming_body())
response = httpx.Response(200, stream=stream, request=REQUEST)
response = httpx.Response(
200,
stream=stream,
)
assert response.status_code == 200
assert not response.is_closed
@ -272,7 +338,10 @@ def test_sync_streaming_response():
@pytest.mark.asyncio
async def test_async_streaming_response():
stream = AsyncIteratorStream(aiterator=async_streaming_body())
response = httpx.Response(200, stream=stream, request=REQUEST)
response = httpx.Response(
200,
stream=stream,
)
assert response.status_code == 200
assert not response.is_closed
@ -286,7 +355,10 @@ async def test_async_streaming_response():
def test_cannot_read_after_stream_consumed():
stream = IteratorStream(iterator=streaming_body())
response = httpx.Response(200, stream=stream, request=REQUEST)
response = httpx.Response(
200,
stream=stream,
)
content = b""
for part in response.iter_bytes():
@ -299,7 +371,10 @@ def test_cannot_read_after_stream_consumed():
@pytest.mark.asyncio
async def test_cannot_aread_after_stream_consumed():
stream = AsyncIteratorStream(aiterator=async_streaming_body())
response = httpx.Response(200, stream=stream, request=REQUEST)
response = httpx.Response(
200,
stream=stream,
)
content = b""
async for part in response.aiter_bytes():
@ -317,7 +392,10 @@ def test_cannot_read_after_response_closed():
is_closed = True
stream = IteratorStream(iterator=streaming_body(), close_func=close_func)
response = httpx.Response(200, stream=stream, request=REQUEST)
response = httpx.Response(
200,
stream=stream,
)
response.close()
assert is_closed
@ -337,7 +415,10 @@ async def test_cannot_aread_after_response_closed():
stream = AsyncIteratorStream(
aiterator=async_streaming_body(), close_func=close_func
)
response = httpx.Response(200, stream=stream, request=REQUEST)
response = httpx.Response(
200,
stream=stream,
)
await response.aclose()
assert is_closed
@ -349,14 +430,19 @@ async def test_cannot_aread_after_response_closed():
@pytest.mark.asyncio
async def test_elapsed_not_available_until_closed():
stream = AsyncIteratorStream(aiterator=async_streaming_body())
response = httpx.Response(200, stream=stream, request=REQUEST)
response = httpx.Response(
200,
stream=stream,
)
with pytest.raises(RuntimeError):
response.elapsed
def test_unknown_status_code():
response = httpx.Response(600, request=REQUEST)
response = httpx.Response(
600,
)
assert response.status_code == 600
assert response.reason_phrase == ""
assert response.text == ""
@ -366,7 +452,11 @@ def test_json_with_specified_encoding():
data = {"greeting": "hello", "recipient": "world"}
content = json.dumps(data).encode("utf-16")
headers = {"Content-Type": "application/json, charset=utf-16"}
response = httpx.Response(200, content=content, headers=headers, request=REQUEST)
response = httpx.Response(
200,
content=content,
headers=headers,
)
assert response.json() == data
@ -374,7 +464,11 @@ def test_json_with_options():
data = {"greeting": "hello", "recipient": "world", "amount": 1}
content = json.dumps(data).encode("utf-16")
headers = {"Content-Type": "application/json, charset=utf-16"}
response = httpx.Response(200, content=content, headers=headers, request=REQUEST)
response = httpx.Response(
200,
content=content,
headers=headers,
)
assert response.json(parse_int=str)["amount"] == "1"
@ -382,7 +476,11 @@ def test_json_without_specified_encoding():
data = {"greeting": "hello", "recipient": "world"}
content = json.dumps(data).encode("utf-32-be")
headers = {"Content-Type": "application/json"}
response = httpx.Response(200, content=content, headers=headers, request=REQUEST)
response = httpx.Response(
200,
content=content,
headers=headers,
)
assert response.json() == data
@ -393,7 +491,9 @@ def test_json_without_specified_encoding_decode_error():
# force incorrect guess from `guess_json_utf` to trigger error
with mock.patch("httpx._models.guess_json_utf", return_value="utf-32"):
response = httpx.Response(
200, content=content, headers=headers, request=REQUEST
200,
content=content,
headers=headers,
)
with pytest.raises(json.decoder.JSONDecodeError):
response.json()
@ -427,7 +527,11 @@ def test_json_without_specified_encoding_value_error():
],
)
def test_link_headers(headers, expected):
response = httpx.Response(200, content=None, headers=headers, request=REQUEST)
response = httpx.Response(
200,
content=None,
headers=headers,
)
assert response.links == expected
@ -436,8 +540,20 @@ def test_decode_error_with_request(header_value):
headers = [(b"Content-Encoding", header_value)]
body = b"test 123"
compressed_body = brotli.compress(body)[3:]
with pytest.raises(ValueError):
httpx.Response(
200,
headers=headers,
content=compressed_body,
)
with pytest.raises(httpx.DecodingError):
httpx.Response(200, headers=headers, content=compressed_body, request=REQUEST)
httpx.Response(
200,
headers=headers,
content=compressed_body,
request=httpx.Request("GET", "https://www.example.org/"),
)
@pytest.mark.parametrize("header_value", (b"deflate", b"gzip", b"br"))
@ -461,13 +577,14 @@ def test_response_with_unset_request():
def test_set_request_after_init():
response = httpx.Response(200, content=b"Hello, world!")
response.request = REQUEST
response.request = httpx.Request("GET", "https://www.example.org")
assert response.request == REQUEST
assert response.request.method == "GET"
assert response.request.url == "https://www.example.org"
def test_cannot_access_unset_request():
response = httpx.Response(200, content=b"Hello, world!")
with pytest.raises(RuntimeError):
assert response.request is not None
response.request

65
tests/test_decoders.py
View File

@ -14,8 +14,6 @@ from httpx._decoders import (
TextDecoder,
)
REQUEST = httpx.Request("GET", "https://example.org")
def test_deflate():
"""
@ -29,7 +27,9 @@ def test_deflate():
headers = [(b"Content-Encoding", b"deflate")]
response = httpx.Response(
200, headers=headers, content=compressed_body, request=REQUEST
200,
headers=headers,
content=compressed_body,
)
assert response.content == body
@ -45,7 +45,9 @@ def test_zlib():
headers = [(b"Content-Encoding", b"deflate")]
response = httpx.Response(
200, headers=headers, content=compressed_body, request=REQUEST
200,
headers=headers,
content=compressed_body,
)
assert response.content == body
@ -57,7 +59,9 @@ def test_gzip():
headers = [(b"Content-Encoding", b"gzip")]
response = httpx.Response(
200, headers=headers, content=compressed_body, request=REQUEST
200,
headers=headers,
content=compressed_body,
)
assert response.content == body
@ -68,7 +72,9 @@ def test_brotli():
headers = [(b"Content-Encoding", b"br")]
response = httpx.Response(
200, headers=headers, content=compressed_body, request=REQUEST
200,
headers=headers,
content=compressed_body,
)
assert response.content == body
@ -86,7 +92,9 @@ def test_multi():
headers = [(b"Content-Encoding", b"deflate, gzip")]
response = httpx.Response(
200, headers=headers, content=compressed_body, request=REQUEST
200,
headers=headers,
content=compressed_body,
)
assert response.content == body
@ -97,13 +105,17 @@ def test_multi_with_identity():
headers = [(b"Content-Encoding", b"br, identity")]
response = httpx.Response(
200, headers=headers, content=compressed_body, request=REQUEST
200,
headers=headers,
content=compressed_body,
)
assert response.content == body
headers = [(b"Content-Encoding", b"identity, br")]
response = httpx.Response(
200, headers=headers, content=compressed_body, request=REQUEST
200,
headers=headers,
content=compressed_body,
)
assert response.content == body
@ -119,7 +131,11 @@ async def test_streaming():
headers = [(b"Content-Encoding", b"gzip")]
stream = AsyncIteratorStream(aiterator=compress(body))
response = httpx.Response(200, headers=headers, stream=stream, request=REQUEST)
response = httpx.Response(
200,
headers=headers,
stream=stream,
)
assert not hasattr(response, "body")
assert await response.aread() == body
@ -127,7 +143,11 @@ async def test_streaming():
@pytest.mark.parametrize("header_value", (b"deflate", b"gzip", b"br", b"identity"))
def test_empty_content(header_value):
headers = [(b"Content-Encoding", header_value)]
response = httpx.Response(200, headers=headers, content=b"", request=REQUEST)
response = httpx.Response(
200,
headers=headers,
content=b"",
)
assert response.content == b""
@ -147,7 +167,11 @@ def test_decoding_errors(header_value):
body = b"test 123"
compressed_body = brotli.compress(body)[3:]
with pytest.raises(httpx.DecodingError):
httpx.Response(200, headers=headers, content=compressed_body, request=REQUEST)
request = httpx.Request("GET", "https://example.org")
httpx.Response(200, headers=headers, content=compressed_body, request=request)
with pytest.raises(ValueError):
httpx.Response(200, headers=headers, content=compressed_body)
@pytest.mark.parametrize(
@ -176,13 +200,19 @@ async def test_text_decoder(data, encoding):
# Accessing `.text` on a read response.
stream = AsyncIteratorStream(aiterator=iterator())
response = httpx.Response(200, stream=stream, request=REQUEST)
response = httpx.Response(
200,
stream=stream,
)
await response.aread()
assert response.text == (b"".join(data)).decode(encoding)
# Streaming `.aiter_text` iteratively.
stream = AsyncIteratorStream(aiterator=iterator())
response = httpx.Response(200, stream=stream, request=REQUEST)
response = httpx.Response(
200,
stream=stream,
)
text = "".join([part async for part in response.aiter_text()])
assert text == (b"".join(data)).decode(encoding)
@ -199,7 +229,6 @@ async def test_text_decoder_known_encoding():
200,
headers=[(b"Content-Type", b"text/html; charset=shift-jis")],
stream=stream,
request=REQUEST,
)
await response.aread()
@ -288,5 +317,9 @@ def test_invalid_content_encoding_header():
headers = [(b"Content-Encoding", b"invalid-header")]
body = b"test 123"
response = httpx.Response(200, headers=headers, content=body, request=REQUEST)
response = httpx.Response(
200,
headers=headers,
content=body,
)
assert response.content == body