Make all the tests from test_redirects to be async
This commit is contained in:
Kar Petrosyan
parent
e70d0b08c9
commit
4965a61830
@ -113,46 +113,41 @@ def redirects(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(200, html="<html><body>Hello, world!</body></html>")
|
||||
|
||||
|
||||
def test_redirect_301():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
response = client.post("https://example.org/redirect_301", follow_redirects=True)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/"
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
def test_redirect_302():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
response = client.post("https://example.org/redirect_302", follow_redirects=True)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/"
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
def test_redirect_303():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
response = client.get("https://example.org/redirect_303", follow_redirects=True)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/"
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
def test_next_request():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
request = client.build_request("POST", "https://example.org/redirect_303")
|
||||
response = client.send(request, follow_redirects=False)
|
||||
assert response.status_code == httpx.codes.SEE_OTHER
|
||||
assert response.url == "https://example.org/redirect_303"
|
||||
assert response.next_request is not None
|
||||
|
||||
response = client.send(response.next_request, follow_redirects=False)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/"
|
||||
assert response.next_request is None
|
||||
@pytest.mark.anyio
|
||||
async def test_redirect_301():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
response = await client.post(
|
||||
"https://example.org/redirect_301", follow_redirects=True
|
||||
)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/"
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_async_next_request():
|
||||
async def test_redirect_302():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
response = await client.post(
|
||||
"https://example.org/redirect_302", follow_redirects=True
|
||||
)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/"
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_redirect_303():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
response = await client.get(
|
||||
"https://example.org/redirect_303", follow_redirects=True
|
||||
)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/"
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_next_request():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
request = client.build_request("POST", "https://example.org/redirect_303")
|
||||
response = await client.send(request, follow_redirects=False)
|
||||
@ -166,82 +161,88 @@ async def test_async_next_request():
|
||||
assert response.next_request is None
|
||||
|
||||
|
||||
def test_head_redirect():
|
||||
@pytest.mark.anyio
|
||||
async def test_head_redirect():
|
||||
"""
|
||||
Contrary to Requests, redirects remain enabled by default for HEAD requests.
|
||||
"""
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
response = client.head("https://example.org/redirect_302", follow_redirects=True)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/"
|
||||
assert response.request.method == "HEAD"
|
||||
assert len(response.history) == 1
|
||||
assert response.text == ""
|
||||
|
||||
|
||||
def test_relative_redirect():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
response = client.get(
|
||||
"https://example.org/relative_redirect", follow_redirects=True
|
||||
)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/"
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
def test_malformed_redirect():
|
||||
# https://github.com/encode/httpx/issues/771
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
response = client.get(
|
||||
"http://example.org/malformed_redirect", follow_redirects=True
|
||||
)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org:443/"
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
def test_invalid_redirect():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
with pytest.raises(httpx.RemoteProtocolError):
|
||||
client.get("http://example.org/invalid_redirect", follow_redirects=True)
|
||||
|
||||
|
||||
def test_no_scheme_redirect():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
response = client.get(
|
||||
"https://example.org/no_scheme_redirect", follow_redirects=True
|
||||
)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/"
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
def test_fragment_redirect():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
response = client.get(
|
||||
"https://example.org/relative_redirect#fragment", follow_redirects=True
|
||||
)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/#fragment"
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
def test_multiple_redirects():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
response = client.get(
|
||||
"https://example.org/multiple_redirects?count=20", follow_redirects=True
|
||||
)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/multiple_redirects"
|
||||
assert len(response.history) == 20
|
||||
assert response.history[0].url == "https://example.org/multiple_redirects?count=20"
|
||||
assert response.history[1].url == "https://example.org/multiple_redirects?count=19"
|
||||
assert len(response.history[0].history) == 0
|
||||
assert len(response.history[1].history) == 1
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
response = await client.head(
|
||||
"https://example.org/redirect_302", follow_redirects=True
|
||||
)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/"
|
||||
assert response.request.method == "HEAD"
|
||||
assert len(response.history) == 1
|
||||
assert response.text == ""
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_async_too_many_redirects():
|
||||
async def test_relative_redirect():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
response = await client.get(
|
||||
"https://example.org/relative_redirect", follow_redirects=True
|
||||
)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/"
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_malformed_redirect():
|
||||
# https://github.com/encode/httpx/issues/771
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
response = await client.get(
|
||||
"http://example.org/malformed_redirect", follow_redirects=True
|
||||
)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org:443/"
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_no_scheme_redirect():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
response = await client.get(
|
||||
"https://example.org/no_scheme_redirect", follow_redirects=True
|
||||
)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/"
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_fragment_redirect():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
response = await client.get(
|
||||
"https://example.org/relative_redirect#fragment", follow_redirects=True
|
||||
)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/#fragment"
|
||||
assert len(response.history) == 1
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_multiple_redirects():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
response = await client.get(
|
||||
"https://example.org/multiple_redirects?count=20", follow_redirects=True
|
||||
)
|
||||
assert response.status_code == httpx.codes.OK
|
||||
assert response.url == "https://example.org/multiple_redirects"
|
||||
assert len(response.history) == 20
|
||||
assert (
|
||||
response.history[0].url == "https://example.org/multiple_redirects?count=20"
|
||||
)
|
||||
assert (
|
||||
response.history[1].url == "https://example.org/multiple_redirects?count=19"
|
||||
)
|
||||
assert len(response.history[0].history) == 0
|
||||
assert len(response.history[1].history) == 1
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_too_many_redirects():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
with pytest.raises(httpx.TooManyRedirects):
|
||||
await client.get(
|
||||
@ -249,125 +250,130 @@ async def test_async_too_many_redirects():
|
||||
)
|
||||
|
||||
|
||||
def test_sync_too_many_redirects():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
with pytest.raises(httpx.TooManyRedirects):
|
||||
client.get(
|
||||
"https://example.org/multiple_redirects?count=21", follow_redirects=True
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
async def test_redirect_loop():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
with pytest.raises(httpx.TooManyRedirects):
|
||||
await client.get("https://example.org/redirect_loop", follow_redirects=True)
|
||||
|
||||
|
||||
def test_redirect_loop():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
with pytest.raises(httpx.TooManyRedirects):
|
||||
client.get("https://example.org/redirect_loop", follow_redirects=True)
|
||||
@pytest.mark.anyio
|
||||
async def test_cross_domain_redirect_with_auth_header():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
url = "https://example.com/cross_domain"
|
||||
headers = {"Authorization": "abc"}
|
||||
response = await client.get(url, headers=headers, follow_redirects=True)
|
||||
assert response.url == "https://example.org/cross_domain_target"
|
||||
assert "authorization" not in response.json()["headers"]
|
||||
|
||||
|
||||
def test_cross_domain_redirect_with_auth_header():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
url = "https://example.com/cross_domain"
|
||||
headers = {"Authorization": "abc"}
|
||||
response = client.get(url, headers=headers, follow_redirects=True)
|
||||
assert response.url == "https://example.org/cross_domain_target"
|
||||
assert "authorization" not in response.json()["headers"]
|
||||
@pytest.mark.anyio
|
||||
async def test_cross_domain_https_redirect_with_auth_header():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
url = "http://example.com/cross_domain"
|
||||
headers = {"Authorization": "abc"}
|
||||
response = await client.get(url, headers=headers, follow_redirects=True)
|
||||
assert response.url == "https://example.org/cross_domain_target"
|
||||
assert "authorization" not in response.json()["headers"]
|
||||
|
||||
|
||||
def test_cross_domain_https_redirect_with_auth_header():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
url = "http://example.com/cross_domain"
|
||||
headers = {"Authorization": "abc"}
|
||||
response = client.get(url, headers=headers, follow_redirects=True)
|
||||
assert response.url == "https://example.org/cross_domain_target"
|
||||
assert "authorization" not in response.json()["headers"]
|
||||
@pytest.mark.anyio
|
||||
async def test_cross_domain_redirect_with_auth():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
url = "https://example.com/cross_domain"
|
||||
response = await client.get(url, auth=("user", "pass"), follow_redirects=True)
|
||||
assert response.url == "https://example.org/cross_domain_target"
|
||||
assert "authorization" not in response.json()["headers"]
|
||||
|
||||
|
||||
def test_cross_domain_redirect_with_auth():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
url = "https://example.com/cross_domain"
|
||||
response = client.get(url, auth=("user", "pass"), follow_redirects=True)
|
||||
assert response.url == "https://example.org/cross_domain_target"
|
||||
assert "authorization" not in response.json()["headers"]
|
||||
@pytest.mark.anyio
|
||||
async def test_same_domain_redirect():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
url = "https://example.org/cross_domain"
|
||||
headers = {"Authorization": "abc"}
|
||||
response = await client.get(url, headers=headers, follow_redirects=True)
|
||||
assert response.url == "https://example.org/cross_domain_target"
|
||||
assert response.json()["headers"]["authorization"] == "abc"
|
||||
|
||||
|
||||
def test_same_domain_redirect():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
url = "https://example.org/cross_domain"
|
||||
headers = {"Authorization": "abc"}
|
||||
response = client.get(url, headers=headers, follow_redirects=True)
|
||||
assert response.url == "https://example.org/cross_domain_target"
|
||||
assert response.json()["headers"]["authorization"] == "abc"
|
||||
@pytest.mark.anyio
|
||||
async def test_same_domain_https_redirect_with_auth_header():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
url = "http://example.org/cross_domain"
|
||||
headers = {"Authorization": "abc"}
|
||||
response = await client.get(url, headers=headers, follow_redirects=True)
|
||||
assert response.url == "https://example.org/cross_domain_target"
|
||||
assert response.json()["headers"]["authorization"] == "abc"
|
||||
|
||||
|
||||
def test_same_domain_https_redirect_with_auth_header():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
url = "http://example.org/cross_domain"
|
||||
headers = {"Authorization": "abc"}
|
||||
response = client.get(url, headers=headers, follow_redirects=True)
|
||||
assert response.url == "https://example.org/cross_domain_target"
|
||||
assert response.json()["headers"]["authorization"] == "abc"
|
||||
|
||||
|
||||
def test_body_redirect():
|
||||
@pytest.mark.anyio
|
||||
async def test_body_redirect():
|
||||
"""
|
||||
A 308 redirect should preserve the request body.
|
||||
"""
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
url = "https://example.org/redirect_body"
|
||||
content = b"Example request body"
|
||||
response = client.post(url, content=content, follow_redirects=True)
|
||||
assert response.url == "https://example.org/redirect_body_target"
|
||||
assert response.json()["body"] == "Example request body"
|
||||
assert "content-length" in response.json()["headers"]
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
url = "https://example.org/redirect_body"
|
||||
content = b"Example request body"
|
||||
response = await client.post(url, content=content, follow_redirects=True)
|
||||
assert response.url == "https://example.org/redirect_body_target"
|
||||
assert response.json()["body"] == "Example request body"
|
||||
assert "content-length" in response.json()["headers"]
|
||||
|
||||
|
||||
def test_no_body_redirect():
|
||||
@pytest.mark.anyio
|
||||
async def test_no_body_redirect():
|
||||
"""
|
||||
A 303 redirect should remove the request body.
|
||||
"""
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
url = "https://example.org/redirect_no_body"
|
||||
content = b"Example request body"
|
||||
response = client.post(url, content=content, follow_redirects=True)
|
||||
assert response.url == "https://example.org/redirect_body_target"
|
||||
assert response.json()["body"] == ""
|
||||
assert "content-length" not in response.json()["headers"]
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
url = "https://example.org/redirect_no_body"
|
||||
content = b"Example request body"
|
||||
response = await client.post(url, content=content, follow_redirects=True)
|
||||
assert response.url == "https://example.org/redirect_body_target"
|
||||
assert response.json()["body"] == ""
|
||||
assert "content-length" not in response.json()["headers"]
|
||||
|
||||
|
||||
def test_can_stream_if_no_redirect():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
url = "https://example.org/redirect_301"
|
||||
with client.stream("GET", url, follow_redirects=False) as response:
|
||||
pass
|
||||
assert response.status_code == httpx.codes.MOVED_PERMANENTLY
|
||||
assert response.headers["location"] == "https://example.org/"
|
||||
@pytest.mark.anyio
|
||||
async def test_can_stream_if_no_redirect():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
url = "https://example.org/redirect_301"
|
||||
async with client.stream("GET", url, follow_redirects=False) as response:
|
||||
pass
|
||||
assert response.status_code == httpx.codes.MOVED_PERMANENTLY
|
||||
assert response.headers["location"] == "https://example.org/"
|
||||
|
||||
|
||||
class ConsumeBodyTransport(httpx.MockTransport):
|
||||
def handle_request(self, request: httpx.Request) -> httpx.Response:
|
||||
assert isinstance(request.stream, httpx.SyncByteStream)
|
||||
list(request.stream)
|
||||
async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
|
||||
assert isinstance(request.stream, httpx.AsyncByteStream)
|
||||
async for _ in request.stream:
|
||||
pass
|
||||
return self.handler(request) # type: ignore[return-value]
|
||||
|
||||
|
||||
def test_cannot_redirect_streaming_body():
|
||||
client = httpx.Client(transport=ConsumeBodyTransport(redirects))
|
||||
url = "https://example.org/redirect_body"
|
||||
@pytest.mark.anyio
|
||||
async def test_cannot_redirect_streaming_body():
|
||||
async with httpx.AsyncClient(transport=ConsumeBodyTransport(redirects)) as client:
|
||||
url = "https://example.org/redirect_body"
|
||||
|
||||
def streaming_body() -> typing.Iterator[bytes]:
|
||||
yield b"Example request body" # pragma: no cover
|
||||
async def streaming_body() -> typing.AsyncIterator[bytes]:
|
||||
yield b"Example request body" # pragma: no cover
|
||||
|
||||
with pytest.raises(httpx.StreamConsumed):
|
||||
client.post(url, content=streaming_body(), follow_redirects=True)
|
||||
with pytest.raises(httpx.StreamConsumed):
|
||||
await client.post(url, content=streaming_body(), follow_redirects=True)
|
||||
|
||||
|
||||
def test_cross_subdomain_redirect():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
url = "https://example.com/cross_subdomain"
|
||||
response = client.get(url, follow_redirects=True)
|
||||
assert response.url == "https://www.example.org/cross_subdomain"
|
||||
@pytest.mark.anyio
|
||||
async def test_cross_subdomain_redirect():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
url = "https://example.com/cross_subdomain"
|
||||
response = await client.get(url, follow_redirects=True)
|
||||
assert response.url == "https://www.example.org/cross_subdomain"
|
||||
|
||||
|
||||
def cookie_sessions(request: httpx.Request) -> httpx.Response:
|
||||
@pytest.mark.anyio
|
||||
async def cookie_sessions(request: httpx.Request) -> httpx.Response:
|
||||
if request.url.path == "/":
|
||||
cookie = request.headers.get("Cookie")
|
||||
if cookie is not None:
|
||||
@ -400,46 +406,49 @@ def cookie_sessions(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(status_code, headers=headers)
|
||||
|
||||
|
||||
def test_redirect_cookie_behavior():
|
||||
client = httpx.Client(
|
||||
@pytest.mark.anyio
|
||||
async def test_redirect_cookie_behavior():
|
||||
async with httpx.AsyncClient(
|
||||
transport=httpx.MockTransport(cookie_sessions), follow_redirects=True
|
||||
)
|
||||
) as client:
|
||||
# The client is not logged in.
|
||||
response = await client.get("https://example.com/")
|
||||
assert response.url == "https://example.com/"
|
||||
assert response.text == "Not logged in"
|
||||
|
||||
# The client is not logged in.
|
||||
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")
|
||||
assert response.url == "https://example.com/"
|
||||
assert response.text == "Logged in"
|
||||
|
||||
# Login redirects to the homepage, setting a session cookie.
|
||||
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/")
|
||||
assert response.url == "https://example.com/"
|
||||
assert response.text == "Logged in"
|
||||
|
||||
# The client is logged in.
|
||||
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")
|
||||
assert response.url == "https://example.com/"
|
||||
assert response.text == "Not logged in"
|
||||
|
||||
# Logout redirects to the homepage, expiring the session cookie.
|
||||
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 = client.get("https://example.com/")
|
||||
assert response.url == "https://example.com/"
|
||||
assert response.text == "Not logged in"
|
||||
|
||||
|
||||
def test_redirect_custom_scheme():
|
||||
client = httpx.Client(transport=httpx.MockTransport(redirects))
|
||||
with pytest.raises(httpx.UnsupportedProtocol) as e:
|
||||
client.post("https://example.org/redirect_custom_scheme", follow_redirects=True)
|
||||
assert str(e.value) == "Scheme 'market' not supported."
|
||||
# The client is not logged in.
|
||||
response = await client.get("https://example.com/")
|
||||
assert response.url == "https://example.com/"
|
||||
assert response.text == "Not logged in"
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_async_invalid_redirect():
|
||||
async def test_redirect_custom_scheme():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
with pytest.raises(httpx.UnsupportedProtocol) as e:
|
||||
await client.post(
|
||||
"https://example.org/redirect_custom_scheme", follow_redirects=True
|
||||
)
|
||||
assert str(e.value) == "Scheme 'market' not supported."
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_invalid_redirect():
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
|
||||
with pytest.raises(httpx.RemoteProtocolError):
|
||||
await client.get(
|
||||
|
||||
Loading…
Reference in New Issue
Block a user