319 lines
10 KiB
Python
Executable File
319 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
import pytest
|
|
|
|
import httpx
|
|
|
|
|
|
def echo_headers(request: httpx.Request) -> httpx.Response:
|
|
data = {"headers": dict(request.headers)}
|
|
return httpx.Response(200, json=data)
|
|
|
|
|
|
def echo_repeated_headers_multi_items(request: httpx.Request) -> httpx.Response:
|
|
data = {"headers": list(request.headers.multi_items())}
|
|
return httpx.Response(200, json=data)
|
|
|
|
|
|
def echo_repeated_headers_items(request: httpx.Request) -> httpx.Response:
|
|
data = {"headers": list(request.headers.items())}
|
|
return httpx.Response(200, json=data)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_client_header():
|
|
"""
|
|
Set a header in the Client.
|
|
"""
|
|
url = "http://example.org/echo_headers"
|
|
headers = {"Example-Header": "example-value"}
|
|
|
|
async with httpx.AsyncClient(
|
|
transport=httpx.MockTransport(echo_headers), headers=headers
|
|
) as client:
|
|
response = await client.get(url)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"headers": {
|
|
"accept": "*/*",
|
|
"accept-encoding": "gzip, deflate, br, zstd",
|
|
"connection": "keep-alive",
|
|
"example-header": "example-value",
|
|
"host": "example.org",
|
|
"user-agent": f"python-httpx/{httpx.__version__}",
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async 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"}
|
|
async with httpx.AsyncClient(
|
|
transport=httpx.MockTransport(echo_headers), headers=client_headers
|
|
) as client:
|
|
response = await client.get(url, headers=request_headers)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"headers": {
|
|
"accept": "*/*",
|
|
"accept-encoding": "gzip, deflate, br, zstd",
|
|
"connection": "keep-alive",
|
|
"host": "example.org",
|
|
"user-agent": "python-myclient/0.2.1",
|
|
"x-auth-token": "FooBarBazToken",
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_header_merge_conflicting_headers():
|
|
url = "http://example.org/echo_headers"
|
|
client_headers = {"X-Auth-Token": "FooBar"}
|
|
request_headers = {"X-Auth-Token": "BazToken"}
|
|
async with httpx.AsyncClient(
|
|
transport=httpx.MockTransport(echo_headers), headers=client_headers
|
|
) as client:
|
|
response = await client.get(url, headers=request_headers)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"headers": {
|
|
"accept": "*/*",
|
|
"accept-encoding": "gzip, deflate, br, zstd",
|
|
"connection": "keep-alive",
|
|
"host": "example.org",
|
|
"user-agent": f"python-httpx/{httpx.__version__}",
|
|
"x-auth-token": "BazToken",
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_header_update():
|
|
url = "http://example.org/echo_headers"
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(echo_headers)) as client:
|
|
first_response = await client.get(url)
|
|
client.headers.update(
|
|
{"User-Agent": "python-myclient/0.2.1", "Another-Header": "AThing"}
|
|
)
|
|
second_response = await client.get(url)
|
|
|
|
assert first_response.status_code == 200
|
|
assert first_response.json() == {
|
|
"headers": {
|
|
"accept": "*/*",
|
|
"accept-encoding": "gzip, deflate, br, zstd",
|
|
"connection": "keep-alive",
|
|
"host": "example.org",
|
|
"user-agent": f"python-httpx/{httpx.__version__}",
|
|
}
|
|
}
|
|
|
|
assert second_response.status_code == 200
|
|
assert second_response.json() == {
|
|
"headers": {
|
|
"accept": "*/*",
|
|
"accept-encoding": "gzip, deflate, br, zstd",
|
|
"another-header": "AThing",
|
|
"connection": "keep-alive",
|
|
"host": "example.org",
|
|
"user-agent": "python-myclient/0.2.1",
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_header_repeated_items():
|
|
url = "http://example.org/echo_headers"
|
|
async with httpx.AsyncClient(
|
|
transport=httpx.MockTransport(echo_repeated_headers_items)
|
|
) as client:
|
|
response = await client.get(
|
|
url, headers=[("x-header", "1"), ("x-header", "2,3")]
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
echoed_headers = response.json()["headers"]
|
|
# as per RFC 7230, the whitespace after a comma is insignificant
|
|
# so we split and strip here so that we can do a safe comparison
|
|
assert ["x-header", ["1", "2", "3"]] in [
|
|
[k, [subv.lstrip() for subv in v.split(",")]] for k, v in echoed_headers
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_header_repeated_multi_items():
|
|
url = "http://example.org/echo_headers"
|
|
async with httpx.AsyncClient(
|
|
transport=httpx.MockTransport(echo_repeated_headers_multi_items)
|
|
) as client:
|
|
response = await client.get(
|
|
url, headers=[("x-header", "1"), ("x-header", "2,3")]
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
echoed_headers = response.json()["headers"]
|
|
assert ["x-header", "1"] in echoed_headers
|
|
assert ["x-header", "2,3"] in echoed_headers
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_remove_default_header():
|
|
"""
|
|
Remove a default header from the Client.
|
|
"""
|
|
url = "http://example.org/echo_headers"
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(echo_headers)) as client:
|
|
del client.headers["User-Agent"]
|
|
|
|
response = await client.get(url)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"headers": {
|
|
"accept": "*/*",
|
|
"accept-encoding": "gzip, deflate, br, zstd",
|
|
"connection": "keep-alive",
|
|
"host": "example.org",
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_header_does_not_exist():
|
|
headers = httpx.Headers({"foo": "bar"})
|
|
with pytest.raises(KeyError):
|
|
del headers["baz"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_header_with_incorrect_value():
|
|
with pytest.raises(
|
|
TypeError,
|
|
match=f"Header value must be str or bytes, not {type(None)}",
|
|
):
|
|
httpx.Headers({"foo": None}) # type: ignore
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async 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
|
|
be present.
|
|
"""
|
|
url = "http://username:password@example.org:80/echo_headers"
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(echo_headers)) as client:
|
|
response = await client.get(url)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"headers": {
|
|
"accept": "*/*",
|
|
"accept-encoding": "gzip, deflate, br, zstd",
|
|
"connection": "keep-alive",
|
|
"host": "example.org",
|
|
"user-agent": f"python-httpx/{httpx.__version__}",
|
|
"authorization": "Basic dXNlcm5hbWU6cGFzc3dvcmQ=",
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async 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"
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(echo_headers)) as client:
|
|
response = await client.get(url)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"headers": {
|
|
"accept": "*/*",
|
|
"accept-encoding": "gzip, deflate, br, zstd",
|
|
"connection": "keep-alive",
|
|
"host": "example.org:123",
|
|
"user-agent": f"python-httpx/{httpx.__version__}",
|
|
"authorization": "Basic dXNlcm5hbWU6cGFzc3dvcmQ=",
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_request_auto_headers():
|
|
request = httpx.Request("GET", "https://www.example.org/")
|
|
assert "host" in request.headers
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_same_origin():
|
|
origin = httpx.URL("https://example.com")
|
|
request = httpx.Request("GET", "HTTPS://EXAMPLE.COM:443")
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
headers = client._redirect_headers(request, origin, "GET")
|
|
|
|
assert headers["Host"] == request.url.netloc.decode("ascii")
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_not_same_origin():
|
|
origin = httpx.URL("https://example.com")
|
|
request = httpx.Request("GET", "HTTP://EXAMPLE.COM:80")
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
headers = client._redirect_headers(request, origin, "GET")
|
|
|
|
assert headers["Host"] == origin.netloc.decode("ascii")
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_is_https_redirect():
|
|
url = httpx.URL("https://example.com")
|
|
request = httpx.Request(
|
|
"GET", "http://example.com", headers={"Authorization": "empty"}
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
headers = client._redirect_headers(request, url, "GET")
|
|
|
|
assert "Authorization" in headers
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_is_not_https_redirect():
|
|
url = httpx.URL("https://www.example.com")
|
|
request = httpx.Request(
|
|
"GET", "http://example.com", headers={"Authorization": "empty"}
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
headers = client._redirect_headers(request, url, "GET")
|
|
|
|
assert "Authorization" not in headers
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_is_not_https_redirect_if_not_default_ports():
|
|
url = httpx.URL("https://example.com:1337")
|
|
request = httpx.Request(
|
|
"GET", "http://example.com:9999", headers={"Authorization": "empty"}
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
headers = client._redirect_headers(request, url, "GET")
|
|
|
|
assert "Authorization" not in headers
|