Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_frw_validation_error_format(app: FastAPI):
@app.get("/wrong_path/{param}")
def route_for_test(param: int) -> None: # pragma: no cover
pass
client = Client(base_url="http://testserver", app=app)
response = await client.get("/wrong_path/asd")
assert response.status_code == HTTP_422_UNPROCESSABLE_ENTITY
error_data = response.json()
assert "errors" in error_data
async def test_soft_limit(server):
"""
The soft_limit config should limit the maximum number of keep-alive connections.
"""
pool_limits = httpx.PoolLimits(soft_limit=1)
async with ConnectionPool(pool_limits=pool_limits) as http:
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://localhost:8000/")
await response.read()
assert len(http.active_connections) == 0
assert len(http.keepalive_connections) == 1
def test_threaded_streaming_request():
url = "https://example.org/echo_request_body"
with Client(dispatch=MockDispatch()) as client:
response = client.post(url, data=streaming_body())
assert response.status_code == 200
assert response.text == "Hello, world!"
def test_threaded_request_body():
url = "https://example.org/echo_request_body"
with Client(dispatch=MockDispatch()) as client:
response = client.post(url, data=b"Hello, world!")
assert response.status_code == 200
assert response.text == "Hello, world!"
async def test_http2_live_request():
async with Client(http2=True) as client:
try:
resp = await client.get("https://nghttp2.org/httpbin/anything")
except TimeoutException:
pytest.xfail(reason="nghttp2.org appears to be unresponsive")
except socket.gaierror:
pytest.xfail(reason="You appear to be offline")
assert resp.status_code == 200
assert resp.http_version == "HTTP/2"
async def test_redirect_302():
client = Client(dispatch=MockDispatch())
response = await client.post("https://example.org/redirect_302")
assert response.status_code == codes.OK
assert response.url == URL("https://example.org/")
assert len(response.history) == 1
def test_client_cookies():
client = Client()
client.cookies = {"a": "b"}
assert isinstance(client.cookies, Cookies)
mycookies = list(client.cookies.jar)
assert len(mycookies) == 1
assert mycookies[0].name == "a" and mycookies[0].value == "b"
async def test_uds(uds_server):
url = uds_server.url
uds = uds_server.config.uds
assert uds is not None
async with httpx.Client(uds=uds) as client:
response = await client.get(url)
assert response.status_code == 200
assert response.text == "Hello, world!"
assert response.encoding == "iso-8859-1"
def get_new_session(self):
return httpx.Client()
def test_wsgi_exc():
client = httpx.Client(app=raise_exc)
with pytest.raises(ValueError):
client.get("http://www.example.org/")