mirror of
https://github.com/fastapi/fastapi.git
synced 2026-10-30 05:41:02 -04:00
397 lines
15 KiB
Python
397 lines
15 KiB
Python
import re
|
|
from typing import Annotated
|
|
|
|
import pytest
|
|
from fastapi import Body, Cookie, FastAPI, File, Form, Header, Path, Query, UploadFile
|
|
from fastapi.testclient import TestClient
|
|
from inline_snapshot import snapshot
|
|
from pydantic import Field
|
|
|
|
|
|
@pytest.mark.parametrize("parameter", [Path, Query, Header, Cookie, Body, Form])
|
|
@pytest.mark.parametrize(
|
|
"mode,expected", [(None, "123"), ("smart", "123"), ("left_to_right", 123)]
|
|
)
|
|
def test_union_mode(parameter, mode, expected):
|
|
app = FastAPI()
|
|
|
|
@app.post("/{value}" if parameter is Path else "/")
|
|
def read_value(value: Annotated[int | str, parameter(union_mode=mode)]):
|
|
return value
|
|
|
|
client = TestClient(app)
|
|
if parameter is Path:
|
|
response = client.post("/123")
|
|
elif parameter is Query:
|
|
response = client.post("/", params={"value": "123"})
|
|
elif parameter is Header:
|
|
response = client.post("/", headers={"value": "123"})
|
|
elif parameter is Cookie:
|
|
response = client.post("/", headers={"cookie": "value=123"})
|
|
elif parameter is Body:
|
|
response = client.post("/", json="123")
|
|
else:
|
|
response = client.post("/", data={"value": "123"})
|
|
assert response.status_code == 200
|
|
assert response.json() == expected
|
|
|
|
|
|
@pytest.mark.parametrize("parameter", [Query, Header, Body, Form])
|
|
@pytest.mark.parametrize("fail_fast,error_count", [(True, 1), (False, 2)])
|
|
def test_fail_fast(parameter, fail_fast, error_count):
|
|
app = FastAPI()
|
|
|
|
@app.post("/")
|
|
def read_values(value: Annotated[list[int], parameter(fail_fast=fail_fast)]):
|
|
return value
|
|
|
|
client = TestClient(app)
|
|
if parameter is Query:
|
|
response = client.post("/", params={"value": ["invalid", "also-invalid"]})
|
|
elif parameter is Header:
|
|
response = client.post(
|
|
"/", headers=[("value", "invalid"), ("value", "also-invalid")]
|
|
)
|
|
elif parameter is Body:
|
|
response = client.post("/", json=["invalid", "also-invalid"])
|
|
else:
|
|
response = client.post("/", data={"value": ["invalid", "also-invalid"]})
|
|
assert response.status_code == 422
|
|
assert len(response.json()["detail"]) == error_count
|
|
assert response.json()["detail"][0]["type"] == "int_parsing"
|
|
assert response.json()["detail"][0]["loc"][-1] == 0
|
|
|
|
if parameter is Query:
|
|
response = client.post("/", params={"value": ["1", "2"]})
|
|
elif parameter is Header:
|
|
response = client.post("/", headers=[("value", "1"), ("value", "2")])
|
|
elif parameter is Body:
|
|
response = client.post("/", json=[1, 2])
|
|
else:
|
|
response = client.post("/", data={"value": ["1", "2"]})
|
|
assert response.status_code == 200
|
|
assert response.json() == [1, 2]
|
|
|
|
|
|
def test_omitted_options_preserve_pydantic_validation_metadata():
|
|
app = FastAPI()
|
|
|
|
@app.post("/")
|
|
def read_values(
|
|
number: Annotated[int | str, Field(union_mode="left_to_right"), Query()],
|
|
values: Annotated[list[int], Field(fail_fast=True), Body()],
|
|
):
|
|
return {"number": number, "values": values}
|
|
|
|
client = TestClient(app)
|
|
response = client.post("/?number=123", json=[1, 2])
|
|
assert response.status_code == 200
|
|
assert response.json() == {"number": 123, "values": [1, 2]}
|
|
|
|
response = client.post("/?number=123", json=["invalid", "also-invalid"])
|
|
assert response.status_code == 422
|
|
assert len(response.json()["detail"]) == 1
|
|
|
|
|
|
def test_explicit_options_override_pydantic_validation_metadata():
|
|
app = FastAPI()
|
|
|
|
@app.post("/")
|
|
def read_values(
|
|
number: Annotated[
|
|
int | str, Field(union_mode="left_to_right"), Query(union_mode="smart")
|
|
],
|
|
values: Annotated[list[int], Field(fail_fast=True), Body(fail_fast=False)],
|
|
):
|
|
return {"number": number, "values": values}
|
|
|
|
client = TestClient(app)
|
|
response = client.post("/?number=123", json=[1, 2])
|
|
assert response.status_code == 200
|
|
assert response.json() == {"number": "123", "values": [1, 2]}
|
|
|
|
response = client.post("/?number=123", json=["invalid", "also-invalid"])
|
|
assert response.status_code == 422
|
|
assert [error["loc"] for error in response.json()["detail"]] == [
|
|
["body", 0],
|
|
["body", 1],
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("parameter", [Path, Query, Header, Cookie, Body, Form])
|
|
@pytest.mark.parametrize("value,expected_status", [("ABC", 200), ("xyz", 422)])
|
|
def test_compiled_pattern(parameter, value, expected_status):
|
|
app = FastAPI()
|
|
|
|
@app.post("/{value}" if parameter is Path else "/")
|
|
def read_value(
|
|
value: Annotated[str, parameter(pattern=re.compile("^abc$", re.IGNORECASE))],
|
|
):
|
|
return value
|
|
|
|
client = TestClient(app)
|
|
if parameter is Path:
|
|
response = client.post(f"/{value}")
|
|
elif parameter is Query:
|
|
response = client.post("/", params={"value": value})
|
|
elif parameter is Header:
|
|
response = client.post("/", headers={"value": value})
|
|
elif parameter is Cookie:
|
|
response = client.post("/", headers={"cookie": f"value={value}"})
|
|
elif parameter is Body:
|
|
response = client.post("/", json=value)
|
|
else:
|
|
response = client.post("/", data={"value": value})
|
|
assert response.status_code == expected_status
|
|
if expected_status == 200:
|
|
assert response.json() == value
|
|
else:
|
|
assert response.json()["detail"][0]["type"] == "string_pattern_mismatch"
|
|
|
|
|
|
@pytest.mark.parametrize("fail_fast,error_count", [(True, 1), (False, 2)])
|
|
def test_file_list_fail_fast(fail_fast, error_count):
|
|
app = FastAPI()
|
|
|
|
@app.post("/")
|
|
def upload(files: Annotated[list[UploadFile], File(fail_fast=fail_fast)]):
|
|
return [file.filename for file in files]
|
|
|
|
client = TestClient(app)
|
|
response = client.post(
|
|
"/",
|
|
files=[("files", (None, "not-a-file")), ("files", (None, "also-not-a-file"))],
|
|
)
|
|
assert response.status_code == 422
|
|
assert len(response.json()["detail"]) == error_count
|
|
assert response.json()["detail"][0]["loc"] == ["body", "files", 0]
|
|
|
|
response = client.post("/", files={"files": ("example.txt", b"hello")})
|
|
assert response.status_code == 200
|
|
assert response.json() == ["example.txt"]
|
|
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
def customize_schema(schema):
|
|
schema["x-custom"] = "preserved"
|
|
|
|
|
|
@app.get("/query")
|
|
def read_query(
|
|
value: Annotated[int | str, Query(union_mode="left_to_right")],
|
|
items: Annotated[list[int], Query(fail_fast=True)],
|
|
count: Annotated[int, Query(validate_default=True)] = 1,
|
|
label: Annotated[
|
|
str,
|
|
Query(
|
|
pattern=re.compile("^abc$", re.IGNORECASE),
|
|
json_schema_extra=customize_schema,
|
|
),
|
|
] = "ABC",
|
|
):
|
|
return {"value": value, "items": items, "count": count, "label": label}
|
|
|
|
|
|
@app.post("/body")
|
|
def read_body(
|
|
items: Annotated[list[int], Body(fail_fast=True)],
|
|
number: Annotated[int | str, Body(union_mode="left_to_right")],
|
|
count: Annotated[
|
|
int, Body(validate_default=True, json_schema_extra=customize_schema)
|
|
] = 1,
|
|
):
|
|
return {"items": items, "number": number, "count": count}
|
|
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def test_query_options():
|
|
response = client.get("/query", params={"value": "123", "items": ["1", "2"]})
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"value": 123,
|
|
"items": [1, 2],
|
|
"count": 1,
|
|
"label": "ABC",
|
|
}
|
|
|
|
|
|
def test_body_options():
|
|
response = client.post("/body", json={"items": [1, 2], "number": "123"})
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"items": [1, 2],
|
|
"number": 123,
|
|
"count": 1,
|
|
}
|
|
|
|
|
|
def test_openapi_schema():
|
|
response = client.get("/openapi.json")
|
|
assert response.status_code == 200
|
|
assert response.json() == snapshot(
|
|
{
|
|
"openapi": "3.1.0",
|
|
"info": {"title": "FastAPI", "version": "0.1.0"},
|
|
"paths": {
|
|
"/query": {
|
|
"get": {
|
|
"summary": "Read Query",
|
|
"operationId": "read_query_query_get",
|
|
"parameters": [
|
|
{
|
|
"name": "value",
|
|
"in": "query",
|
|
"required": True,
|
|
"schema": {
|
|
"anyOf": [{"type": "integer"}, {"type": "string"}],
|
|
"title": "Value",
|
|
},
|
|
},
|
|
{
|
|
"name": "items",
|
|
"in": "query",
|
|
"required": True,
|
|
"schema": {
|
|
"type": "array",
|
|
"items": {"type": "integer"},
|
|
"title": "Items",
|
|
},
|
|
},
|
|
{
|
|
"name": "count",
|
|
"in": "query",
|
|
"required": False,
|
|
"schema": {
|
|
"type": "integer",
|
|
"default": 1,
|
|
"title": "Count",
|
|
},
|
|
},
|
|
{
|
|
"name": "label",
|
|
"in": "query",
|
|
"required": False,
|
|
"schema": {
|
|
"type": "string",
|
|
"pattern": "^abc$",
|
|
"x-custom": "preserved",
|
|
"default": "ABC",
|
|
"title": "Label",
|
|
},
|
|
},
|
|
],
|
|
"responses": {
|
|
"200": {
|
|
"description": "Successful Response",
|
|
"content": {"application/json": {"schema": {}}},
|
|
},
|
|
"422": {
|
|
"description": "Validation Error",
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {
|
|
"$ref": "#/components/schemas/HTTPValidationError"
|
|
}
|
|
}
|
|
},
|
|
},
|
|
},
|
|
}
|
|
},
|
|
"/body": {
|
|
"post": {
|
|
"summary": "Read Body",
|
|
"operationId": "read_body_body_post",
|
|
"requestBody": {
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {
|
|
"$ref": "#/components/schemas/Body_read_body_body_post"
|
|
}
|
|
}
|
|
},
|
|
"required": True,
|
|
},
|
|
"responses": {
|
|
"200": {
|
|
"description": "Successful Response",
|
|
"content": {"application/json": {"schema": {}}},
|
|
},
|
|
"422": {
|
|
"description": "Validation Error",
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {
|
|
"$ref": "#/components/schemas/HTTPValidationError"
|
|
}
|
|
}
|
|
},
|
|
},
|
|
},
|
|
}
|
|
},
|
|
},
|
|
"components": {
|
|
"schemas": {
|
|
"Body_read_body_body_post": {
|
|
"properties": {
|
|
"items": {
|
|
"items": {"type": "integer"},
|
|
"type": "array",
|
|
"title": "Items",
|
|
},
|
|
"number": {
|
|
"anyOf": [{"type": "integer"}, {"type": "string"}],
|
|
"title": "Number",
|
|
},
|
|
"count": {
|
|
"type": "integer",
|
|
"title": "Count",
|
|
"default": 1,
|
|
"x-custom": "preserved",
|
|
},
|
|
},
|
|
"type": "object",
|
|
"required": ["items", "number"],
|
|
"title": "Body_read_body_body_post",
|
|
},
|
|
"HTTPValidationError": {
|
|
"properties": {
|
|
"detail": {
|
|
"items": {
|
|
"$ref": "#/components/schemas/ValidationError"
|
|
},
|
|
"type": "array",
|
|
"title": "Detail",
|
|
}
|
|
},
|
|
"type": "object",
|
|
"title": "HTTPValidationError",
|
|
},
|
|
"ValidationError": {
|
|
"properties": {
|
|
"loc": {
|
|
"items": {
|
|
"anyOf": [{"type": "string"}, {"type": "integer"}]
|
|
},
|
|
"type": "array",
|
|
"title": "Location",
|
|
},
|
|
"msg": {"type": "string", "title": "Message"},
|
|
"type": {"type": "string", "title": "Error Type"},
|
|
"input": {"title": "Input"},
|
|
"ctx": {"type": "object", "title": "Context"},
|
|
},
|
|
"type": "object",
|
|
"required": ["loc", "msg", "type"],
|
|
"title": "ValidationError",
|
|
},
|
|
}
|
|
},
|
|
}
|
|
)
|