httpx/tests/client/test_redirects.py

import typing

import pytest

import httpx


def redirects(request: httpx.Request) -> httpx.Response:
    if request.url.scheme not in ("http", "https"):
        raise httpx.UnsupportedProtocol(f"Scheme {request.url.scheme!r} not supported.")

    if request.url.path == "/redirect_301":
        status_code = httpx.codes.MOVED_PERMANENTLY
        content = b"<a href='https://example.org/'>here</a>"
        headers = {"location": "https://example.org/"}
        return httpx.Response(status_code, headers=headers, content=content)

    elif request.url.path == "/redirect_302":
        status_code = httpx.codes.FOUND
        headers = {"location": "https://example.org/"}
        return httpx.Response(status_code, headers=headers)

    elif request.url.path == "/redirect_303":
        status_code = httpx.codes.SEE_OTHER
        headers = {"location": "https://example.org/"}
        return httpx.Response(status_code, headers=headers)

    elif request.url.path == "/relative_redirect":
        status_code = httpx.codes.SEE_OTHER
        headers = {"location": "/"}
        return httpx.Response(status_code, headers=headers)

    elif request.url.path == "/malformed_redirect":
        status_code = httpx.codes.SEE_OTHER
        headers = {"location": "https://:443/"}
        return httpx.Response(status_code, headers=headers)

    elif request.url.path == "/invalid_redirect":
        status_code = httpx.codes.SEE_OTHER
        raw_headers = [(b"location", "https://😇/".encode("utf-8"))]
        return httpx.Response(status_code, headers=raw_headers)

    elif request.url.path == "/no_scheme_redirect":
        status_code = httpx.codes.SEE_OTHER
        headers = {"location": "//example.org/"}
        return httpx.Response(status_code, headers=headers)

    elif request.url.path == "/multiple_redirects":
        params = httpx.QueryParams(request.url.query)
        count = int(params.get("count", "0"))
        redirect_count = count - 1
        status_code = httpx.codes.SEE_OTHER if count else httpx.codes.OK
        if count:
            location = "/multiple_redirects"
            if redirect_count:
                location += f"?count={redirect_count}"
            headers = {"location": location}
        else:
            headers = {}
        return httpx.Response(status_code, headers=headers)

    if request.url.path == "/redirect_loop":
        status_code = httpx.codes.SEE_OTHER
        headers = {"location": "/redirect_loop"}
        return httpx.Response(status_code, headers=headers)

    elif request.url.path == "/cross_domain":
        status_code = httpx.codes.SEE_OTHER
        headers = {"location": "https://example.org/cross_domain_target"}
        return httpx.Response(status_code, headers=headers)

    elif request.url.path == "/cross_domain_target":
        status_code = httpx.codes.OK
        data = {
            "body": request.content.decode("ascii"),
            "headers": dict(request.headers),
        }
        return httpx.Response(status_code, json=data)

    elif request.url.path == "/redirect_body":
        status_code = httpx.codes.PERMANENT_REDIRECT
        headers = {"location": "/redirect_body_target"}
        return httpx.Response(status_code, headers=headers)

    elif request.url.path == "/redirect_no_body":
        status_code = httpx.codes.SEE_OTHER
        headers = {"location": "/redirect_body_target"}
        return httpx.Response(status_code, headers=headers)

    elif request.url.path == "/redirect_body_target":
        data = {
            "body": request.content.decode("ascii"),
            "headers": dict(request.headers),
        }
        return httpx.Response(200, json=data)

    elif request.url.path == "/cross_subdomain":
        if request.headers["Host"] != "www.example.org":
            status_code = httpx.codes.PERMANENT_REDIRECT
            headers = {"location": "https://www.example.org/cross_subdomain"}
            return httpx.Response(status_code, headers=headers)
        else:
            return httpx.Response(200, text="Hello, world!")

    elif request.url.path == "/redirect_custom_scheme":
        status_code = httpx.codes.MOVED_PERMANENTLY
        headers = {"location": "market://details?id=42"}
        return httpx.Response(status_code, headers=headers)

    if request.method == "HEAD":
        return httpx.Response(200)

    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 response.request.method == "GET"
    assert len(response.history) == 1


def test_redirect_301_keep_method():
    client = httpx.Client(
        transport=httpx.MockTransport(redirects), keep_method_for_redirects=True
    )
    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 response.request.method == "POST"
    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 response.request.method == "GET"
    assert len(response.history) == 1


def test_redirect_302_keep_method():
    client = httpx.Client(
        transport=httpx.MockTransport(redirects), keep_method_for_redirects=True
    )
    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 response.request.method == "POST"
    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 response.request.method == "GET"
    assert len(response.history) == 1


def test_redirect_303_keep_method():
    client = httpx.Client(
        transport=httpx.MockTransport(redirects), keep_method_for_redirects=True
    )
    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 response.request.method == "GET"
    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_async_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)
        assert response.status_code == httpx.codes.SEE_OTHER
        assert response.url == "https://example.org/redirect_303"
        assert response.next_request is not None

        response = await 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


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


@pytest.mark.anyio
async def test_async_too_many_redirects():
    async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
        with pytest.raises(httpx.TooManyRedirects):
            await client.get(
                "https://example.org/multiple_redirects?count=21", follow_redirects=True
            )


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)


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"]


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"]


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"]


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"


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():
    """
    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"]


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"]


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/"


class ConsumeBodyTransport(httpx.MockTransport):
    def handle_request(self, request: httpx.Request) -> httpx.Response:
        assert isinstance(request.stream, httpx.SyncByteStream)
        list(request.stream)
        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"

    def streaming_body() -> typing.Iterator[bytes]:
        yield b"Example request body"  # pragma: no cover

    with pytest.raises(httpx.StreamConsumed):
        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"


def cookie_sessions(request: httpx.Request) -> httpx.Response:
    if request.url.path == "/":
        cookie = request.headers.get("Cookie")
        if cookie is not None:
            content = b"Logged in"
        else:
            content = b"Not logged in"
        return httpx.Response(200, content=content)

    elif request.url.path == "/login":
        status_code = httpx.codes.SEE_OTHER
        headers = {
            "location": "/",
            "set-cookie": (
                "session=eyJ1c2VybmFtZSI6ICJ0b21; path=/; Max-Age=1209600; "
                "httponly; samesite=lax"
            ),
        }
        return httpx.Response(status_code, headers=headers)

    else:
        assert request.url.path == "/logout"
        status_code = httpx.codes.SEE_OTHER
        headers = {
            "location": "/",
            "set-cookie": (
                "session=null; path=/; expires=Thu, 01 Jan 1970 00:00:00 GMT; "
                "httponly; samesite=lax"
            ),
        }
        return httpx.Response(status_code, headers=headers)


def test_redirect_cookie_behavior():
    client = httpx.Client(
        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"

    # 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 = 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"

    # 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."


@pytest.mark.anyio
async def test_async_invalid_redirect():
    async with httpx.AsyncClient(transport=httpx.MockTransport(redirects)) as client:
        with pytest.raises(httpx.RemoteProtocolError):
            await client.get(
                "http://example.org/invalid_redirect", follow_redirects=True
            )