mirror of
https://github.com/fastapi/fastapi.git
synced 2026-10-30 05:41:02 -04:00
448 lines
15 KiB
Python
448 lines
15 KiB
Python
from typing import Annotated
|
|
|
|
import pytest
|
|
from fastapi import Cookie, Depends, FastAPI, Header, Query
|
|
from fastapi.testclient import TestClient
|
|
from pydantic import AliasChoices, BaseModel, ConfigDict, Field
|
|
|
|
|
|
def test_optional_header_model():
|
|
app = FastAPI()
|
|
|
|
class RequestHeaders(BaseModel):
|
|
x_request_id: str
|
|
x_attempt: int
|
|
x_tag: list[str] = Field(default_factory=list)
|
|
enabled: bool = True
|
|
|
|
@app.get("/")
|
|
def read(headers: Annotated[RequestHeaders | None, Header()] = None):
|
|
return headers
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/")
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() is None
|
|
response = client.get(
|
|
"/",
|
|
headers=[
|
|
("X-Request-ID", ""),
|
|
("X-Attempt", "2"),
|
|
("X-Tag", "first"),
|
|
("x-tag", "second"),
|
|
],
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == {
|
|
"x_request_id": "",
|
|
"x_attempt": 2,
|
|
"x_tag": ["first", "second"],
|
|
"enabled": True,
|
|
}
|
|
response = client.get("/", headers={"X-Request-ID": ""})
|
|
assert response.status_code == 422
|
|
assert response.json()["detail"][0]["loc"] == ["header", "x-attempt"]
|
|
response = client.get("/", headers={"X-Request-ID": "request", "X-Attempt": "bad"})
|
|
assert response.status_code == 422
|
|
assert response.json()["detail"][0]["loc"] == ["header", "x-attempt"]
|
|
assert response.json()["detail"][0]["type"] == "int_parsing"
|
|
|
|
|
|
def test_header_conversion_is_per_optional_model():
|
|
app = FastAPI()
|
|
|
|
class RequestHeaders(BaseModel):
|
|
x_id: int
|
|
|
|
@app.get("/")
|
|
def read(
|
|
converted: Annotated[RequestHeaders | None, Header()] = None,
|
|
literal: Annotated[
|
|
RequestHeaders | None, Header(convert_underscores=False)
|
|
] = None,
|
|
):
|
|
return [converted, literal]
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/", headers={"X-ID": "1"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == [{"x_id": 1}, None]
|
|
response = client.get("/", headers={"X_ID": "2"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == [None, {"x_id": 2}]
|
|
response = client.get("/", headers={"X-ID": "1", "X_ID": "2"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == [{"x_id": 1}, {"x_id": 2}]
|
|
assert app.openapi()["paths"]["/"]["get"]["parameters"] == [
|
|
{
|
|
"name": "x-id",
|
|
"in": "header",
|
|
"required": False,
|
|
"schema": {"type": "integer", "title": "X Id"},
|
|
},
|
|
{
|
|
"name": "x_id",
|
|
"in": "header",
|
|
"required": False,
|
|
"schema": {"type": "integer", "title": "X Id"},
|
|
},
|
|
]
|
|
assert RequestHeaders.model_fields["x_id"].alias is None
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["x-request_id", "X-LEGACY_ID"])
|
|
def test_header_aliases_are_literal_and_case_insensitive(name):
|
|
app = FastAPI()
|
|
|
|
class RequestHeaders(BaseModel):
|
|
request_id: str = Field(
|
|
validation_alias=AliasChoices("X-Request_ID", "X-Legacy_ID")
|
|
)
|
|
x_attempt: int
|
|
|
|
@app.get("/")
|
|
def read(headers: Annotated[RequestHeaders | None, Header()] = None):
|
|
return headers
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/", headers={"X-Request-ID": "ignored"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() is None
|
|
response = client.get("/", headers={name: "request"})
|
|
assert response.status_code == 422
|
|
assert response.json()["detail"][0]["loc"] == ["header", "x-attempt"]
|
|
response = client.get("/", headers={name: "request", "X-Attempt": "2"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == {"request_id": "request", "x_attempt": 2}
|
|
|
|
|
|
@pytest.mark.parametrize("extra", ["ignore", "allow", "forbid"])
|
|
def test_automatic_headers_follow_extra_configuration(extra):
|
|
app = FastAPI()
|
|
|
|
class RequestHeaders(BaseModel):
|
|
model_config = ConfigDict(extra=extra)
|
|
x_tag: str = "default"
|
|
|
|
@app.get("/")
|
|
def read(headers: Annotated[RequestHeaders | None, Header()] = None):
|
|
return headers
|
|
|
|
client = TestClient(app)
|
|
client.headers.clear()
|
|
response = client.get("/")
|
|
if extra == "forbid":
|
|
assert response.status_code == 422
|
|
assert response.json()["detail"] == [
|
|
{
|
|
"type": "extra_forbidden",
|
|
"loc": ["header", "host"],
|
|
"msg": "Extra inputs are not permitted",
|
|
"input": "testserver",
|
|
}
|
|
]
|
|
else:
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == (
|
|
{"x_tag": "default", "host": "testserver"} if extra == "allow" else None
|
|
)
|
|
|
|
|
|
def test_allowed_header_extras_require_model_fields():
|
|
app = FastAPI()
|
|
|
|
class RequestHeaders(BaseModel):
|
|
model_config = ConfigDict(extra="allow")
|
|
x_id: int
|
|
|
|
@app.get("/")
|
|
def read(headers: Annotated[RequestHeaders | None, Header()] = None):
|
|
return headers
|
|
|
|
client = TestClient(app)
|
|
client.headers.clear()
|
|
response = client.get("/")
|
|
assert response.status_code == 422
|
|
assert response.json()["detail"][0]["loc"] == ["header", "x-id"]
|
|
response = client.get("/", headers={"X-ID": "1"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == {"x_id": 1, "host": "testserver"}
|
|
|
|
|
|
def test_optional_header_models_with_dependency_and_individual_header():
|
|
app = FastAPI()
|
|
|
|
class Trace(BaseModel):
|
|
x_trace: str
|
|
|
|
class ClientInfo(BaseModel):
|
|
x_client: str
|
|
|
|
def get_client_info(info: Annotated[ClientInfo | None, Header()] = None):
|
|
return info
|
|
|
|
@app.get("/")
|
|
def read(
|
|
host: Annotated[str, Header()],
|
|
info: Annotated[ClientInfo | None, Depends(get_client_info)],
|
|
trace: Annotated[Trace | None, Header()] = None,
|
|
):
|
|
return [trace, info]
|
|
|
|
client = TestClient(app)
|
|
client.headers.clear()
|
|
response = client.get("/")
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == [None, None]
|
|
response = client.get("/", headers={"X-Trace": "trace"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == [{"x_trace": "trace"}, None]
|
|
response = client.get("/", headers={"X-Client": "client"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == [None, {"x_client": "client"}]
|
|
response = client.get("/", headers={"X-Trace": "trace", "X-Client": "client"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == [{"x_trace": "trace"}, {"x_client": "client"}]
|
|
assert {
|
|
p["name"]: p["required"]
|
|
for p in app.openapi()["paths"]["/"]["get"]["parameters"]
|
|
} == {"host": True, "x-trace": False, "x-client": False}
|
|
|
|
|
|
def test_header_model_parameter_default_and_required_group():
|
|
app = FastAPI()
|
|
|
|
class Trace(BaseModel):
|
|
x_request_id: str
|
|
|
|
@app.get("/default")
|
|
def read_default(
|
|
trace: Annotated[Trace | None, Header()] = Trace(x_request_id="default"),
|
|
):
|
|
return trace
|
|
|
|
@app.get("/required")
|
|
def read_required(trace: Annotated[Trace | None, Header()]):
|
|
return trace
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/default")
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == {"x_request_id": "default"}
|
|
response = client.get("/required")
|
|
assert response.status_code == 422
|
|
assert response.json()["detail"] == [
|
|
{"type": "missing", "loc": ["header"], "msg": "Field required", "input": None}
|
|
]
|
|
for path in ("/default", "/required"):
|
|
response = client.get(path, headers={"X-Request-ID": "request"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == {"x_request_id": "request"}
|
|
schema = app.openapi()
|
|
assert schema["paths"]["/default"]["get"]["parameters"][0]["required"] is False
|
|
assert schema["paths"]["/required"]["get"]["parameters"][0]["required"] is True
|
|
|
|
|
|
def test_optional_cookie_model():
|
|
app = FastAPI()
|
|
|
|
class Cookies(BaseModel):
|
|
session_id: str
|
|
counter: int
|
|
enabled: bool = True
|
|
|
|
@app.get("/")
|
|
def read(cookies: Annotated[Cookies | None, Cookie()] = None):
|
|
return cookies
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/")
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() is None
|
|
response = client.get("/", headers={"cookie": "session_id=; counter=1; counter=2"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == {"session_id": "", "counter": 2, "enabled": True}
|
|
response = client.get("/", headers={"cookie": "session_id="})
|
|
assert response.status_code == 422
|
|
assert response.json()["detail"][0]["loc"] == ["cookie", "counter"]
|
|
response = client.get("/", headers={"cookie": "session_id=session; counter=bad"})
|
|
assert response.status_code == 422
|
|
assert response.json()["detail"][0]["loc"] == ["cookie", "counter"]
|
|
assert response.json()["detail"][0]["type"] == "int_parsing"
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["sessionId", "legacyId"])
|
|
def test_cookie_aliases_are_case_sensitive(name):
|
|
app = FastAPI()
|
|
|
|
class Cookies(BaseModel):
|
|
session_id: int = Field(
|
|
alias="sessionId", validation_alias=AliasChoices("sessionId", "legacyId")
|
|
)
|
|
|
|
@app.get("/")
|
|
def read(cookies: Annotated[Cookies | None, Cookie()] = None):
|
|
return cookies
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/", headers={"cookie": f"{name.upper()}=1"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() is None
|
|
response = client.get("/", headers={"cookie": f"{name}=1"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == {"sessionId": 1}
|
|
response = client.get("/", headers={"cookie": f"{name}=bad"})
|
|
assert response.status_code == 422
|
|
assert response.json()["detail"][0]["loc"] == ["cookie", name]
|
|
|
|
|
|
@pytest.mark.parametrize("extra", ["ignore", "allow", "forbid"])
|
|
@pytest.mark.parametrize(
|
|
"cookies",
|
|
[{"session_id": "session"}, {"session_id": "session", "tracking": "value"}],
|
|
)
|
|
def test_cookie_extras_and_declared_siblings(extra, cookies):
|
|
app = FastAPI()
|
|
|
|
class Preferences(BaseModel):
|
|
model_config = ConfigDict(extra=extra)
|
|
theme: str = "light"
|
|
|
|
@app.get("/")
|
|
def read(
|
|
session_id: Annotated[str, Cookie()],
|
|
preferences: Annotated[Preferences | None, Cookie()] = None,
|
|
):
|
|
return preferences
|
|
|
|
response = TestClient(app).get(
|
|
"/",
|
|
headers={
|
|
"cookie": "; ".join(f"{key}={value}" for key, value in cookies.items())
|
|
},
|
|
)
|
|
if extra == "forbid":
|
|
assert response.status_code == 422
|
|
assert [error["loc"] for error in response.json()["detail"]] == [
|
|
["cookie", name] for name in cookies
|
|
]
|
|
assert all(
|
|
error["type"] == "extra_forbidden" for error in response.json()["detail"]
|
|
)
|
|
else:
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == (
|
|
{"theme": "light", **cookies} if extra == "allow" else None
|
|
)
|
|
|
|
|
|
def test_optional_cookie_models_with_dependency():
|
|
app = FastAPI()
|
|
|
|
class Preferences(BaseModel):
|
|
theme: str
|
|
|
|
class Session(BaseModel):
|
|
session_id: str
|
|
|
|
def get_session(session: Annotated[Session | None, Cookie()] = None):
|
|
return session
|
|
|
|
@app.get("/")
|
|
def read(
|
|
session: Annotated[Session | None, Depends(get_session)],
|
|
preferences: Annotated[Preferences | None, Cookie()] = None,
|
|
):
|
|
return [preferences, session]
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/")
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == [None, None]
|
|
response = client.get("/", headers={"cookie": "session_id=session"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == [None, {"session_id": "session"}]
|
|
response = client.get("/", headers={"cookie": "theme=dark"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == [{"theme": "dark"}, None]
|
|
response = client.get("/", headers={"cookie": "session_id=session; theme=dark"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == [{"theme": "dark"}, {"session_id": "session"}]
|
|
assert {
|
|
p["name"]: p["required"]
|
|
for p in app.openapi()["paths"]["/"]["get"]["parameters"]
|
|
} == {"theme": False, "session_id": False}
|
|
|
|
|
|
def test_cookie_model_default_factory_and_required_group():
|
|
app = FastAPI()
|
|
calls = []
|
|
|
|
class Preferences(BaseModel):
|
|
theme: str
|
|
|
|
def factory():
|
|
calls.append("factory")
|
|
return Preferences(theme="light")
|
|
|
|
@app.get("/default")
|
|
def read_default(
|
|
preferences: Annotated[Preferences | None, Cookie(default_factory=factory)],
|
|
):
|
|
return preferences
|
|
|
|
@app.get("/required")
|
|
def read_required(preferences: Annotated[Preferences | None, Cookie()]):
|
|
return preferences
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/default")
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == {"theme": "light"}
|
|
assert calls == ["factory"]
|
|
response = client.get("/required")
|
|
assert response.status_code == 422
|
|
assert response.json()["detail"] == [
|
|
{"type": "missing", "loc": ["cookie"], "msg": "Field required", "input": None}
|
|
]
|
|
for path in ("/default", "/required"):
|
|
response = client.get(path, headers={"cookie": "theme=dark"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == {"theme": "dark"}
|
|
assert calls == ["factory"]
|
|
schema = app.openapi()
|
|
assert schema["paths"]["/default"]["get"]["parameters"][0]["required"] is False
|
|
assert schema["paths"]["/required"]["get"]["parameters"][0]["required"] is True
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"request_kwargs, expected",
|
|
[
|
|
({}, [None, None, None]),
|
|
({"params": {"value": "1"}}, [{"value": 1}, None, None]),
|
|
({"headers": {"value": "2"}}, [None, {"value": 2}, None]),
|
|
({"headers": {"cookie": "value=3"}}, [None, None, {"value": 3}]),
|
|
(
|
|
{"params": {"value": "1"}, "headers": {"value": "2", "cookie": "value=3"}},
|
|
[{"value": 1}, {"value": 2}, {"value": 3}],
|
|
),
|
|
],
|
|
)
|
|
def test_optional_models_keep_request_locations_separate(request_kwargs, expected):
|
|
app = FastAPI()
|
|
|
|
class Model(BaseModel):
|
|
value: int
|
|
|
|
@app.get("/")
|
|
def read(
|
|
query: Annotated[Model | None, Query()] = None,
|
|
header: Annotated[Model | None, Header()] = None,
|
|
cookie: Annotated[Model | None, Cookie()] = None,
|
|
):
|
|
return [query, header, cookie]
|
|
|
|
response = TestClient(app).get("/", **request_kwargs)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json() == expected
|