From: Kar Petrosyan Date: Thu, 27 Feb 2025 15:19:45 +0000 (+0400) Subject: Make all the tests from test_redirects to be async X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=4965a61830a81525f2faffc27742798de1db4436;p=thirdparty%2Fhttpx.git Make all the tests from test_redirects to be async --- diff --git a/tests/client/test_redirects.py b/tests/client/test_redirects.py index f6582713..d607c17e 100644 --- a/tests/client/test_redirects.py +++ b/tests/client/test_redirects.py @@ -113,46 +113,41 @@ def redirects(request: httpx.Request) -> httpx.Response: return httpx.Response(200, html="Hello, world!") -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 - +@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 -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 +@pytest.mark.anyio +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 -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_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_async_next_request(): +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(): + 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_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 - 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.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_async_too_many_redirects(): +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 - ) - - -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_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_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_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_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_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_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_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_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_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_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" +@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_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 - ) - - # The client is not logged in. - response = client.get("https://example.com/") - assert response.url == "https://example.com/" - assert response.text == "Not logged in" + ) 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" - # 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" + # 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" - # The client is logged in. - response = client.get("https://example.com/") - 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" - # 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" + # 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" - # The client is not logged in. - response = client.get("https://example.com/") - 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/") + 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." +@pytest.mark.anyio +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_async_invalid_redirect(): +async def test_invalid_redirect(): async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client: with pytest.raises(httpx.RemoteProtocolError): await client.get(