self.auto_error = auto_error
async def __call__(self, request: Request) -> Optional[str]:
- api_key: str = request.headers.get(self.model.name)
+ api_key = request.headers.get(self.model.name)
if not api_key:
if self.auto_error:
raise HTTPException(
async def __call__(
self, request: Request
) -> Optional[HTTPAuthorizationCredentials]:
- authorization: str = request.headers.get("Authorization")
+ authorization = request.headers.get("Authorization")
scheme, credentials = get_authorization_scheme_param(authorization)
if not (authorization and scheme and credentials):
if self.auto_error:
async def __call__( # type: ignore
self, request: Request
) -> Optional[HTTPBasicCredentials]:
- authorization: str = request.headers.get("Authorization")
+ authorization = request.headers.get("Authorization")
scheme, param = get_authorization_scheme_param(authorization)
if self.realm:
unauthorized_headers = {"WWW-Authenticate": f'Basic realm="{self.realm}"'}
async def __call__(
self, request: Request
) -> Optional[HTTPAuthorizationCredentials]:
- authorization: str = request.headers.get("Authorization")
+ authorization = request.headers.get("Authorization")
scheme, credentials = get_authorization_scheme_param(authorization)
if not (authorization and scheme and credentials):
if self.auto_error:
async def __call__(
self, request: Request
) -> Optional[HTTPAuthorizationCredentials]:
- authorization: str = request.headers.get("Authorization")
+ authorization = request.headers.get("Authorization")
scheme, credentials = get_authorization_scheme_param(authorization)
if not (authorization and scheme and credentials):
if self.auto_error:
self.auto_error = auto_error
async def __call__(self, request: Request) -> Optional[str]:
- authorization: str = request.headers.get("Authorization")
+ authorization = request.headers.get("Authorization")
if not authorization:
if self.auto_error:
raise HTTPException(
)
async def __call__(self, request: Request) -> Optional[str]:
- authorization: str = request.headers.get("Authorization")
+ authorization = request.headers.get("Authorization")
scheme, param = get_authorization_scheme_param(authorization)
if not authorization or scheme.lower() != "bearer":
if self.auto_error:
)
async def __call__(self, request: Request) -> Optional[str]:
- authorization: str = request.headers.get("Authorization")
+ authorization = request.headers.get("Authorization")
scheme, param = get_authorization_scheme_param(authorization)
if not authorization or scheme.lower() != "bearer":
if self.auto_error:
self.auto_error = auto_error
async def __call__(self, request: Request) -> Optional[str]:
- authorization: str = request.headers.get("Authorization")
+ authorization = request.headers.get("Authorization")
if not authorization:
if self.auto_error:
raise HTTPException(
-from typing import Tuple
+from typing import Optional, Tuple
-def get_authorization_scheme_param(authorization_header_value: str) -> Tuple[str, str]:
+def get_authorization_scheme_param(
+ authorization_header_value: Optional[str],
+) -> Tuple[str, str]:
if not authorization_header_value:
return "", ""
scheme, _, param = authorization_header_value.partition(" ")
"Topic :: Internet :: WWW/HTTP",
]
dependencies = [
- "starlette==0.20.4",
+ "starlette==0.21.0",
"pydantic >=1.6.2,!=1.7,!=1.7.1,!=1.7.2,!=1.7.3,!=1.8,!=1.8.1,<2.0.0",
]
dynamic = ["version"]
def test_get_invalid():
- response = client.get("/foo", params={"client_id": None})
+ response = client.get("/foo")
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
def test_delete():
- response = client.delete("/items/foo", json={"name": "Foo"})
+ response = client.request("DELETE", "/items/foo", json={"name": "Foo"})
assert response.status_code == 200, response.text
assert response.json() == {"item_id": "foo", "item": {"name": "Foo", "price": None}}
def test_get_with_body():
body = {"name": "Foo", "description": "Some description", "price": 5.5}
- response = client.get("/product", json=body)
+ response = client.request("GET", "/product", json=body)
assert response.json() == body
return {"hidden_query": hidden_query}
-client = TestClient(app)
-
openapi_shema = {
"openapi": "3.0.2",
"info": {"title": "FastAPI", "version": "0.1.0"},
def test_openapi_schema():
+ client = TestClient(app)
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_shema
],
)
def test_hidden_cookie(path, cookies, expected_status, expected_response):
- response = client.get(path, cookies=cookies)
+ client = TestClient(app, cookies=cookies)
+ response = client.get(path)
assert response.status_code == expected_status
assert response.json() == expected_response
],
)
def test_hidden_header(path, headers, expected_status, expected_response):
+ client = TestClient(app)
response = client.get(path, headers=headers)
assert response.status_code == expected_status
assert response.json() == expected_response
def test_hidden_path():
+ client = TestClient(app)
response = client.get("/hidden_path/hidden_path")
assert response.status_code == 200
assert response.json() == {"hidden_path": "hidden_path"}
],
)
def test_hidden_query(path, expected_status, expected_response):
+ client = TestClient(app)
response = client.get(path)
assert response.status_code == expected_status
assert response.json() == expected_response
return current_user
-client = TestClient(app)
-
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "FastAPI", "version": "0.1.0"},
def test_openapi_schema():
+ client = TestClient(app)
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == openapi_schema
def test_security_api_key():
- response = client.get("/users/me", cookies={"key": "secret"})
+ client = TestClient(app, cookies={"key": "secret"})
+ response = client.get("/users/me")
assert response.status_code == 200, response.text
assert response.json() == {"username": "secret"}
def test_security_api_key_no_key():
+ client = TestClient(app)
response = client.get("/users/me")
assert response.status_code == 403, response.text
assert response.json() == {"detail": "Not authenticated"}
return current_user
-client = TestClient(app)
-
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "FastAPI", "version": "0.1.0"},
def test_openapi_schema():
+ client = TestClient(app)
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == openapi_schema
def test_security_api_key():
- response = client.get("/users/me", cookies={"key": "secret"})
+ client = TestClient(app, cookies={"key": "secret"})
+ response = client.get("/users/me")
assert response.status_code == 200, response.text
assert response.json() == {"username": "secret"}
def test_security_api_key_no_key():
+ client = TestClient(app)
response = client.get("/users/me")
assert response.status_code == 403, response.text
assert response.json() == {"detail": "Not authenticated"}
return current_user
-client = TestClient(app)
-
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "FastAPI", "version": "0.1.0"},
def test_openapi_schema():
+ client = TestClient(app)
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == openapi_schema
def test_security_api_key():
- response = client.get("/users/me", cookies={"key": "secret"})
+ client = TestClient(app, cookies={"key": "secret"})
+ response = client.get("/users/me")
assert response.status_code == 200, response.text
assert response.json() == {"username": "secret"}
def test_security_api_key_no_key():
+ client = TestClient(app)
response = client.get("/users/me")
assert response.status_code == 200, response.text
assert response.json() == {"msg": "Create an account first"}
def test_tuple_form_valid():
- response = client.post("/tuple-form/", data=[("values", "1"), ("values", "2")])
+ response = client.post("/tuple-form/", data={"values": ("1", "2")})
assert response.status_code == 200, response.text
assert response.json() == [1, 2]
def test_tuple_form_invalid():
- response = client.post(
- "/tuple-form/", data=[("values", "1"), ("values", "2"), ("values", "3")]
- )
+ response = client.post("/tuple-form/", data={"values": ("1", "2", "3")})
assert response.status_code == 422, response.text
- response = client.post("/tuple-form/", data=[("values", "1")])
+ response = client.post("/tuple-form/", data={"values": ("1")})
assert response.status_code == 422, response.text
assert response.status_code == 200, response.text
client = TestClient(app)
- response = client.get("/", allow_redirects=False)
+ response = client.get("/", follow_redirects=False)
assert response.status_code == 307, response.text
assert response.headers["location"] == "https://testserver/"
response = client.post(
"/items/",
headers={"content-type": "application/json"},
- data="{some broken json}",
+ content="{some broken json}",
)
assert response.status_code == 422, response.text
assert response.json() == {
def test_explicit_content_type():
response = client.post(
"/items/",
- data='{"name": "Foo", "price": 50.5}',
+ content='{"name": "Foo", "price": 50.5}',
headers={"Content-Type": "application/json"},
)
assert response.status_code == 200, response.text
def test_geo_json():
response = client.post(
"/items/",
- data='{"name": "Foo", "price": 50.5}',
+ content='{"name": "Foo", "price": 50.5}',
headers={"Content-Type": "application/geo+json"},
)
assert response.status_code == 200, response.text
def test_no_content_type_is_json():
response = client.post(
"/items/",
- data='{"name": "Foo", "price": 50.5}',
+ content='{"name": "Foo", "price": 50.5}',
)
assert response.status_code == 200, response.text
assert response.json() == {
]
}
- response = client.post("/items/", data=data, headers={"Content-Type": "text/plain"})
+ response = client.post(
+ "/items/", content=data, headers={"Content-Type": "text/plain"}
+ )
assert response.status_code == 422, response.text
assert response.json() == invalid_dict
response = client.post(
- "/items/", data=data, headers={"Content-Type": "application/geo+json-seq"}
+ "/items/", content=data, headers={"Content-Type": "application/geo+json-seq"}
)
assert response.status_code == 422, response.text
assert response.json() == invalid_dict
response = client.post(
- "/items/", data=data, headers={"Content-Type": "application/not-really-json"}
+ "/items/", content=data, headers={"Content-Type": "application/not-really-json"}
)
assert response.status_code == 422, response.text
assert response.json() == invalid_dict
response = client.post(
"/items/",
headers={"content-type": "application/json"},
- data="{some broken json}",
+ content="{some broken json}",
)
assert response.status_code == 422, response.text
assert response.json() == {
def test_explicit_content_type(client: TestClient):
response = client.post(
"/items/",
- data='{"name": "Foo", "price": 50.5}',
+ content='{"name": "Foo", "price": 50.5}',
headers={"Content-Type": "application/json"},
)
assert response.status_code == 200, response.text
def test_geo_json(client: TestClient):
response = client.post(
"/items/",
- data='{"name": "Foo", "price": 50.5}',
+ content='{"name": "Foo", "price": 50.5}',
headers={"Content-Type": "application/geo+json"},
)
assert response.status_code == 200, response.text
def test_no_content_type_is_json(client: TestClient):
response = client.post(
"/items/",
- data='{"name": "Foo", "price": 50.5}',
+ content='{"name": "Foo", "price": 50.5}',
)
assert response.status_code == 200, response.text
assert response.json() == {
]
}
- response = client.post("/items/", data=data, headers={"Content-Type": "text/plain"})
+ response = client.post(
+ "/items/", content=data, headers={"Content-Type": "text/plain"}
+ )
assert response.status_code == 422, response.text
assert response.json() == invalid_dict
response = client.post(
- "/items/", data=data, headers={"Content-Type": "application/geo+json-seq"}
+ "/items/", content=data, headers={"Content-Type": "application/geo+json-seq"}
)
assert response.status_code == 422, response.text
assert response.json() == invalid_dict
response = client.post(
- "/items/", data=data, headers={"Content-Type": "application/not-really-json"}
+ "/items/", content=data, headers={"Content-Type": "application/not-really-json"}
)
assert response.status_code == 422, response.text
assert response.json() == invalid_dict
from docs_src.cookie_params.tutorial001 import app
-client = TestClient(app)
-
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "FastAPI", "version": "0.1.0"},
],
)
def test(path, cookies, expected_status, expected_response):
- response = client.get(path, cookies=cookies)
+ client = TestClient(app, cookies=cookies)
+ response = client.get(path)
assert response.status_code == expected_status
assert response.json() == expected_response
}
-@pytest.fixture(name="client")
-def get_client():
- from docs_src.cookie_params.tutorial001_py310 import app
-
- client = TestClient(app)
- return client
-
-
@needs_py310
@pytest.mark.parametrize(
"path,cookies,expected_status,expected_response",
("/items", {"session": "cookiesession"}, 200, {"ads_id": None}),
],
)
-def test(path, cookies, expected_status, expected_response, client: TestClient):
- response = client.get(path, cookies=cookies)
+def test(path, cookies, expected_status, expected_response):
+ from docs_src.cookie_params.tutorial001_py310 import app
+
+ client = TestClient(app, cookies=cookies)
+ response = client.get(path)
assert response.status_code == expected_status
assert response.json() == expected_response
data = gzip.compress(data)
headers["Content-Encoding"] = "gzip"
headers["Content-Type"] = "application/json"
- response = client.post("/sum", data=data, headers=headers)
+ response = client.post("/sum", content=data, headers=headers)
assert response.json() == {"sum": n}
def test_get():
- response = client.get("/typer", allow_redirects=False)
+ response = client.get("/typer", follow_redirects=False)
assert response.status_code == 307, response.text
assert response.headers["location"] == "https://typer.tiangolo.com"
def test_redirect_response_class():
- response = client.get("/fastapi", allow_redirects=False)
+ response = client.get("/fastapi", follow_redirects=False)
assert response.status_code == 307
assert response.headers["location"] == "https://fastapi.tiangolo.com"
def test_redirect_status_code():
- response = client.get("/pydantic", allow_redirects=False)
+ response = client.get("/pydantic", follow_redirects=False)
assert response.status_code == 302
assert response.headers["location"] == "https://pydantic-docs.helpmanual.io/"
def test_post():
- response = client.post("/items/", data=b"this is actually not validated")
+ response = client.post("/items/", content=b"this is actually not validated")
assert response.status_code == 200, response.text
assert response.json() == {
"size": 30,
- x-men
- x-avengers
"""
- response = client.post("/items/", data=yaml_data)
+ response = client.post("/items/", content=yaml_data)
assert response.status_code == 200, response.text
assert response.json() == {
"name": "Deadpoolio",
x - x-men
x - x-avengers
"""
- response = client.post("/items/", data=yaml_data)
+ response = client.post("/items/", content=yaml_data)
assert response.status_code == 422, response.text
assert response.json() == {"detail": "Invalid YAML"}
- x-avengers
- sneaky: object
"""
- response = client.post("/items/", data=yaml_data)
+ response = client.post("/items/", content=yaml_data)
assert response.status_code == 422, response.text
assert response.json() == {
"detail": [
from docs_src.websockets.tutorial002 import app
-client = TestClient(app)
-
def test_main():
+ client = TestClient(app)
response = client.get("/")
assert response.status_code == 200, response.text
assert b"<!DOCTYPE html>" in response.content
def test_websocket_with_cookie():
+ client = TestClient(app, cookies={"session": "fakesession"})
with pytest.raises(WebSocketDisconnect):
- with client.websocket_connect(
- "/items/foo/ws", cookies={"session": "fakesession"}
- ) as websocket:
+ with client.websocket_connect("/items/foo/ws") as websocket:
message = "Message one"
websocket.send_text(message)
data = websocket.receive_text()
def test_websocket_with_header():
+ client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/bar/ws?token=some-token") as websocket:
message = "Message one"
def test_websocket_with_header_and_query():
+ client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/2/ws?q=3&token=some-token") as websocket:
message = "Message one"
def test_websocket_no_credentials():
+ client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws"):
pytest.fail(
def test_websocket_invalid_data():
+ client = TestClient(app)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/items/foo/ws?q=bar&token=some-token"):
pytest.fail(