async def test_get(server, backend):
- url = "http://127.0.0.1:8000/"
+ url = server.url
async with httpx.AsyncClient(backend=backend) as client:
response = await client.get(url)
assert response.status_code == 200
"""
Verify that the client is capable of making a simple request if not given a backend.
"""
- url = "http://127.0.0.1:8000/"
+ url = server.url
async with httpx.AsyncClient() as client:
response = await client.get(url)
assert response.status_code == 200
async def test_post(server, backend):
- url = "http://127.0.0.1:8000/"
+ url = server.url
async with httpx.AsyncClient(backend=backend) as client:
response = await client.post(url, data=b"Hello, world!")
assert response.status_code == 200
async def test_post_json(server, backend):
- url = "http://127.0.0.1:8000/"
+ url = server.url
async with httpx.AsyncClient(backend=backend) as client:
response = await client.post(url, json={"text": "Hello, world!"})
assert response.status_code == 200
async def test_stream_response(server, backend):
async with httpx.AsyncClient(backend=backend) as client:
- response = await client.request("GET", "http://127.0.0.1:8000/", stream=True)
+ response = await client.request("GET", server.url, stream=True)
assert response.status_code == 200
body = await response.read()
assert body == b"Hello, world!"
async def test_access_content_stream_response(server, backend):
async with httpx.AsyncClient(backend=backend) as client:
- response = await client.request("GET", "http://127.0.0.1:8000/", stream=True)
+ response = await client.request("GET", server.url, stream=True)
assert response.status_code == 200
with pytest.raises(httpx.ResponseNotRead):
response.content
yield b"world!"
async with httpx.AsyncClient(backend=backend) as client:
- response = await client.request(
- "POST", "http://127.0.0.1:8000/", data=hello_world()
- )
+ response = await client.request("POST", server.url, data=hello_world())
assert response.status_code == 200
async with httpx.AsyncClient(backend=backend) as client:
for status_code in (200, 400, 404, 500, 505):
response = await client.request(
- "GET", f"http://127.0.0.1:8000/status/{status_code}"
+ "GET", server.url.copy_with(path=f"/status/{status_code}")
)
if 400 <= status_code < 600:
async def test_options(server, backend):
- url = "http://127.0.0.1:8000/"
async with httpx.AsyncClient(backend=backend) as client:
- response = await client.options(url)
+ response = await client.options(server.url)
assert response.status_code == 200
assert response.text == "Hello, world!"
async def test_head(server, backend):
- url = "http://127.0.0.1:8000/"
async with httpx.AsyncClient(backend=backend) as client:
- response = await client.head(url)
+ response = await client.head(server.url)
assert response.status_code == 200
assert response.text == ""
async def test_put(server, backend):
- url = "http://127.0.0.1:8000/"
async with httpx.AsyncClient(backend=backend) as client:
- response = await client.put(url, data=b"Hello, world!")
+ response = await client.put(server.url, data=b"Hello, world!")
assert response.status_code == 200
async def test_patch(server, backend):
- url = "http://127.0.0.1:8000/"
async with httpx.AsyncClient(backend=backend) as client:
- response = await client.patch(url, data=b"Hello, world!")
+ response = await client.patch(server.url, data=b"Hello, world!")
assert response.status_code == 200
async def test_delete(server, backend):
- url = "http://127.0.0.1:8000/"
async with httpx.AsyncClient(backend=backend) as client:
- response = await client.delete(url)
+ response = await client.delete(server.url)
assert response.status_code == 200
assert response.text == "Hello, world!"
async def test_100_continue(server, backend):
- url = "http://127.0.0.1:8000/echo_body"
headers = {"Expect": "100-continue"}
data = b"Echo request body"
async with httpx.AsyncClient(backend=backend) as client:
- response = await client.post(url, headers=headers, data=data)
+ response = await client.post(
+ server.url.copy_with(path="/echo_body"), headers=headers, data=data
+ )
assert response.status_code == 200
assert response.content == data
def test_get(server):
- url = "http://127.0.0.1:8000/"
+ url = server.url
with httpx.Client() as http:
response = http.get(url)
assert response.status_code == 200
- assert response.url == httpx.URL(url)
+ assert response.url == url
assert response.content == b"Hello, world!"
assert response.text == "Hello, world!"
assert response.http_version == "HTTP/1.1"
assert response.encoding == "iso-8859-1"
- assert response.request.url == httpx.URL(url)
+ assert response.request.url == url
assert response.headers
assert response.is_redirect is False
assert repr(response) == "<Response [200 OK]>"
def test_post(server):
with httpx.Client() as http:
- response = http.post("http://127.0.0.1:8000/", data=b"Hello, world!")
+ response = http.post(server.url, data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
def test_post_json(server):
with httpx.Client() as http:
- response = http.post("http://127.0.0.1:8000/", json={"text": "Hello, world!"})
+ response = http.post(server.url, json={"text": "Hello, world!"})
assert response.status_code == 200
assert response.reason_phrase == "OK"
def test_stream_response(server):
with httpx.Client() as http:
- response = http.get("http://127.0.0.1:8000/", stream=True)
+ response = http.get(server.url, stream=True)
assert response.status_code == 200
content = response.read()
assert content == b"Hello, world!"
def test_stream_iterator(server):
with httpx.Client() as http:
- response = http.get("http://127.0.0.1:8000/", stream=True)
+ response = http.get(server.url, stream=True)
assert response.status_code == 200
body = b""
for chunk in response.stream():
def test_raw_iterator(server):
with httpx.Client() as http:
- response = http.get("http://127.0.0.1:8000/", stream=True)
+ response = http.get(server.url, stream=True)
assert response.status_code == 200
body = b""
for chunk in response.raw():
with httpx.Client() as client:
for status_code in (200, 400, 404, 500, 505):
response = client.request(
- "GET", "http://127.0.0.1:8000/status/{}".format(status_code)
+ "GET", server.url.copy_with(path="/status/{}".format(status_code))
)
if 400 <= status_code < 600:
with pytest.raises(httpx.exceptions.HTTPError) as exc_info:
def test_options(server):
with httpx.Client() as http:
- response = http.options("http://127.0.0.1:8000/")
+ response = http.options(server.url)
assert response.status_code == 200
assert response.reason_phrase == "OK"
def test_head(server):
with httpx.Client() as http:
- response = http.head("http://127.0.0.1:8000/")
+ response = http.head(server.url)
assert response.status_code == 200
assert response.reason_phrase == "OK"
def test_put(server):
with httpx.Client() as http:
- response = http.put("http://127.0.0.1:8000/", data=b"Hello, world!")
+ response = http.put(server.url, data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
def test_patch(server):
with httpx.Client() as http:
- response = http.patch("http://127.0.0.1:8000/", data=b"Hello, world!")
+ response = http.patch(server.url, data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
def test_delete(server):
with httpx.Client() as http:
- response = http.delete("http://127.0.0.1:8000/")
+ response = http.delete(server.url)
assert response.status_code == 200
assert response.reason_phrase == "OK"
def test_base_url(server):
- base_url = "http://127.0.0.1:8000/"
+ base_url = server.url
with httpx.Client(base_url=base_url) as http:
response = http.get("/")
assert response.status_code == 200
- assert str(response.url) == base_url
+ assert response.url == base_url
def test_merge_url():
from uvicorn.config import Config
from uvicorn.main import Server
-from httpx import AsyncioBackend
+from httpx import URL, AsyncioBackend
@pytest.fixture(params=[pytest.param(AsyncioBackend, marks=pytest.mark.asyncio)])
class TestServer(Server):
+ @property
+ def url(self) -> URL:
+ protocol = "https" if self.config.is_ssl else "http"
+ return URL(f"{protocol}://{self.config.host}:{self.config.port}/")
+
def install_signal_handlers(self) -> None:
# Disable the default installation of handlers for signals such as SIGTERM,
# because it can only be done in the main thread.
loop.create_task(super().serve(sockets=sockets)),
loop.create_task(self.watch_restarts()),
}
-
await asyncio.wait(tasks)
async def restart(self) -> None:
Connections should default to staying in a keep-alive state.
"""
async with httpx.ConnectionPool(backend=backend) as http:
- response = await http.request("GET", "http://127.0.0.1:8000/")
+ response = await http.request("GET", server.url)
await response.read()
assert len(http.active_connections) == 0
assert len(http.keepalive_connections) == 1
- response = await http.request("GET", "http://127.0.0.1:8000/")
+ response = await http.request("GET", server.url)
await response.read()
assert len(http.active_connections) == 0
assert len(http.keepalive_connections) == 1
Connections to differing connection keys should result in multiple connections.
"""
async with httpx.ConnectionPool(backend=backend) as http:
- response = await http.request("GET", "http://127.0.0.1:8000/")
+ response = await http.request("GET", server.url)
await response.read()
assert len(http.active_connections) == 0
assert len(http.keepalive_connections) == 1
pool_limits = httpx.PoolLimits(soft_limit=1)
async with httpx.ConnectionPool(pool_limits=pool_limits, backend=backend) as http:
- response = await http.request("GET", "http://127.0.0.1:8000/")
+ response = await http.request("GET", server.url)
await response.read()
assert len(http.active_connections) == 0
assert len(http.keepalive_connections) == 1
A streaming request should hold the connection open until the response is read.
"""
async with httpx.ConnectionPool(backend=backend) as http:
- response = await http.request("GET", "http://127.0.0.1:8000/")
+ response = await http.request("GET", server.url)
assert len(http.active_connections) == 1
assert len(http.keepalive_connections) == 0
Multiple conncurrent requests should open multiple conncurrent connections.
"""
async with httpx.ConnectionPool(backend=backend) as http:
- response_a = await http.request("GET", "http://127.0.0.1:8000/")
+ response_a = await http.request("GET", server.url)
assert len(http.active_connections) == 1
assert len(http.keepalive_connections) == 0
- response_b = await http.request("GET", "http://127.0.0.1:8000/")
+ response_b = await http.request("GET", server.url)
assert len(http.active_connections) == 2
assert len(http.keepalive_connections) == 0
"""
headers = [(b"connection", b"close")]
async with httpx.ConnectionPool(backend=backend) as http:
- response = await http.request("GET", "http://127.0.0.1:8000/", headers=headers)
+ response = await http.request("GET", server.url, headers=headers)
await response.read()
assert len(http.active_connections) == 0
assert len(http.keepalive_connections) == 0
A standard close should keep the connection open.
"""
async with httpx.ConnectionPool(backend=backend) as http:
- response = await http.request("GET", "http://127.0.0.1:8000/")
+ response = await http.request("GET", server.url)
await response.read()
await response.close()
assert len(http.active_connections) == 0
A premature close should close the connection.
"""
async with httpx.ConnectionPool(backend=backend) as http:
- response = await http.request("GET", "http://127.0.0.1:8000/")
+ response = await http.request("GET", server.url)
await response.close()
assert len(http.active_connections) == 0
assert len(http.keepalive_connections) == 0
should be reestablished.
"""
async with httpx.ConnectionPool(backend=backend) as http:
- response = await http.request("GET", "http://127.0.0.1:8000/")
+ response = await http.request("GET", server.url)
await response.read()
# Shutdown the server to close the keep-alive connection
await restart(server)
- response = await http.request("GET", "http://127.0.0.1:8000/")
+ response = await http.request("GET", server.url)
await response.read()
assert len(http.active_connections) == 0
assert len(http.keepalive_connections) == 1
should be reestablished.
"""
async with httpx.ConnectionPool(backend=backend) as http:
- response = await http.request("GET", "http://127.0.0.1:8000/")
+ response = await http.request("GET", server.url)
await response.read()
# Shutdown the server to close the keep-alive connection
await restart(server)
- response = await http.request("GET", "http://127.0.0.1:8000/")
+ response = await http.request("GET", server.url)
await response.read()
assert len(http.active_connections) == 0
assert len(http.keepalive_connections) == 1
properly on a disconnected connection.
"""
async with httpx.ConnectionPool(pool_limits=httpx.PoolLimits(hard_limit=1)) as http:
- response = await http.request("GET", "http://127.0.0.1:8000/")
+ response = await http.request("GET", server.url)
await response.read()
# Close the connection so we're forced to recycle it
await restart(server)
- response = await http.request("GET", "http://127.0.0.1:8000/")
+ response = await http.request("GET", server.url)
assert response.status_code == 200
async def test_get(server, backend):
- async with HTTPConnection(origin="http://127.0.0.1:8000/", backend=backend) as conn:
- response = await conn.request("GET", "http://127.0.0.1:8000/")
+ async with HTTPConnection(origin=server.url, backend=backend) as conn:
+ response = await conn.request("GET", server.url)
await response.read()
assert response.status_code == 200
assert response.content == b"Hello, world!"
async def test_post(server, backend):
- async with HTTPConnection(origin="http://127.0.0.1:8000/", backend=backend) as conn:
- response = await conn.request(
- "GET", "http://127.0.0.1:8000/", data=b"Hello, world!"
- )
+ async with HTTPConnection(origin=server.url, backend=backend) as conn:
+ response = await conn.request("GET", server.url, data=b"Hello, world!")
assert response.status_code == 200
An HTTPS request, with default SSL configuration set on the client.
"""
async with HTTPConnection(
- origin="https://127.0.0.1:8001/", verify=False, backend=backend
+ origin=https_server.url, verify=False, backend=backend
) as conn:
- response = await conn.request("GET", "https://127.0.0.1:8001/")
+ response = await conn.request("GET", https_server.url)
await response.read()
assert response.status_code == 200
assert response.content == b"Hello, world!"
"""
An HTTPS request, with SSL configuration set on the request.
"""
- async with HTTPConnection(
- origin="https://127.0.0.1:8001/", backend=backend
- ) as conn:
- response = await conn.request("GET", "https://127.0.0.1:8001/", verify=False)
+ async with HTTPConnection(origin=https_server.url, backend=backend) as conn:
+ response = await conn.request("GET", https_server.url, verify=False)
await response.read()
assert response.status_code == 200
assert response.content == b"Hello, world!"
def test_get(server):
- response = httpx.get("http://127.0.0.1:8000/")
+ response = httpx.get(server.url)
assert response.status_code == 200
assert response.reason_phrase == "OK"
assert response.text == "Hello, world!"
def test_post(server):
- response = httpx.post("http://127.0.0.1:8000/", data=b"Hello, world!")
+ response = httpx.post(server.url, data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
yield b", "
yield b"world!"
- response = httpx.post("http://127.0.0.1:8000/", data=data())
+ response = httpx.post(server.url, data=data())
assert response.status_code == 200
assert response.reason_phrase == "OK"
def test_options(server):
- response = httpx.options("http://127.0.0.1:8000/")
+ response = httpx.options(server.url)
assert response.status_code == 200
assert response.reason_phrase == "OK"
def test_head(server):
- response = httpx.head("http://127.0.0.1:8000/")
+ response = httpx.head(server.url)
assert response.status_code == 200
assert response.reason_phrase == "OK"
def test_put(server):
- response = httpx.put("http://127.0.0.1:8000/", data=b"Hello, world!")
+ response = httpx.put(server.url, data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
def test_patch(server):
- response = httpx.patch("http://127.0.0.1:8000/", data=b"Hello, world!")
+ response = httpx.patch(server.url, data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
def test_delete(server):
- response = httpx.delete("http://127.0.0.1:8000/")
+ response = httpx.delete(server.url)
assert response.status_code == 200
assert response.reason_phrase == "OK"
ctx = SSLConfig().load_ssl_context_no_verify(HTTPVersionConfig())
timeout = TimeoutConfig(5)
- stream = await backend.connect("127.0.0.1", 8001, None, timeout)
+ stream = await backend.connect(
+ https_server.url.host, https_server.url.port, None, timeout
+ )
try:
assert stream.is_connection_dropped() is False
assert stream.stream_writer.get_extra_info("cipher", default=None) is None
- stream = await backend.start_tls(stream, "127.0.0.1", ctx, timeout)
+ stream = await backend.start_tls(stream, https_server.url.host, ctx, timeout)
assert stream.is_connection_dropped() is False
assert stream.stream_writer.get_extra_info("cipher", default=None) is not None
async with AsyncClient(timeout=timeout, backend=backend) as client:
with pytest.raises(ReadTimeout):
- await client.get("http://127.0.0.1:8000/slow_response")
+ await client.get(server.url.copy_with(path="/slow_response"))
async def test_write_timeout(server, backend):
async with AsyncClient(timeout=timeout, backend=backend) as client:
with pytest.raises(WriteTimeout):
data = b"*" * 1024 * 1024 * 100
- await client.put("http://127.0.0.1:8000/slow_response", data=data)
+ await client.put(server.url.copy_with(path="/slow_response"), data=data)
async def test_connect_timeout(server, backend):
pool_limits = PoolLimits(hard_limit=1, pool_timeout=1e-6)
async with AsyncClient(pool_limits=pool_limits, backend=backend) as client:
- response = await client.get("http://127.0.0.1:8000/", stream=True)
+ response = await client.get(server.url, stream=True)
with pytest.raises(PoolTimeout):
await client.get("http://localhost:8000/")
utils.get_logger("httpx")
async with httpx.AsyncClient() as client:
- await client.get("http://127.0.0.1:8000/")
+ await client.get(server.url)
if httpx_debug in ("1", "True"):
assert "httpx.dispatch.connection_pool" in capsys.readouterr().err