'application/json'
```
-* `def __init__(self, headers)`
+* `def __init__(self, headers, encoding=None)`
+* `def copy()` - **Headers**
## `Cookies`
from datetime import timedelta
+import httpcore
import pytest
import httpx
+from httpx import ASGIDispatch
@pytest.mark.usefixtures("async_environment")
assert response.elapsed > timedelta(seconds=0)
+@pytest.mark.usefixtures("async_environment")
+async def test_get_invalid_url(server):
+ async with httpx.AsyncClient() as client:
+ with pytest.raises(httpx.InvalidURL):
+ await client.get("invalid://example.org")
+
+
@pytest.mark.usefixtures("async_environment")
async def test_build_request(server):
url = server.url.copy_with(path="/echo_headers")
assert response.status_code == 200
assert response.content == data
+
+
+def test_dispatch_deprecated():
+ dispatch = httpcore.AsyncHTTPTransport()
+
+ with pytest.warns(DeprecationWarning) as record:
+ client = httpx.AsyncClient(dispatch=dispatch)
+
+ assert client.transport is dispatch
+ assert len(record) == 1
+ assert record[0].message.args[0] == (
+ "The dispatch argument is deprecated since v0.13 and will be "
+ "removed in a future release, please use 'transport'"
+ )
+
+
+def test_asgi_dispatch_deprecated():
+ with pytest.warns(DeprecationWarning) as record:
+ ASGIDispatch(None)
+
+ assert len(record) == 1
+ assert (
+ record[0].message.args[0]
+ == "ASGIDispatch is deprecated, please use ASGITransport"
+ )
from datetime import timedelta
+import httpcore
import pytest
import httpx
+from httpx import WSGIDispatch
def test_get(server):
with httpx.Client() as client:
for status_code in (200, 400, 404, 500, 505):
response = client.request(
- "GET", server.url.copy_with(path="/status/{}".format(status_code))
+ "GET", server.url.copy_with(path=f"/status/{status_code}")
)
if 400 <= status_code < 600:
with pytest.raises(httpx.HTTPError) as exc_info:
response.raise_for_status()
assert exc_info.value.response == response
+ assert exc_info.value.request.url.path == f"/status/{status_code}"
else:
assert response.raise_for_status() is None # type: ignore
assert url.scheme == "https"
assert url.is_ssl
+
+
+def test_dispatch_deprecated():
+ dispatch = httpcore.SyncHTTPTransport()
+
+ with pytest.warns(DeprecationWarning) as record:
+ client = httpx.Client(dispatch=dispatch)
+
+ assert client.transport is dispatch
+ assert len(record) == 1
+ assert record[0].message.args[0] == (
+ "The dispatch argument is deprecated since v0.13 and will be "
+ "removed in a future release, please use 'transport'"
+ )
+
+
+def test_wsgi_dispatch_deprecated():
+ with pytest.warns(DeprecationWarning) as record:
+ WSGIDispatch(None)
+
+ assert len(record) == 1
+ assert (
+ record[0].message.args[0]
+ == "WSGIDispatch is deprecated, please use WSGITransport"
+ )
),
],
)
-def test_proxies_environ(monkeypatch, url, env, expected):
+@pytest.mark.parametrize("client_class", [httpx.Client, httpx.AsyncClient])
+def test_proxies_environ(monkeypatch, client_class, url, env, expected):
for name, value in env.items():
monkeypatch.setenv(name, value)
- client = httpx.AsyncClient()
+ client = client_class()
transport = client.transport_for_url(httpx.URL(url))
if expected is None:
assert scope["type"] == "http"
if scope["path"].startswith("/slow_response"):
await slow_response(scope, receive, send)
- elif scope["path"].startswith("/premature_close"):
- await premature_close(scope, receive, send)
elif scope["path"].startswith("/status"):
await status_code(scope, receive, send)
elif scope["path"].startswith("/echo_body"):
await send({"type": "http.response.body", "body": b"Hello, world!"})
-async def premature_close(scope, receive, send):
- await send(
- {
- "type": "http.response.start",
- "status": 200,
- "headers": [[b"content-type", b"text/plain"]],
- }
- )
-
-
async def status_code(scope, receive, send):
status_code = int(scope["path"].replace("/status/", ""))
await send(
assert h.raw == [(b"b", b"4")]
-def test_copy_headers():
+def test_copy_headers_method():
+ headers = httpx.Headers({"custom": "example"})
+ headers_copy = headers.copy()
+ assert headers == headers_copy
+ assert headers is not headers_copy
+
+
+def test_copy_headers_init():
headers = httpx.Headers({"custom": "example"})
headers_copy = httpx.Headers(headers)
assert headers == headers_copy
assert origin.port == 443
+def test_origin_repr():
+ origin = Origin("https://example.com:8080")
+ assert str(origin) == "Origin(scheme='https' host='example.com' port=8080)"
+
+
def test_url_copywith_for_authority():
copy_with_kwargs = {
"username": "username",