Better test case consistency. Prefer import httpx and httpx.Client. (#1222)
* Prefer httpx.Client over httpx.AsyncClient in test cases, unless required. * Prefer httpx.Client in test_headers * Consistent httpx imports and httpx.Client usage * Use 'import httpx' consistently in tests. Prefer httpx.Client.
This commit is contained in:
Tom Christie
GitHub
parent
534400ee42
commit
28c72050e0
@ -2,9 +2,8 @@ import typing
|
||||
from http.cookiejar import Cookie, CookieJar
|
||||
|
||||
import httpcore
|
||||
import pytest
|
||||
|
||||
from httpx import AsyncClient, Cookies
|
||||
import httpx
|
||||
from httpx._content_streams import ByteStream, ContentStream, JSONStream
|
||||
|
||||
|
||||
@ -16,13 +15,13 @@ def get_header_value(headers, key, default=None):
|
||||
return default
|
||||
|
||||
|
||||
class MockTransport(httpcore.AsyncHTTPTransport):
|
||||
async def request(
|
||||
class MockTransport(httpcore.SyncHTTPTransport):
|
||||
def request(
|
||||
self,
|
||||
method: bytes,
|
||||
url: typing.Tuple[bytes, bytes, typing.Optional[int], bytes],
|
||||
headers: typing.List[typing.Tuple[bytes, bytes]] = None,
|
||||
stream: httpcore.AsyncByteStream = None,
|
||||
stream: httpcore.SyncByteStream = None,
|
||||
timeout: typing.Mapping[str, typing.Optional[float]] = None,
|
||||
) -> typing.Tuple[
|
||||
bytes, int, bytes, typing.List[typing.Tuple[bytes, bytes]], ContentStream
|
||||
@ -41,23 +40,21 @@ class MockTransport(httpcore.AsyncHTTPTransport):
|
||||
raise NotImplementedError() # pragma: no cover
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_set_cookie() -> None:
|
||||
def test_set_cookie() -> None:
|
||||
"""
|
||||
Send a request including a cookie.
|
||||
"""
|
||||
url = "http://example.org/echo_cookies"
|
||||
cookies = {"example-name": "example-value"}
|
||||
|
||||
client = AsyncClient(transport=MockTransport())
|
||||
response = await client.get(url, cookies=cookies)
|
||||
client = httpx.Client(transport=MockTransport())
|
||||
response = client.get(url, cookies=cookies)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"cookies": "example-name=example-value"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_set_cookie_with_cookiejar() -> None:
|
||||
def test_set_cookie_with_cookiejar() -> None:
|
||||
"""
|
||||
Send a request including a cookie, using a `CookieJar` instance.
|
||||
"""
|
||||
@ -85,15 +82,14 @@ async def test_set_cookie_with_cookiejar() -> None:
|
||||
)
|
||||
cookies.set_cookie(cookie)
|
||||
|
||||
client = AsyncClient(transport=MockTransport())
|
||||
response = await client.get(url, cookies=cookies)
|
||||
client = httpx.Client(transport=MockTransport())
|
||||
response = client.get(url, cookies=cookies)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"cookies": "example-name=example-value"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setting_client_cookies_to_cookiejar() -> None:
|
||||
def test_setting_client_cookies_to_cookiejar() -> None:
|
||||
"""
|
||||
Send a request including a cookie, using a `CookieJar` instance.
|
||||
"""
|
||||
@ -121,59 +117,56 @@ async def test_setting_client_cookies_to_cookiejar() -> None:
|
||||
)
|
||||
cookies.set_cookie(cookie)
|
||||
|
||||
client = AsyncClient(transport=MockTransport())
|
||||
client = httpx.Client(transport=MockTransport())
|
||||
client.cookies = cookies # type: ignore
|
||||
response = await client.get(url)
|
||||
response = client.get(url)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"cookies": "example-name=example-value"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_set_cookie_with_cookies_model() -> None:
|
||||
def test_set_cookie_with_cookies_model() -> None:
|
||||
"""
|
||||
Send a request including a cookie, using a `Cookies` instance.
|
||||
"""
|
||||
|
||||
url = "http://example.org/echo_cookies"
|
||||
cookies = Cookies()
|
||||
cookies = httpx.Cookies()
|
||||
cookies["example-name"] = "example-value"
|
||||
|
||||
client = AsyncClient(transport=MockTransport())
|
||||
response = await client.get(url, cookies=cookies)
|
||||
client = httpx.Client(transport=MockTransport())
|
||||
response = client.get(url, cookies=cookies)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"cookies": "example-name=example-value"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_cookie() -> None:
|
||||
def test_get_cookie() -> None:
|
||||
url = "http://example.org/set_cookie"
|
||||
|
||||
client = AsyncClient(transport=MockTransport())
|
||||
response = await client.get(url)
|
||||
client = httpx.Client(transport=MockTransport())
|
||||
response = client.get(url)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.cookies["example-name"] == "example-value"
|
||||
assert client.cookies["example-name"] == "example-value"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cookie_persistence() -> None:
|
||||
def test_cookie_persistence() -> None:
|
||||
"""
|
||||
Ensure that Client instances persist cookies between requests.
|
||||
"""
|
||||
client = AsyncClient(transport=MockTransport())
|
||||
client = httpx.Client(transport=MockTransport())
|
||||
|
||||
response = await client.get("http://example.org/echo_cookies")
|
||||
response = client.get("http://example.org/echo_cookies")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"cookies": None}
|
||||
|
||||
response = await client.get("http://example.org/set_cookie")
|
||||
response = client.get("http://example.org/set_cookie")
|
||||
assert response.status_code == 200
|
||||
assert response.cookies["example-name"] == "example-value"
|
||||
assert client.cookies["example-name"] == "example-value"
|
||||
|
||||
response = await client.get("http://example.org/echo_cookies")
|
||||
response = client.get("http://example.org/echo_cookies")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"cookies": "example-name=example-value"}
|
||||
|
||||
@ -5,17 +5,17 @@ import typing
|
||||
import httpcore
|
||||
import pytest
|
||||
|
||||
from httpx import AsyncClient, Headers, Request, __version__
|
||||
import httpx
|
||||
from httpx._content_streams import ContentStream, JSONStream
|
||||
|
||||
|
||||
class MockTransport(httpcore.AsyncHTTPTransport):
|
||||
async def request(
|
||||
class MockTransport(httpcore.SyncHTTPTransport):
|
||||
def request(
|
||||
self,
|
||||
method: bytes,
|
||||
url: typing.Tuple[bytes, bytes, typing.Optional[int], bytes],
|
||||
headers: typing.List[typing.Tuple[bytes, bytes]] = None,
|
||||
stream: httpcore.AsyncByteStream = None,
|
||||
stream: httpcore.SyncByteStream = None,
|
||||
timeout: typing.Mapping[str, typing.Optional[float]] = None,
|
||||
) -> typing.Tuple[
|
||||
bytes, int, bytes, typing.List[typing.Tuple[bytes, bytes]], ContentStream
|
||||
@ -28,16 +28,15 @@ class MockTransport(httpcore.AsyncHTTPTransport):
|
||||
return b"HTTP/1.1", 200, b"OK", [], body
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_client_header():
|
||||
def test_client_header():
|
||||
"""
|
||||
Set a header in the Client.
|
||||
"""
|
||||
url = "http://example.org/echo_headers"
|
||||
headers = {"Example-Header": "example-value"}
|
||||
|
||||
client = AsyncClient(transport=MockTransport(), headers=headers)
|
||||
response = await client.get(url)
|
||||
client = httpx.Client(transport=MockTransport(), headers=headers)
|
||||
response = client.get(url)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {
|
||||
@ -47,18 +46,17 @@ async def test_client_header():
|
||||
"connection": "keep-alive",
|
||||
"example-header": "example-value",
|
||||
"host": "example.org",
|
||||
"user-agent": f"python-httpx/{__version__}",
|
||||
"user-agent": f"python-httpx/{httpx.__version__}",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_header_merge():
|
||||
def test_header_merge():
|
||||
url = "http://example.org/echo_headers"
|
||||
client_headers = {"User-Agent": "python-myclient/0.2.1"}
|
||||
request_headers = {"X-Auth-Token": "FooBarBazToken"}
|
||||
client = AsyncClient(transport=MockTransport(), headers=client_headers)
|
||||
response = await client.get(url, headers=request_headers)
|
||||
client = httpx.Client(transport=MockTransport(), headers=client_headers)
|
||||
response = client.get(url, headers=request_headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {
|
||||
@ -73,13 +71,12 @@ async def test_header_merge():
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_header_merge_conflicting_headers():
|
||||
def test_header_merge_conflicting_headers():
|
||||
url = "http://example.org/echo_headers"
|
||||
client_headers = {"X-Auth-Token": "FooBar"}
|
||||
request_headers = {"X-Auth-Token": "BazToken"}
|
||||
client = AsyncClient(transport=MockTransport(), headers=client_headers)
|
||||
response = await client.get(url, headers=request_headers)
|
||||
client = httpx.Client(transport=MockTransport(), headers=client_headers)
|
||||
response = client.get(url, headers=request_headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {
|
||||
@ -88,21 +85,20 @@ async def test_header_merge_conflicting_headers():
|
||||
"accept-encoding": "gzip, deflate, br",
|
||||
"connection": "keep-alive",
|
||||
"host": "example.org",
|
||||
"user-agent": f"python-httpx/{__version__}",
|
||||
"user-agent": f"python-httpx/{httpx.__version__}",
|
||||
"x-auth-token": "BazToken",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_header_update():
|
||||
def test_header_update():
|
||||
url = "http://example.org/echo_headers"
|
||||
client = AsyncClient(transport=MockTransport())
|
||||
first_response = await client.get(url)
|
||||
client = httpx.Client(transport=MockTransport())
|
||||
first_response = client.get(url)
|
||||
client.headers.update(
|
||||
{"User-Agent": "python-myclient/0.2.1", "Another-Header": "AThing"}
|
||||
)
|
||||
second_response = await client.get(url)
|
||||
second_response = client.get(url)
|
||||
|
||||
assert first_response.status_code == 200
|
||||
assert first_response.json() == {
|
||||
@ -111,7 +107,7 @@ async def test_header_update():
|
||||
"accept-encoding": "gzip, deflate, br",
|
||||
"connection": "keep-alive",
|
||||
"host": "example.org",
|
||||
"user-agent": f"python-httpx/{__version__}",
|
||||
"user-agent": f"python-httpx/{httpx.__version__}",
|
||||
}
|
||||
}
|
||||
|
||||
@ -129,13 +125,12 @@ async def test_header_update():
|
||||
|
||||
|
||||
def test_header_does_not_exist():
|
||||
headers = Headers({"foo": "bar"})
|
||||
headers = httpx.Headers({"foo": "bar"})
|
||||
with pytest.raises(KeyError):
|
||||
del headers["baz"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_host_with_auth_and_port_in_url():
|
||||
def test_host_with_auth_and_port_in_url():
|
||||
"""
|
||||
The Host header should only include the hostname, or hostname:port
|
||||
(for non-default ports only). Any userinfo or default port should not
|
||||
@ -143,8 +138,8 @@ async def test_host_with_auth_and_port_in_url():
|
||||
"""
|
||||
url = "http://username:password@example.org:80/echo_headers"
|
||||
|
||||
client = AsyncClient(transport=MockTransport())
|
||||
response = await client.get(url)
|
||||
client = httpx.Client(transport=MockTransport())
|
||||
response = client.get(url)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {
|
||||
@ -153,22 +148,21 @@ async def test_host_with_auth_and_port_in_url():
|
||||
"accept-encoding": "gzip, deflate, br",
|
||||
"connection": "keep-alive",
|
||||
"host": "example.org",
|
||||
"user-agent": f"python-httpx/{__version__}",
|
||||
"user-agent": f"python-httpx/{httpx.__version__}",
|
||||
"authorization": "Basic dXNlcm5hbWU6cGFzc3dvcmQ=",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_host_with_non_default_port_in_url():
|
||||
def test_host_with_non_default_port_in_url():
|
||||
"""
|
||||
If the URL includes a non-default port, then it should be included in
|
||||
the Host header.
|
||||
"""
|
||||
url = "http://username:password@example.org:123/echo_headers"
|
||||
|
||||
client = AsyncClient(transport=MockTransport())
|
||||
response = await client.get(url)
|
||||
client = httpx.Client(transport=MockTransport())
|
||||
response = client.get(url)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {
|
||||
@ -177,12 +171,12 @@ async def test_host_with_non_default_port_in_url():
|
||||
"accept-encoding": "gzip, deflate, br",
|
||||
"connection": "keep-alive",
|
||||
"host": "example.org:123",
|
||||
"user-agent": f"python-httpx/{__version__}",
|
||||
"user-agent": f"python-httpx/{httpx.__version__}",
|
||||
"authorization": "Basic dXNlcm5hbWU6cGFzc3dvcmQ=",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def test_request_auto_headers():
|
||||
request = Request("GET", "https://www.example.org/")
|
||||
request = httpx.Request("GET", "https://www.example.org/")
|
||||
assert "host" in request.headers
|
||||
|
||||
@ -1,38 +1,38 @@
|
||||
from httpx import URL, AsyncClient, Cookies, Headers, Timeout
|
||||
import httpx
|
||||
|
||||
|
||||
def test_client_base_url():
|
||||
client = AsyncClient()
|
||||
client = httpx.Client()
|
||||
client.base_url = "https://www.example.org/" # type: ignore
|
||||
assert isinstance(client.base_url, URL)
|
||||
assert client.base_url == URL("https://www.example.org/")
|
||||
assert isinstance(client.base_url, httpx.URL)
|
||||
assert client.base_url == httpx.URL("https://www.example.org/")
|
||||
|
||||
|
||||
def test_client_base_url_without_trailing_slash():
|
||||
client = AsyncClient()
|
||||
client = httpx.Client()
|
||||
client.base_url = "https://www.example.org/path" # type: ignore
|
||||
assert isinstance(client.base_url, URL)
|
||||
assert client.base_url == URL("https://www.example.org/path/")
|
||||
assert isinstance(client.base_url, httpx.URL)
|
||||
assert client.base_url == httpx.URL("https://www.example.org/path/")
|
||||
|
||||
|
||||
def test_client_base_url_with_trailing_slash():
|
||||
client = AsyncClient()
|
||||
client = httpx.Client()
|
||||
client.base_url = "https://www.example.org/path/" # type: ignore
|
||||
assert isinstance(client.base_url, URL)
|
||||
assert client.base_url == URL("https://www.example.org/path/")
|
||||
assert isinstance(client.base_url, httpx.URL)
|
||||
assert client.base_url == httpx.URL("https://www.example.org/path/")
|
||||
|
||||
|
||||
def test_client_headers():
|
||||
client = AsyncClient()
|
||||
client = httpx.Client()
|
||||
client.headers = {"a": "b"} # type: ignore
|
||||
assert isinstance(client.headers, Headers)
|
||||
assert isinstance(client.headers, httpx.Headers)
|
||||
assert client.headers["A"] == "b"
|
||||
|
||||
|
||||
def test_client_cookies():
|
||||
client = AsyncClient()
|
||||
client = httpx.Client()
|
||||
client.cookies = {"a": "b"} # type: ignore
|
||||
assert isinstance(client.cookies, Cookies)
|
||||
assert isinstance(client.cookies, httpx.Cookies)
|
||||
mycookies = list(client.cookies.jar)
|
||||
assert len(mycookies) == 1
|
||||
assert mycookies[0].name == "a" and mycookies[0].value == "b"
|
||||
@ -40,11 +40,11 @@ def test_client_cookies():
|
||||
|
||||
def test_client_timeout():
|
||||
expected_timeout = 12.0
|
||||
client = AsyncClient()
|
||||
client = httpx.Client()
|
||||
|
||||
client.timeout = expected_timeout # type: ignore
|
||||
|
||||
assert isinstance(client.timeout, Timeout)
|
||||
assert isinstance(client.timeout, httpx.Timeout)
|
||||
assert client.timeout.connect == expected_timeout
|
||||
assert client.timeout.read == expected_timeout
|
||||
assert client.timeout.write == expected_timeout
|
||||
|
||||
@ -37,13 +37,13 @@ def url_to_origin(url: str):
|
||||
],
|
||||
)
|
||||
def test_proxies_parameter(proxies, expected_proxies):
|
||||
client = httpx.AsyncClient(proxies=proxies)
|
||||
client = httpx.Client(proxies=proxies)
|
||||
|
||||
for proxy_key, url in expected_proxies:
|
||||
pattern = URLPattern(proxy_key)
|
||||
assert pattern in client._proxies
|
||||
proxy = client._proxies[pattern]
|
||||
assert isinstance(proxy, httpcore.AsyncHTTPProxy)
|
||||
assert isinstance(proxy, httpcore.SyncHTTPProxy)
|
||||
assert proxy.proxy_origin == url_to_origin(url)
|
||||
|
||||
assert len(expected_proxies) == len(client._proxies)
|
||||
@ -110,30 +110,34 @@ PROXY_URL = "http://[::1]"
|
||||
],
|
||||
)
|
||||
def test_transport_for_request(url, proxies, expected):
|
||||
client = httpx.AsyncClient(proxies=proxies)
|
||||
client = httpx.Client(proxies=proxies)
|
||||
transport = client._transport_for_url(httpx.URL(url))
|
||||
|
||||
if expected is None:
|
||||
assert transport is client._transport
|
||||
else:
|
||||
assert isinstance(transport, httpcore.AsyncHTTPProxy)
|
||||
assert isinstance(transport, httpcore.SyncHTTPProxy)
|
||||
assert transport.proxy_origin == url_to_origin(expected)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_proxy_close():
|
||||
client = httpx.AsyncClient(proxies={"all://": PROXY_URL})
|
||||
await client.aclose()
|
||||
try:
|
||||
client = httpx.AsyncClient(proxies={"all://": PROXY_URL})
|
||||
finally:
|
||||
await client.aclose()
|
||||
|
||||
|
||||
def test_sync_proxy_close():
|
||||
client = httpx.Client(proxies={"all://": PROXY_URL})
|
||||
client.close()
|
||||
try:
|
||||
client = httpx.Client(proxies={"all://": PROXY_URL})
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
|
||||
def test_unsupported_proxy_scheme():
|
||||
with pytest.raises(ValueError):
|
||||
httpx.AsyncClient(proxies="ftp://127.0.0.1")
|
||||
httpx.Client(proxies="ftp://127.0.0.1")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
@ -257,7 +261,7 @@ def test_proxies_environ(monkeypatch, client_class, url, env, expected):
|
||||
)
|
||||
def test_for_deprecated_proxy_params(proxies, expected_scheme):
|
||||
with pytest.deprecated_call() as block:
|
||||
httpx.AsyncClient(proxies=proxies)
|
||||
httpx.Client(proxies=proxies)
|
||||
|
||||
warning_message = str(block.pop(DeprecationWarning))
|
||||
|
||||
|
||||
@ -1,19 +1,18 @@
|
||||
import typing
|
||||
|
||||
import httpcore
|
||||
import pytest
|
||||
|
||||
from httpx import URL, AsyncClient, QueryParams
|
||||
import httpx
|
||||
from httpx._content_streams import ContentStream, JSONStream
|
||||
|
||||
|
||||
class MockTransport(httpcore.AsyncHTTPTransport):
|
||||
async def request(
|
||||
class MockTransport(httpcore.SyncHTTPTransport):
|
||||
def request(
|
||||
self,
|
||||
method: bytes,
|
||||
url: typing.Tuple[bytes, bytes, typing.Optional[int], bytes],
|
||||
headers: typing.List[typing.Tuple[bytes, bytes]] = None,
|
||||
stream: httpcore.AsyncByteStream = None,
|
||||
stream: httpcore.SyncByteStream = None,
|
||||
timeout: typing.Mapping[str, typing.Optional[float]] = None,
|
||||
) -> typing.Tuple[
|
||||
bytes, int, bytes, typing.List[typing.Tuple[bytes, bytes]], ContentStream
|
||||
@ -23,31 +22,30 @@ class MockTransport(httpcore.AsyncHTTPTransport):
|
||||
|
||||
|
||||
def test_client_queryparams():
|
||||
client = AsyncClient(params={"a": "b"})
|
||||
assert isinstance(client.params, QueryParams)
|
||||
client = httpx.Client(params={"a": "b"})
|
||||
assert isinstance(client.params, httpx.QueryParams)
|
||||
assert client.params["a"] == "b"
|
||||
|
||||
|
||||
def test_client_queryparams_string():
|
||||
client = AsyncClient(params="a=b")
|
||||
assert isinstance(client.params, QueryParams)
|
||||
client = httpx.Client(params="a=b")
|
||||
assert isinstance(client.params, httpx.QueryParams)
|
||||
assert client.params["a"] == "b"
|
||||
|
||||
client = AsyncClient()
|
||||
client = httpx.Client()
|
||||
client.params = "a=b" # type: ignore
|
||||
assert isinstance(client.params, QueryParams)
|
||||
assert isinstance(client.params, httpx.QueryParams)
|
||||
assert client.params["a"] == "b"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_client_queryparams_echo():
|
||||
def test_client_queryparams_echo():
|
||||
url = "http://example.org/echo_queryparams"
|
||||
client_queryparams = "first=str"
|
||||
request_queryparams = {"second": "dict"}
|
||||
client = AsyncClient(transport=MockTransport(), params=client_queryparams)
|
||||
response = await client.get(url, params=request_queryparams)
|
||||
client = httpx.Client(transport=MockTransport(), params=client_queryparams)
|
||||
response = client.get(url, params=request_queryparams)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.url == URL(
|
||||
assert response.url == httpx.URL(
|
||||
"http://example.org/echo_queryparams?first=str&second=dict"
|
||||
)
|
||||
|
||||
@ -5,18 +5,8 @@ from urllib.parse import parse_qs
|
||||
import httpcore
|
||||
import pytest
|
||||
|
||||
from httpx import (
|
||||
URL,
|
||||
AsyncClient,
|
||||
Client,
|
||||
NotRedirectResponse,
|
||||
RemoteProtocolError,
|
||||
RequestBodyUnavailable,
|
||||
TooManyRedirects,
|
||||
UnsupportedProtocol,
|
||||
codes,
|
||||
)
|
||||
from httpx._content_streams import AsyncIteratorStream, ByteStream, ContentStream
|
||||
import httpx
|
||||
from httpx._content_streams import ByteStream, ContentStream, IteratorStream
|
||||
|
||||
|
||||
def get_header_value(headers, key, default=None):
|
||||
@ -44,45 +34,45 @@ class MockTransport:
|
||||
|
||||
path, _, query = path.partition(b"?")
|
||||
if path == b"/no_redirect":
|
||||
return b"HTTP/1.1", codes.OK, b"OK", [], ByteStream(b"")
|
||||
return b"HTTP/1.1", httpx.codes.OK, b"OK", [], ByteStream(b"")
|
||||
|
||||
elif path == b"/redirect_301":
|
||||
|
||||
async def body():
|
||||
def body():
|
||||
yield b"<a href='https://example.org/'>here</a>"
|
||||
|
||||
status_code = codes.MOVED_PERMANENTLY
|
||||
status_code = httpx.codes.MOVED_PERMANENTLY
|
||||
headers = [(b"location", b"https://example.org/")]
|
||||
stream = AsyncIteratorStream(aiterator=body())
|
||||
stream = IteratorStream(iterator=body())
|
||||
return b"HTTP/1.1", status_code, b"Moved Permanently", headers, stream
|
||||
|
||||
elif path == b"/redirect_302":
|
||||
status_code = codes.FOUND
|
||||
status_code = httpx.codes.FOUND
|
||||
headers = [(b"location", b"https://example.org/")]
|
||||
return b"HTTP/1.1", status_code, b"Found", headers, ByteStream(b"")
|
||||
|
||||
elif path == b"/redirect_303":
|
||||
status_code = codes.SEE_OTHER
|
||||
status_code = httpx.codes.SEE_OTHER
|
||||
headers = [(b"location", b"https://example.org/")]
|
||||
return b"HTTP/1.1", status_code, b"See Other", headers, ByteStream(b"")
|
||||
|
||||
elif path == b"/relative_redirect":
|
||||
status_code = codes.SEE_OTHER
|
||||
status_code = httpx.codes.SEE_OTHER
|
||||
headers = [(b"location", b"/")]
|
||||
return b"HTTP/1.1", status_code, b"See Other", headers, ByteStream(b"")
|
||||
|
||||
elif path == b"/malformed_redirect":
|
||||
status_code = codes.SEE_OTHER
|
||||
status_code = httpx.codes.SEE_OTHER
|
||||
headers = [(b"location", b"https://:443/")]
|
||||
return b"HTTP/1.1", status_code, b"See Other", headers, ByteStream(b"")
|
||||
|
||||
elif path == b"/invalid_redirect":
|
||||
status_code = codes.SEE_OTHER
|
||||
status_code = httpx.codes.SEE_OTHER
|
||||
headers = [(b"location", "https://😇/".encode("utf-8"))]
|
||||
return b"HTTP/1.1", status_code, b"See Other", headers, ByteStream(b"")
|
||||
|
||||
elif path == b"/no_scheme_redirect":
|
||||
status_code = codes.SEE_OTHER
|
||||
status_code = httpx.codes.SEE_OTHER
|
||||
headers = [(b"location", b"//example.org/")]
|
||||
return b"HTTP/1.1", status_code, b"See Other", headers, ByteStream(b"")
|
||||
|
||||
@ -90,7 +80,7 @@ class MockTransport:
|
||||
params = parse_qs(query.decode("ascii"))
|
||||
count = int(params.get("count", "0")[0])
|
||||
redirect_count = count - 1
|
||||
code = codes.SEE_OTHER if count else codes.OK
|
||||
code = httpx.codes.SEE_OTHER if count else httpx.codes.OK
|
||||
phrase = b"See Other" if count else b"OK"
|
||||
location = b"/multiple_redirects"
|
||||
if redirect_count:
|
||||
@ -99,12 +89,12 @@ class MockTransport:
|
||||
return b"HTTP/1.1", code, phrase, headers, ByteStream(b"")
|
||||
|
||||
if path == b"/redirect_loop":
|
||||
code = codes.SEE_OTHER
|
||||
code = httpx.codes.SEE_OTHER
|
||||
headers = [(b"location", b"/redirect_loop")]
|
||||
return b"HTTP/1.1", code, b"See Other", headers, ByteStream(b"")
|
||||
|
||||
elif path == b"/cross_domain":
|
||||
code = codes.SEE_OTHER
|
||||
code = httpx.codes.SEE_OTHER
|
||||
headers = [(b"location", b"https://example.org/cross_domain_target")]
|
||||
return b"HTTP/1.1", code, b"See Other", headers, ByteStream(b"")
|
||||
|
||||
@ -116,12 +106,12 @@ class MockTransport:
|
||||
return b"HTTP/1.1", 200, b"OK", [], stream
|
||||
|
||||
elif path == b"/redirect_body":
|
||||
code = codes.PERMANENT_REDIRECT
|
||||
code = httpx.codes.PERMANENT_REDIRECT
|
||||
headers = [(b"location", b"/redirect_body_target")]
|
||||
return b"HTTP/1.1", code, b"Permanent Redirect", headers, ByteStream(b"")
|
||||
|
||||
elif path == b"/redirect_no_body":
|
||||
code = codes.SEE_OTHER
|
||||
code = httpx.codes.SEE_OTHER
|
||||
headers = [(b"location", b"/redirect_body_target")]
|
||||
return b"HTTP/1.1", code, b"See Other", headers, ByteStream(b"")
|
||||
|
||||
@ -141,7 +131,7 @@ class MockTransport:
|
||||
headers = [(b"location", b"https://www.example.org/cross_subdomain")]
|
||||
return (
|
||||
b"HTTP/1.1",
|
||||
codes.PERMANENT_REDIRECT,
|
||||
httpx.codes.PERMANENT_REDIRECT,
|
||||
b"Permanent Redirect",
|
||||
headers,
|
||||
ByteStream(b""),
|
||||
@ -150,7 +140,7 @@ class MockTransport:
|
||||
return b"HTTP/1.1", 200, b"OK", [], ByteStream(b"Hello, world!")
|
||||
|
||||
elif path == b"/redirect_custom_scheme":
|
||||
status_code = codes.MOVED_PERMANENTLY
|
||||
status_code = httpx.codes.MOVED_PERMANENTLY
|
||||
headers = [(b"location", b"market://details?id=42")]
|
||||
return (
|
||||
b"HTTP/1.1",
|
||||
@ -183,130 +173,116 @@ class SyncMockTransport(MockTransport, httpcore.SyncHTTPTransport):
|
||||
return self._request(*args, **kwargs)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_no_redirect():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
def test_no_redirect():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
url = "https://example.com/no_redirect"
|
||||
response = await client.get(url)
|
||||
response = client.get(url)
|
||||
assert response.status_code == 200
|
||||
with pytest.raises(NotRedirectResponse):
|
||||
await response.anext()
|
||||
with pytest.raises(httpx.NotRedirectResponse):
|
||||
response.next()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_redirect_301():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
response = await client.post("https://example.org/redirect_301")
|
||||
assert response.status_code == codes.OK
|
||||
assert response.url == URL("https://example.org/")
|
||||
def test_redirect_301():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
response = client.post("https://example.org/redirect_301")
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == httpx.URL("https://example.org/")
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_redirect_302():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
response = await client.post("https://example.org/redirect_302")
|
||||
assert response.status_code == codes.OK
|
||||
assert response.url == URL("https://example.org/")
|
||||
def test_redirect_302():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
response = client.post("https://example.org/redirect_302")
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == httpx.URL("https://example.org/")
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_redirect_303():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
response = await client.get("https://example.org/redirect_303")
|
||||
assert response.status_code == codes.OK
|
||||
assert response.url == URL("https://example.org/")
|
||||
def test_redirect_303():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
response = client.get("https://example.org/redirect_303")
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == httpx.URL("https://example.org/")
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_disallow_redirects():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
response = await client.post(
|
||||
"https://example.org/redirect_303", allow_redirects=False
|
||||
)
|
||||
assert response.status_code == codes.SEE_OTHER
|
||||
assert response.url == URL("https://example.org/redirect_303")
|
||||
def test_disallow_redirects():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
response = client.post("https://example.org/redirect_303", allow_redirects=False)
|
||||
assert response.status_code == httpx.codes.SEE_OTHER
|
||||
assert response.url == httpx.URL("https://example.org/redirect_303")
|
||||
assert response.is_redirect is True
|
||||
assert len(response.history) == 0
|
||||
|
||||
response = await response.anext()
|
||||
assert response.status_code == codes.OK
|
||||
assert response.url == URL("https://example.org/")
|
||||
response = response.next()
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == httpx.URL("https://example.org/")
|
||||
assert response.is_redirect is False
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_head_redirect():
|
||||
def test_head_redirect():
|
||||
"""
|
||||
Contrary to Requests, redirects remain enabled by default for HEAD requests.
|
||||
"""
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
response = await client.head("https://example.org/redirect_302")
|
||||
assert response.status_code == codes.OK
|
||||
assert response.url == URL("https://example.org/")
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
response = client.head("https://example.org/redirect_302")
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == httpx.URL("https://example.org/")
|
||||
assert response.request.method == "HEAD"
|
||||
assert len(response.history) == 1
|
||||
assert response.text == ""
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_relative_redirect():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
response = await client.get("https://example.org/relative_redirect")
|
||||
assert response.status_code == codes.OK
|
||||
assert response.url == URL("https://example.org/")
|
||||
def test_relative_redirect():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
response = client.get("https://example.org/relative_redirect")
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == httpx.URL("https://example.org/")
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_malformed_redirect():
|
||||
def test_malformed_redirect():
|
||||
# https://github.com/encode/httpx/issues/771
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
response = await client.get("http://example.org/malformed_redirect")
|
||||
assert response.status_code == codes.OK
|
||||
assert response.url == URL("https://example.org:443/")
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
response = client.get("http://example.org/malformed_redirect")
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == httpx.URL("https://example.org:443/")
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_invalid_redirect():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
with pytest.raises(RemoteProtocolError):
|
||||
await client.get("http://example.org/invalid_redirect")
|
||||
def test_invalid_redirect():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
with pytest.raises(httpx.RemoteProtocolError):
|
||||
client.get("http://example.org/invalid_redirect")
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_no_scheme_redirect():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
response = await client.get("https://example.org/no_scheme_redirect")
|
||||
assert response.status_code == codes.OK
|
||||
assert response.url == URL("https://example.org/")
|
||||
def test_no_scheme_redirect():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
response = client.get("https://example.org/no_scheme_redirect")
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == httpx.URL("https://example.org/")
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_fragment_redirect():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
response = await client.get("https://example.org/relative_redirect#fragment")
|
||||
assert response.status_code == codes.OK
|
||||
assert response.url == URL("https://example.org/#fragment")
|
||||
def test_fragment_redirect():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
response = client.get("https://example.org/relative_redirect#fragment")
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == httpx.URL("https://example.org/#fragment")
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_multiple_redirects():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
response = await client.get("https://example.org/multiple_redirects?count=20")
|
||||
assert response.status_code == codes.OK
|
||||
assert response.url == URL("https://example.org/multiple_redirects")
|
||||
def test_multiple_redirects():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
response = client.get("https://example.org/multiple_redirects?count=20")
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == httpx.URL("https://example.org/multiple_redirects")
|
||||
assert len(response.history) == 20
|
||||
assert response.history[0].url == URL(
|
||||
assert response.history[0].url == httpx.URL(
|
||||
"https://example.org/multiple_redirects?count=20"
|
||||
)
|
||||
assert response.history[1].url == URL(
|
||||
assert response.history[1].url == httpx.URL(
|
||||
"https://example.org/multiple_redirects?count=19"
|
||||
)
|
||||
assert len(response.history[0].history) == 0
|
||||
@ -315,128 +291,120 @@ async def test_multiple_redirects():
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_async_too_many_redirects():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
with pytest.raises(TooManyRedirects):
|
||||
await client.get("https://example.org/multiple_redirects?count=21")
|
||||
async with httpx.AsyncClient(transport=AsyncMockTransport()) as client:
|
||||
with pytest.raises(httpx.TooManyRedirects):
|
||||
await client.get("https://example.org/multiple_redirects?count=21")
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_async_too_many_redirects_calling_next():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
url = "https://example.org/multiple_redirects?count=21"
|
||||
response = await client.get(url, allow_redirects=False)
|
||||
with pytest.raises(TooManyRedirects):
|
||||
while response.is_redirect:
|
||||
response = await response.anext()
|
||||
async with httpx.AsyncClient(transport=AsyncMockTransport()) as client:
|
||||
url = "https://example.org/multiple_redirects?count=21"
|
||||
response = await client.get(url, allow_redirects=False)
|
||||
with pytest.raises(httpx.TooManyRedirects):
|
||||
while response.is_redirect:
|
||||
response = await response.anext()
|
||||
|
||||
|
||||
def test_sync_too_many_redirects():
|
||||
client = Client(transport=SyncMockTransport())
|
||||
with pytest.raises(TooManyRedirects):
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
with pytest.raises(httpx.TooManyRedirects):
|
||||
client.get("https://example.org/multiple_redirects?count=21")
|
||||
|
||||
|
||||
def test_sync_too_many_redirects_calling_next():
|
||||
client = Client(transport=SyncMockTransport())
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
url = "https://example.org/multiple_redirects?count=21"
|
||||
response = client.get(url, allow_redirects=False)
|
||||
with pytest.raises(TooManyRedirects):
|
||||
with pytest.raises(httpx.TooManyRedirects):
|
||||
while response.is_redirect:
|
||||
response = response.next()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_redirect_loop():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
with pytest.raises(TooManyRedirects):
|
||||
await client.get("https://example.org/redirect_loop")
|
||||
def test_redirect_loop():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
with pytest.raises(httpx.TooManyRedirects):
|
||||
client.get("https://example.org/redirect_loop")
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_cross_domain_redirect():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
def test_cross_domain_redirect():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
url = "https://example.com/cross_domain"
|
||||
headers = {"Authorization": "abc"}
|
||||
response = await client.get(url, headers=headers)
|
||||
assert response.url == URL("https://example.org/cross_domain_target")
|
||||
response = client.get(url, headers=headers)
|
||||
assert response.url == httpx.URL("https://example.org/cross_domain_target")
|
||||
assert "authorization" not in response.json()["headers"]
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_same_domain_redirect():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
def test_same_domain_redirect():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
url = "https://example.org/cross_domain"
|
||||
headers = {"Authorization": "abc"}
|
||||
response = await client.get(url, headers=headers)
|
||||
assert response.url == URL("https://example.org/cross_domain_target")
|
||||
response = client.get(url, headers=headers)
|
||||
assert response.url == httpx.URL("https://example.org/cross_domain_target")
|
||||
assert response.json()["headers"]["authorization"] == "abc"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_body_redirect():
|
||||
def test_body_redirect():
|
||||
"""
|
||||
A 308 redirect should preserve the request body.
|
||||
"""
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
url = "https://example.org/redirect_body"
|
||||
data = b"Example request body"
|
||||
response = await client.post(url, data=data)
|
||||
assert response.url == URL("https://example.org/redirect_body_target")
|
||||
response = client.post(url, data=data)
|
||||
assert response.url == httpx.URL("https://example.org/redirect_body_target")
|
||||
assert response.json()["body"] == "Example request body"
|
||||
assert "content-length" in response.json()["headers"]
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_no_body_redirect():
|
||||
def test_no_body_redirect():
|
||||
"""
|
||||
A 303 redirect should remove the request body.
|
||||
"""
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
url = "https://example.org/redirect_no_body"
|
||||
data = b"Example request body"
|
||||
response = await client.post(url, data=data)
|
||||
assert response.url == URL("https://example.org/redirect_body_target")
|
||||
response = client.post(url, data=data)
|
||||
assert response.url == httpx.URL("https://example.org/redirect_body_target")
|
||||
assert response.json()["body"] == ""
|
||||
assert "content-length" not in response.json()["headers"]
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_can_stream_if_no_redirect():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
def test_can_stream_if_no_redirect():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
url = "https://example.org/redirect_301"
|
||||
async with client.stream("GET", url, allow_redirects=False) as response:
|
||||
with client.stream("GET", url, allow_redirects=False) as response:
|
||||
assert not response.is_closed
|
||||
assert response.status_code == codes.MOVED_PERMANENTLY
|
||||
assert response.status_code == httpx.codes.MOVED_PERMANENTLY
|
||||
assert response.headers["location"] == "https://example.org/"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_cannot_redirect_streaming_body():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
def test_cannot_redirect_streaming_body():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
url = "https://example.org/redirect_body"
|
||||
|
||||
async def streaming_body():
|
||||
def streaming_body():
|
||||
yield b"Example request body" # pragma: nocover
|
||||
|
||||
with pytest.raises(RequestBodyUnavailable):
|
||||
await client.post(url, data=streaming_body())
|
||||
with pytest.raises(httpx.RequestBodyUnavailable):
|
||||
client.post(url, data=streaming_body())
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_cross_subdomain_redirect():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
def test_cross_subdomain_redirect():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
url = "https://example.com/cross_subdomain"
|
||||
response = await client.get(url)
|
||||
assert response.url == URL("https://www.example.org/cross_subdomain")
|
||||
response = client.get(url)
|
||||
assert response.url == httpx.URL("https://www.example.org/cross_subdomain")
|
||||
|
||||
|
||||
class MockCookieTransport(httpcore.AsyncHTTPTransport):
|
||||
async def request(
|
||||
class MockCookieTransport(httpcore.SyncHTTPTransport):
|
||||
def request(
|
||||
self,
|
||||
method: bytes,
|
||||
url: typing.Tuple[bytes, bytes, typing.Optional[int], bytes],
|
||||
headers: typing.List[typing.Tuple[bytes, bytes]] = None,
|
||||
stream: httpcore.AsyncByteStream = None,
|
||||
stream: httpcore.SyncByteStream = None,
|
||||
timeout: typing.Mapping[str, typing.Optional[float]] = None,
|
||||
) -> typing.Tuple[
|
||||
bytes, int, bytes, typing.List[typing.Tuple[bytes, bytes]], ContentStream
|
||||
@ -451,7 +419,7 @@ class MockCookieTransport(httpcore.AsyncHTTPTransport):
|
||||
return b"HTTP/1.1", 200, b"OK", [], ByteStream(content)
|
||||
|
||||
elif path == b"/login":
|
||||
status_code = codes.SEE_OTHER
|
||||
status_code = httpx.codes.SEE_OTHER
|
||||
headers = [
|
||||
(b"location", b"/"),
|
||||
(
|
||||
@ -466,7 +434,7 @@ class MockCookieTransport(httpcore.AsyncHTTPTransport):
|
||||
|
||||
else:
|
||||
assert path == b"/logout"
|
||||
status_code = codes.SEE_OTHER
|
||||
status_code = httpx.codes.SEE_OTHER
|
||||
headers = [
|
||||
(b"location", b"/"),
|
||||
(
|
||||
@ -480,39 +448,37 @@ class MockCookieTransport(httpcore.AsyncHTTPTransport):
|
||||
return b"HTTP/1.1", status_code, b"See Other", headers, ByteStream(b"")
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_redirect_cookie_behavior():
|
||||
client = AsyncClient(transport=MockCookieTransport())
|
||||
def test_redirect_cookie_behavior():
|
||||
client = httpx.Client(transport=MockCookieTransport())
|
||||
|
||||
# The client is not logged in.
|
||||
response = await client.get("https://example.com/")
|
||||
response = client.get("https://example.com/")
|
||||
assert response.url == "https://example.com/"
|
||||
assert response.text == "Not logged in"
|
||||
|
||||
# Login redirects to the homepage, setting a session cookie.
|
||||
response = await client.post("https://example.com/login")
|
||||
response = client.post("https://example.com/login")
|
||||
assert response.url == "https://example.com/"
|
||||
assert response.text == "Logged in"
|
||||
|
||||
# The client is logged in.
|
||||
response = await client.get("https://example.com/")
|
||||
response = client.get("https://example.com/")
|
||||
assert response.url == "https://example.com/"
|
||||
assert response.text == "Logged in"
|
||||
|
||||
# Logout redirects to the homepage, expiring the session cookie.
|
||||
response = await client.post("https://example.com/logout")
|
||||
response = client.post("https://example.com/logout")
|
||||
assert response.url == "https://example.com/"
|
||||
assert response.text == "Not logged in"
|
||||
|
||||
# The client is not logged in.
|
||||
response = await client.get("https://example.com/")
|
||||
response = client.get("https://example.com/")
|
||||
assert response.url == "https://example.com/"
|
||||
assert response.text == "Not logged in"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("async_environment")
|
||||
async def test_redirect_custom_scheme():
|
||||
client = AsyncClient(transport=AsyncMockTransport())
|
||||
with pytest.raises(UnsupportedProtocol) as e:
|
||||
await client.post("https://example.org/redirect_custom_scheme")
|
||||
def test_redirect_custom_scheme():
|
||||
client = httpx.Client(transport=SyncMockTransport())
|
||||
with pytest.raises(httpx.UnsupportedProtocol) as e:
|
||||
client.post("https://example.org/redirect_custom_scheme")
|
||||
assert str(e.value) == "Scheme b'market' not supported."
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
import pytest
|
||||
|
||||
from httpx import QueryParams
|
||||
import httpx
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
@ -12,7 +12,7 @@ from httpx import QueryParams
|
||||
],
|
||||
)
|
||||
def test_queryparams(source):
|
||||
q = QueryParams(source)
|
||||
q = httpx.QueryParams(source)
|
||||
assert "a" in q
|
||||
assert "A" not in q
|
||||
assert "c" not in q
|
||||
@ -32,59 +32,63 @@ def test_queryparams(source):
|
||||
assert dict(q) == {"a": "456", "b": "789"}
|
||||
assert str(q) == "a=123&a=456&b=789"
|
||||
assert repr(q) == "QueryParams('a=123&a=456&b=789')"
|
||||
assert QueryParams({"a": "123", "b": "456"}) == QueryParams(
|
||||
assert httpx.QueryParams({"a": "123", "b": "456"}) == httpx.QueryParams(
|
||||
[("a", "123"), ("b", "456")]
|
||||
)
|
||||
assert QueryParams({"a": "123", "b": "456"}) == QueryParams("a=123&b=456")
|
||||
assert QueryParams({"a": "123", "b": "456"}) == QueryParams(
|
||||
assert httpx.QueryParams({"a": "123", "b": "456"}) == httpx.QueryParams(
|
||||
"a=123&b=456"
|
||||
)
|
||||
assert httpx.QueryParams({"a": "123", "b": "456"}) == httpx.QueryParams(
|
||||
{"b": "456", "a": "123"}
|
||||
)
|
||||
assert QueryParams() == QueryParams({})
|
||||
assert QueryParams([("a", "123"), ("a", "456")]) == QueryParams("a=123&a=456")
|
||||
assert QueryParams({"a": "123", "b": "456"}) != "invalid"
|
||||
assert httpx.QueryParams() == httpx.QueryParams({})
|
||||
assert httpx.QueryParams([("a", "123"), ("a", "456")]) == httpx.QueryParams(
|
||||
"a=123&a=456"
|
||||
)
|
||||
assert httpx.QueryParams({"a": "123", "b": "456"}) != "invalid"
|
||||
|
||||
q = QueryParams([("a", "123"), ("a", "456")])
|
||||
assert QueryParams(q) == q
|
||||
q = httpx.QueryParams([("a", "123"), ("a", "456")])
|
||||
assert httpx.QueryParams(q) == q
|
||||
|
||||
|
||||
def test_queryparam_types():
|
||||
q = QueryParams(None)
|
||||
q = httpx.QueryParams(None)
|
||||
assert str(q) == ""
|
||||
|
||||
q = QueryParams({"a": True})
|
||||
q = httpx.QueryParams({"a": True})
|
||||
assert str(q) == "a=true"
|
||||
|
||||
q = QueryParams({"a": False})
|
||||
q = httpx.QueryParams({"a": False})
|
||||
assert str(q) == "a=false"
|
||||
|
||||
q = QueryParams({"a": ""})
|
||||
q = httpx.QueryParams({"a": ""})
|
||||
assert str(q) == "a="
|
||||
|
||||
q = QueryParams({"a": None})
|
||||
q = httpx.QueryParams({"a": None})
|
||||
assert str(q) == "a="
|
||||
|
||||
q = QueryParams({"a": 1.23})
|
||||
q = httpx.QueryParams({"a": 1.23})
|
||||
assert str(q) == "a=1.23"
|
||||
|
||||
q = QueryParams({"a": 123})
|
||||
q = httpx.QueryParams({"a": 123})
|
||||
assert str(q) == "a=123"
|
||||
|
||||
q = QueryParams({"a": [1, 2]})
|
||||
q = httpx.QueryParams({"a": [1, 2]})
|
||||
assert str(q) == "a=1&a=2"
|
||||
|
||||
|
||||
def test_queryparam_setters():
|
||||
q = QueryParams({"a": 1})
|
||||
q = httpx.QueryParams({"a": 1})
|
||||
q.update([])
|
||||
|
||||
assert str(q) == "a=1"
|
||||
|
||||
q = QueryParams([("a", 1), ("a", 2)])
|
||||
q = httpx.QueryParams([("a", 1), ("a", 2)])
|
||||
q["a"] = "3"
|
||||
assert str(q) == "a=3"
|
||||
|
||||
q = QueryParams([("a", 1), ("b", 1)])
|
||||
u = QueryParams([("b", 2), ("b", 3)])
|
||||
q = httpx.QueryParams([("a", 1), ("b", 1)])
|
||||
u = httpx.QueryParams([("b", 2), ("b", 3)])
|
||||
q.update(u)
|
||||
|
||||
assert str(q) == "a=1&b=2&b=3"
|
||||
|
||||
@ -20,8 +20,7 @@ async def async_streaming_body():
|
||||
yield b"world!"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_response():
|
||||
def test_response():
|
||||
response = httpx.Response(200, content=b"Hello, world!", request=REQUEST)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
import pytest
|
||||
|
||||
from httpx import URL, InvalidURL
|
||||
import httpx
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
@ -53,15 +53,15 @@ from httpx import URL, InvalidURL
|
||||
],
|
||||
)
|
||||
def test_idna_url(given, idna, host, scheme, port):
|
||||
url = URL(given)
|
||||
assert url == URL(idna)
|
||||
url = httpx.URL(given)
|
||||
assert url == httpx.URL(idna)
|
||||
assert url.host == host
|
||||
assert url.scheme == scheme
|
||||
assert url.port == port
|
||||
|
||||
|
||||
def test_url():
|
||||
url = URL("https://example.org:123/path/to/somewhere?abc=123#anchor")
|
||||
url = httpx.URL("https://example.org:123/path/to/somewhere?abc=123#anchor")
|
||||
assert url.scheme == "https"
|
||||
assert url.host == "example.org"
|
||||
assert url.port == 123
|
||||
@ -74,21 +74,23 @@ def test_url():
|
||||
)
|
||||
|
||||
new = url.copy_with(scheme="http", port=None)
|
||||
assert new == URL("http://example.org/path/to/somewhere?abc=123#anchor")
|
||||
assert new == httpx.URL("http://example.org/path/to/somewhere?abc=123#anchor")
|
||||
assert new.scheme == "http"
|
||||
|
||||
|
||||
def test_url_eq_str():
|
||||
url = URL("https://example.org:123/path/to/somewhere?abc=123#anchor")
|
||||
url = httpx.URL("https://example.org:123/path/to/somewhere?abc=123#anchor")
|
||||
assert url == "https://example.org:123/path/to/somewhere?abc=123#anchor"
|
||||
assert str(url) == url
|
||||
|
||||
|
||||
def test_url_params():
|
||||
url = URL("https://example.org:123/path/to/somewhere", params={"a": "123"})
|
||||
url = httpx.URL("https://example.org:123/path/to/somewhere", params={"a": "123"})
|
||||
assert str(url) == "https://example.org:123/path/to/somewhere?a=123"
|
||||
|
||||
url = URL("https://example.org:123/path/to/somewhere?b=456", params={"a": "123"})
|
||||
url = httpx.URL(
|
||||
"https://example.org:123/path/to/somewhere?b=456", params={"a": "123"}
|
||||
)
|
||||
assert str(url) == "https://example.org:123/path/to/somewhere?b=456&a=123"
|
||||
|
||||
|
||||
@ -96,7 +98,7 @@ def test_url_join():
|
||||
"""
|
||||
Some basic URL joining tests.
|
||||
"""
|
||||
url = URL("https://example.org:123/path/to/somewhere")
|
||||
url = httpx.URL("https://example.org:123/path/to/somewhere")
|
||||
assert url.join("/somewhere-else") == "https://example.org:123/somewhere-else"
|
||||
assert (
|
||||
url.join("somewhere-else") == "https://example.org:123/path/to/somewhere-else"
|
||||
@ -114,7 +116,7 @@ def test_url_join_rfc3986():
|
||||
https://tools.ietf.org/html/rfc3986#section-5.4
|
||||
"""
|
||||
|
||||
url = URL("http://example.com/b/c/d;p?q")
|
||||
url = httpx.URL("http://example.com/b/c/d;p?q")
|
||||
|
||||
assert url.join("g") == "http://example.com/b/c/g"
|
||||
assert url.join("./g") == "http://example.com/b/c/g"
|
||||
@ -164,8 +166,8 @@ def test_url_join_rfc3986():
|
||||
|
||||
def test_url_set():
|
||||
urls = (
|
||||
URL("http://example.org:123/path/to/somewhere"),
|
||||
URL("http://example.org:123/path/to/somewhere/else"),
|
||||
httpx.URL("http://example.org:123/path/to/somewhere"),
|
||||
httpx.URL("http://example.org:123/path/to/somewhere/else"),
|
||||
)
|
||||
|
||||
url_set = set(urls)
|
||||
@ -180,7 +182,7 @@ def test_url_copywith_for_authority():
|
||||
"port": 444,
|
||||
"host": "example.net",
|
||||
}
|
||||
url = URL("https://example.org")
|
||||
url = httpx.URL("https://example.org")
|
||||
new = url.copy_with(**copy_with_kwargs)
|
||||
for k, v in copy_with_kwargs.items():
|
||||
assert getattr(new, k) == v
|
||||
@ -192,7 +194,7 @@ def test_url_copywith_for_userinfo():
|
||||
"username": "tom@example.org",
|
||||
"password": "abc123@ %",
|
||||
}
|
||||
url = URL("https://example.org")
|
||||
url = httpx.URL("https://example.org")
|
||||
new = url.copy_with(**copy_with_kwargs)
|
||||
assert str(new) == "https://tom%40example.org:abc123%40%20%25@example.org"
|
||||
assert new.username == "tom@example.org"
|
||||
@ -200,5 +202,5 @@ def test_url_copywith_for_userinfo():
|
||||
|
||||
|
||||
def test_url_invalid():
|
||||
with pytest.raises(InvalidURL):
|
||||
URL("https://😇/")
|
||||
with pytest.raises(httpx.InvalidURL):
|
||||
httpx.URL("https://😇/")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user