import os
import sys
import typing
-from threading import Lock
from urllib.request import parse_keqv_list
import pytest
+from threading import Lock
import httpx
yield request
+
def test_basic_auth() -> None:
url = "https://example.org/"
auth = ("user", "password123")
assert response.json() == {"auth": "Basic dXNlcjpwYXNzd29yZDEyMw=="}
+
def test_basic_auth_with_stream() -> None:
"""
See: https://github.com/encode/httpx/pull/1312
auth = ("user", "password123")
app = App()
- with httpx.Client(transport=httpx.MockTransport(app), auth=auth) as client:
+ with httpx.Client(
+ transport=httpx.MockTransport(app), auth=auth
+ ) as client:
with client.stream("GET", url) as response:
response.read()
assert response.json() == {"auth": "Basic dXNlcjpwYXNzd29yZDEyMw=="}
+
def test_basic_auth_in_url() -> None:
url = "https://user:password123@example.org/"
app = App()
assert response.json() == {"auth": "Basic dXNlcjpwYXNzd29yZDEyMw=="}
+
def test_basic_auth_on_session() -> None:
url = "https://example.org/"
auth = ("user", "password123")
app = App()
- with httpx.Client(transport=httpx.MockTransport(app), auth=auth) as client:
+ with httpx.Client(
+ transport=httpx.MockTransport(app), auth=auth
+ ) as client:
response = client.get(url)
assert response.status_code == 200
assert response.json() == {"auth": "Basic dXNlcjpwYXNzd29yZDEyMw=="}
+
def test_custom_auth() -> None:
url = "https://example.org/"
app = App()
assert response.json() == {"auth": "Token 123"}
+
def test_netrc_auth_credentials_exist() -> None:
"""
When netrc auth is being used and a request is made to a host that is
app = App()
auth = httpx.NetRCAuth(netrc_file)
- with httpx.Client(transport=httpx.MockTransport(app), auth=auth) as client:
+ with httpx.Client(
+ transport=httpx.MockTransport(app), auth=auth
+ ) as client:
response = client.get(url)
assert response.status_code == 200
}
+
def test_netrc_auth_credentials_do_not_exist() -> None:
"""
When netrc auth is being used and a request is made to a host that is
app = App()
auth = httpx.NetRCAuth(netrc_file)
- with httpx.Client(transport=httpx.MockTransport(app), auth=auth) as client:
+ with httpx.Client(
+ transport=httpx.MockTransport(app), auth=auth
+ ) as client:
response = client.get(url)
assert response.status_code == 200
sys.version_info >= (3, 11),
reason="netrc files without a password are valid from Python >= 3.11",
)
+
def test_netrc_auth_nopassword_parse_error() -> None: # pragma: no cover
"""
Python has different netrc parsing behaviours with different versions.
httpx.NetRCAuth(netrc_file)
+
def test_auth_disable_per_request() -> None:
url = "https://example.org/"
auth = ("user", "password123")
app = App()
- with httpx.Client(transport=httpx.MockTransport(app), auth=auth) as client:
+ with httpx.Client(
+ transport=httpx.MockTransport(app), auth=auth
+ ) as client:
response = client.get(url, auth=None)
assert response.status_code == 200
assert response.json() == {"auth": None}
+
def test_auth_hidden_url() -> None:
url = "http://example-username:example-password@example.org/"
expected = "URL('http://example-username:[secure]@example.org/')"
assert expected == repr(httpx.URL(url))
+
def test_auth_hidden_header() -> None:
url = "https://example.org/"
auth = ("example-username", "example-password")
assert "'authorization': '[secure]'" in str(response.request.headers)
+
def test_auth_property() -> None:
app = App()
assert response.json() == {"auth": "Basic dXNlcjpwYXNzd29yZDEyMw=="}
+
def test_auth_invalid_type() -> None:
app = App()
client.auth = "not a tuple, not a callable" # type: ignore
+
def test_digest_auth_returns_no_auth_if_no_digest_header_in_response() -> None:
url = "https://example.org/"
auth = httpx.DigestAuth(username="user", password="password123")
assert len(response.history) == 0
+
def test_digest_auth_returns_no_auth_if_alternate_auth_scheme() -> None:
url = "https://example.org/"
auth = httpx.DigestAuth(username="user", password="password123")
assert len(response.history) == 0
+
def test_digest_auth_200_response_including_digest_auth_header() -> None:
url = "https://example.org/"
auth = httpx.DigestAuth(username="user", password="password123")
assert len(response.history) == 0
+
def test_digest_auth_401_response_without_digest_auth_header() -> None:
url = "https://example.org/"
auth = httpx.DigestAuth(username="user", password="password123")
("SHA-512-SESS", 64, 128),
],
)
+
def test_digest_auth(
algorithm: str, expected_hash_length: int, expected_response_length: int
) -> None:
assert len(digest_data["cnonce"]) == 16 + 2
+
def test_digest_auth_no_specified_qop() -> None:
url = "https://example.org/"
auth = httpx.DigestAuth(username="user", password="password123")
@pytest.mark.parametrize("qop", ("auth, auth-int", "auth,auth-int", "unknown,auth"))
+
def test_digest_auth_qop_including_spaces_and_auth_returns_auth(qop: str) -> None:
url = "https://example.org/"
auth = httpx.DigestAuth(username="user", password="password123")
assert len(response.history) == 1
+
def test_digest_auth_qop_auth_int_not_implemented() -> None:
url = "https://example.org/"
auth = httpx.DigestAuth(username="user", password="password123")
client.get(url, auth=auth)
+
def test_digest_auth_qop_must_be_auth_or_auth_int() -> None:
url = "https://example.org/"
auth = httpx.DigestAuth(username="user", password="password123")
client.get(url, auth=auth)
+
def test_digest_auth_incorrect_credentials() -> None:
url = "https://example.org/"
auth = httpx.DigestAuth(username="user", password="password123")
assert len(response.history) == 1
+
def test_digest_auth_reuses_challenge() -> None:
url = "https://example.org/"
auth = httpx.DigestAuth(username="user", password="password123")
assert len(response_2.history) == 0
+
def test_digest_auth_resets_nonce_count_after_401() -> None:
url = "https://example.org/"
auth = httpx.DigestAuth(username="user", password="password123")
'Digest realm="httpx@example.org", qop="auth,au', # malformed fields list
],
)
+
def test_digest_auth_raises_protocol_error_on_malformed_header(
auth_header: str,
) -> None:
client.get(url, auth=auth)
+
def test_auth_history() -> None:
"""
Test that intermediate requests sent as part of an authentication flow
return self.handler(request) # type: ignore[return-value]
+
def test_digest_auth_unavailable_streaming_body():
url = "https://example.org/"
auth = httpx.DigestAuth(username="user", password="password123")
client.post(url, content=streaming_body(), auth=auth)
+
def test_auth_reads_response_body() -> None:
"""
Test that we can read the response body in an auth flow if `requires_response_body`
assert response.json() == {"auth": '{"auth":"xyz"}'}
+
def test_auth() -> None:
"""
Test that we can use an auth implementation specific to the async case, to
import httpx
+
def test_get(server):
url = server.url
with httpx.Client(http2=True) as client:
pytest.param("http://", id="no-host"),
],
)
+
def test_get_invalid_url(server, url):
with httpx.Client() as client:
with pytest.raises((httpx.UnsupportedProtocol, httpx.LocalProtocolError)):
client.get(url)
+
def test_build_request(server):
url = server.url.copy_with(path="/echo_headers")
headers = {"Custom-header": "value"}
assert response.json()["Custom-header"] == "value"
+
def test_post(server):
url = server.url
with httpx.Client() as client:
assert response.status_code == 200
+
def test_post_json(server):
url = server.url
with httpx.Client() as client:
assert response.status_code == 200
+
def test_stream_response(server):
with httpx.Client() as client:
with client.stream("GET", server.url) as response:
assert response.content == b"Hello, world!"
+
def test_access_content_stream_response(server):
with httpx.Client() as client:
with client.stream("GET", server.url) as response:
response.content # noqa: B018
+
def test_stream_request(server):
def hello_world() -> typing.Iterator[bytes]:
yield b"Hello, "
assert response.status_code == 200
+
def test_raise_for_status(server):
with httpx.Client() as client:
for status_code in (200, 400, 404, 500, 505):
assert response.raise_for_status() is response
+
def test_options(server):
with httpx.Client() as client:
response = client.options(server.url)
assert response.text == "Hello, world!"
+
def test_head(server):
with httpx.Client() as client:
response = client.head(server.url)
assert response.text == ""
+
def test_put(server):
with httpx.Client() as client:
response = client.put(server.url, content=b"Hello, world!")
assert response.status_code == 200
+
def test_patch(server):
with httpx.Client() as client:
response = client.patch(server.url, content=b"Hello, world!")
assert response.status_code == 200
+
def test_delete(server):
with httpx.Client() as client:
response = client.delete(server.url)
assert response.text == "Hello, world!"
+
def test_100_continue(server):
headers = {"Expect": "100-continue"}
content = b"Echo request body"
assert response.content == content
+
def test_context_managed_transport():
class Transport(httpx.BaseTransport):
def __init__(self) -> None:
]
+
def test_context_managed_transport_and_mount():
class Transport(httpx.BaseTransport):
def __init__(self, name: str) -> None:
transport = Transport(name="transport")
mounted = Transport(name="mounted")
- with httpx.Client(transport=transport, mounts={"http://www.example.org": mounted}):
+ with httpx.Client(
+ transport=transport, mounts={"http://www.example.org": mounted}
+ ):
pass
assert transport.events == [
return httpx.Response(200, text="Hello, world!")
+
def test_client_closed_state_using_implicit_open():
client = httpx.Client(transport=httpx.MockTransport(hello_world))
pass # pragma: no cover
+
def test_client_closed_state_using_with_block():
with httpx.Client(transport=httpx.MockTransport(hello_world)) as client:
assert not client.is_closed
return httpx.Response(200, json=data)
+
def test_mounted_transport():
transport = httpx.MockTransport(unmounted)
mounts = {"custom://": httpx.MockTransport(mounted)}
assert response.json() == {"app": "mounted"}
+
def test_mock_transport():
def hello_world(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, text="Hello, world!")
assert response.text == "Hello, world!"
+
def test_cancellation_during_stream():
"""
If any BaseException is raised during streaming the response, then the
assert stream_was_closed
+
def test_server_extensions(server):
url = server.url
with httpx.Client(http2=True) as client:
raise NotImplementedError() # pragma: no cover
+
def test_set_cookie() -> None:
"""
Send a request including a cookie.
assert response.json() == {"cookies": "example-name=example-value"}
+
def test_set_per_request_cookie_is_deprecated() -> None:
"""
Sending a request including a per-request cookie is deprecated.
url = "http://example.org/echo_cookies"
cookies = {"example-name": "example-value"}
- with httpx.Client(transport=httpx.MockTransport(get_and_set_cookies)) as client:
+ with httpx.Client(
+ transport=httpx.MockTransport(get_and_set_cookies)
+ ) as client:
with pytest.warns(DeprecationWarning):
response = client.get(url, cookies=cookies)
assert response.json() == {"cookies": "example-name=example-value"}
+
def test_set_cookie_with_cookiejar() -> None:
"""
Send a request including a cookie, using a `CookieJar` instance.
assert response.json() == {"cookies": "example-name=example-value"}
+
def test_setting_client_cookies_to_cookiejar() -> None:
"""
Send a request including a cookie, using a `CookieJar` instance.
assert response.json() == {"cookies": "example-name=example-value"}
+
def test_set_cookie_with_cookies_model() -> None:
"""
Send a request including a cookie, using a `Cookies` instance.
cookies = httpx.Cookies()
cookies["example-name"] = "example-value"
- with httpx.Client(transport=httpx.MockTransport(get_and_set_cookies)) as client:
+ with httpx.Client(
+ transport=httpx.MockTransport(get_and_set_cookies)
+ ) as client:
client.cookies = cookies
response = client.get(url)
assert response.json() == {"cookies": "example-name=example-value"}
+
def test_get_cookie() -> None:
url = "http://example.org/set_cookie"
- with httpx.Client(transport=httpx.MockTransport(get_and_set_cookies)) as client:
+ with httpx.Client(
+ transport=httpx.MockTransport(get_and_set_cookies)
+ ) as client:
response = client.get(url)
assert response.status_code == 200
assert client.cookies["example-name"] == "example-value"
+
def test_cookie_persistence() -> None:
"""
Ensure that Client instances persist cookies between requests.
"""
- with httpx.Client(transport=httpx.MockTransport(get_and_set_cookies)) as client:
+ with httpx.Client(
+ transport=httpx.MockTransport(get_and_set_cookies)
+ ) as client:
response = client.get("http://example.org/echo_cookies")
assert response.status_code == 200
assert response.json() == {"cookies": None}
+import pytest
+
import httpx
return httpx.Response(200, headers={"server": "testserver"})
+
def test_event_hooks():
events = []
]
+
def test_event_hooks_raising_exception():
def raise_on_4xx_5xx(response):
response.raise_for_status()
assert exc.response.is_closed
+
def test_event_hooks_with_redirect():
"""
A redirect request should trigger additional 'request' and 'response' event hooks.
return httpx.Response(200, json=data)
+
def test_client_header():
"""
Set a header in the Client.
}
+
def test_header_merge():
url = "http://example.org/echo_headers"
client_headers = {"User-Agent": "python-myclient/0.2.1"}
}
+
def test_header_merge_conflicting_headers():
url = "http://example.org/echo_headers"
client_headers = {"X-Auth-Token": "FooBar"}
}
+
def test_header_update():
url = "http://example.org/echo_headers"
with httpx.Client(transport=httpx.MockTransport(echo_headers)) as client:
}
+
def test_header_repeated_items():
url = "http://example.org/echo_headers"
with httpx.Client(
transport=httpx.MockTransport(echo_repeated_headers_items)
) as client:
- response = client.get(url, headers=[("x-header", "1"), ("x-header", "2,3")])
+ response = client.get(
+ url, headers=[("x-header", "1"), ("x-header", "2,3")]
+ )
assert response.status_code == 200
]
+
def test_header_repeated_multi_items():
url = "http://example.org/echo_headers"
with httpx.Client(
transport=httpx.MockTransport(echo_repeated_headers_multi_items)
) as client:
- response = client.get(url, headers=[("x-header", "1"), ("x-header", "2,3")])
+ response = client.get(
+ url, headers=[("x-header", "1"), ("x-header", "2,3")]
+ )
assert response.status_code == 200
assert ["x-header", "2,3"] in echoed_headers
+
def test_remove_default_header():
"""
Remove a default header from the Client.
}
+
def test_header_does_not_exist():
headers = httpx.Headers({"foo": "bar"})
with pytest.raises(KeyError):
del headers["baz"]
+
def test_header_with_incorrect_value():
with pytest.raises(
TypeError,
httpx.Headers({"foo": None}) # type: ignore
+
def test_host_with_auth_and_port_in_url():
"""
The Host header should only include the hostname, or hostname:port
}
+
def test_host_with_non_default_port_in_url():
"""
If the URL includes a non-default port, then it should be included in
}
+
def test_request_auto_headers():
request = httpx.Request("GET", "https://www.example.org/")
assert "host" in request.headers
+
def test_same_origin():
origin = httpx.URL("https://example.com")
request = httpx.Request("GET", "HTTPS://EXAMPLE.COM:443")
assert headers["Host"] == request.url.netloc.decode("ascii")
+
def test_not_same_origin():
origin = httpx.URL("https://example.com")
request = httpx.Request("GET", "HTTP://EXAMPLE.COM:80")
assert headers["Host"] == origin.netloc.decode("ascii")
+
def test_is_https_redirect():
url = httpx.URL("https://example.com")
request = httpx.Request(
assert "Authorization" in headers
+
def test_is_not_https_redirect():
url = httpx.URL("https://www.example.com")
request = httpx.Request(
assert "Authorization" not in headers
+
def test_is_not_https_redirect_if_not_default_ports():
url = httpx.URL("https://example.com:1337")
request = httpx.Request(
+import pytest
+
import httpx
+
def test_client_base_url():
with httpx.Client() as client:
client.base_url = "https://www.example.org/" # type: ignore
assert client.base_url == "https://www.example.org/"
+
def test_client_base_url_without_trailing_slash():
with httpx.Client() as client:
client.base_url = "https://www.example.org/path" # type: ignore
assert client.base_url == "https://www.example.org/path/"
+
def test_client_base_url_with_trailing_slash():
client = httpx.Client()
client.base_url = "https://www.example.org/path/" # type: ignore
assert client.base_url == "https://www.example.org/path/"
+
def test_client_headers():
with httpx.Client() as client:
client.headers = {"a": "b"} # type: ignore
assert client.headers["A"] == "b"
+
def test_client_cookies():
with httpx.Client() as client:
client.cookies = {"a": "b"} # type: ignore
assert mycookies[0].name == "a" and mycookies[0].value == "b"
+
def test_client_timeout():
expected_timeout = 12.0
with httpx.Client() as client:
assert client.timeout.pool == expected_timeout
+
def test_client_event_hooks():
def on_request(request):
pass # pragma: no cover
assert client.event_hooks == {"request": [on_request], "response": []}
+
def test_client_trust_env():
with httpx.Client() as client:
assert client.trust_env
return httpcore.URL(scheme=u.raw_scheme, host=u.raw_host, port=u.port, target="/")
+
def test_socks_proxy():
url = httpx.URL("http://www.example.com")
),
],
)
+
def test_transport_for_request(url, proxies, expected):
- mounts = {key: httpx.HTTPTransport(proxy=value) for key, value in proxies.items()}
+ mounts = {
+ key: httpx.HTTPTransport(proxy=value) for key, value in proxies.items()
+ }
with httpx.Client(mounts=mounts) as client:
transport = client._transport_for_url(httpx.URL(url))
assert transport._pool._proxy_url == url_to_origin(expected)
+
@pytest.mark.network
def test_proxy_close():
try:
client.close()
+
def test_unsupported_proxy_scheme():
with pytest.raises(ValueError):
httpx.Client(proxy="ftp://127.0.0.1")
),
],
)
+
def test_proxies_environ(monkeypatch, url, env, expected):
for name, value in env.items():
monkeypatch.setenv(name, value)
({"all://": "http://127.0.0.1"}, True),
],
)
+
def test_for_deprecated_proxy_params(proxies, is_valid):
- mounts = {key: httpx.HTTPTransport(proxy=value) for key, value in proxies.items()}
+ mounts = {
+ key: httpx.HTTPTransport(proxy=value) for key, value in proxies.items()
+ }
if not is_valid:
with pytest.raises(ValueError):
httpx.Client(mounts=mounts)
+
def test_proxy_with_mounts():
proxy_transport = httpx.HTTPTransport(proxy="http://127.0.0.1")
+import pytest
+
import httpx
return httpx.Response(200, text="Hello, world")
+
def test_client_queryparams():
client = httpx.Client(params={"a": "b"})
assert isinstance(client.params, httpx.QueryParams)
assert client.params["a"] == "b"
+
def test_client_queryparams_string():
with httpx.Client(params="a=b") as client:
assert isinstance(client.params, httpx.QueryParams)
assert client.params["a"] == "b"
+
def test_client_queryparams_echo():
url = "http://example.org/echo_queryparams"
client_queryparams = "first=str"
return httpx.Response(200, html="<html><body>Hello, world!</body></html>")
+
def test_redirect_301():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
response = client.post(
assert len(response.history) == 1
+
def test_redirect_302():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
response = client.post(
assert len(response.history) == 1
+
def test_redirect_303():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
- response = client.get("https://example.org/redirect_303", follow_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 len(response.history) == 1
+
def test_next_request():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
request = client.build_request("POST", "https://example.org/redirect_303")
assert response.next_request is None
+
def test_head_redirect():
"""
Contrary to Requests, redirects remain enabled by default for HEAD requests.
assert response.text == ""
+
def test_relative_redirect():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
response = client.get(
assert len(response.history) == 1
+
def test_malformed_redirect():
# https://github.com/encode/httpx/issues/771
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
assert len(response.history) == 1
+
def test_no_scheme_redirect():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
response = client.get(
assert len(response.history) == 1
+
def test_fragment_redirect():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
response = client.get(
assert len(response.history) == 1
+
def test_multiple_redirects():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
response = client.get(
assert len(response.history[1].history) == 1
+
def test_too_many_redirects():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
with pytest.raises(httpx.TooManyRedirects):
)
+
def test_redirect_loop():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
with pytest.raises(httpx.TooManyRedirects):
client.get("https://example.org/redirect_loop", follow_redirects=True)
+
def test_cross_domain_redirect_with_auth_header():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
url = "https://example.com/cross_domain"
assert "authorization" not in response.json()["headers"]
+
def test_cross_domain_https_redirect_with_auth_header():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
url = "http://example.com/cross_domain"
assert "authorization" not in response.json()["headers"]
+
def test_cross_domain_redirect_with_auth():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
url = "https://example.com/cross_domain"
assert "authorization" not in response.json()["headers"]
+
def test_same_domain_redirect():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
url = "https://example.org/cross_domain"
assert response.json()["headers"]["authorization"] == "abc"
+
def test_same_domain_https_redirect_with_auth_header():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
url = "http://example.org/cross_domain"
assert response.json()["headers"]["authorization"] == "abc"
+
def test_body_redirect():
"""
A 308 redirect should preserve the request body.
assert "content-length" in response.json()["headers"]
+
def test_no_body_redirect():
"""
A 303 redirect should remove the request body.
assert "content-length" not in response.json()["headers"]
+
def test_can_stream_if_no_redirect():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
url = "https://example.org/redirect_301"
return self.handler(request) # type: ignore[return-value]
+
def test_cannot_redirect_streaming_body():
with httpx.Client(transport=ConsumeBodyTransport(redirects)) as client:
url = "https://example.org/redirect_body"
client.post(url, content=streaming_body(), follow_redirects=True)
+
def test_cross_subdomain_redirect():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
url = "https://example.com/cross_subdomain"
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")
return httpx.Response(status_code, headers=headers)
+
def test_redirect_cookie_behavior():
with httpx.Client(
transport=httpx.MockTransport(cookie_sessions), follow_redirects=True
assert response.text == "Not logged in"
+
def test_redirect_custom_scheme():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
with pytest.raises(httpx.UnsupportedProtocol) as e:
assert str(e.value) == "Scheme 'market' not supported."
+
def test_invalid_redirect():
with httpx.Client(transport=httpx.MockTransport(redirects)) as client:
with pytest.raises(httpx.RemoteProtocolError):
- client.get("http://example.org/invalid_redirect", follow_redirects=True)
+ client.get(
+ "http://example.org/invalid_redirect", follow_redirects=True
+ )
projects / thirdparty / httpx.git / commitdiff
