Sebastián Ramírez
8 months ago
29 changed files with 814 additions and 255 deletions
|
After Width: | Height: | Size: 7.1 KiB |
|
After Width: | Height: | Size: 7.2 KiB |
@ -0,0 +1,35 @@ |
|||||
|
from typing import Optional |
||||
|
|
||||
|
from fastapi import FastAPI, File, Form |
||||
|
from starlette.testclient import TestClient |
||||
|
from typing_extensions import Annotated |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.post("/urlencoded") |
||||
|
async def post_url_encoded(age: Annotated[Optional[int], Form()] = None): |
||||
|
return age |
||||
|
|
||||
|
|
||||
|
@app.post("/multipart") |
||||
|
async def post_multi_part( |
||||
|
age: Annotated[Optional[int], Form()] = None, |
||||
|
file: Annotated[Optional[bytes], File()] = None, |
||||
|
): |
||||
|
return {"file": file, "age": age} |
||||
|
|
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_form_default_url_encoded(): |
||||
|
response = client.post("/urlencoded", data={"age": ""}) |
||||
|
assert response.status_code == 200 |
||||
|
assert response.text == "null" |
||||
|
|
||||
|
|
||||
|
def test_form_default_multi_part(): |
||||
|
response = client.post("/multipart", data={"age": ""}) |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"file": None, "age": None} |
||||
@ -0,0 +1,30 @@ |
|||||
|
from typing import List, Optional |
||||
|
|
||||
|
from fastapi import FastAPI, File |
||||
|
from fastapi.testclient import TestClient |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.post("/files") |
||||
|
async def upload_files(files: Optional[List[bytes]] = File(None)): |
||||
|
if files is None: |
||||
|
return {"files_count": 0} |
||||
|
return {"files_count": len(files), "sizes": [len(f) for f in files]} |
||||
|
|
||||
|
|
||||
|
def test_optional_bytes_list(): |
||||
|
client = TestClient(app) |
||||
|
response = client.post( |
||||
|
"/files", |
||||
|
files=[("files", b"content1"), ("files", b"content2")], |
||||
|
) |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"files_count": 2, "sizes": [8, 8]} |
||||
|
|
||||
|
|
||||
|
def test_optional_bytes_list_no_files(): |
||||
|
client = TestClient(app) |
||||
|
response = client.post("/files") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"files_count": 0} |
||||
@ -0,0 +1,111 @@ |
|||||
|
from fastapi import Cookie, FastAPI, Header, Query |
||||
|
from fastapi._compat import PYDANTIC_V2 |
||||
|
from fastapi.testclient import TestClient |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class Model(BaseModel): |
||||
|
param: str |
||||
|
|
||||
|
if PYDANTIC_V2: |
||||
|
model_config = {"extra": "allow"} |
||||
|
else: |
||||
|
|
||||
|
class Config: |
||||
|
extra = "allow" |
||||
|
|
||||
|
|
||||
|
@app.get("/query") |
||||
|
async def query_model_with_extra(data: Model = Query()): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
@app.get("/header") |
||||
|
async def header_model_with_extra(data: Model = Header()): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
@app.get("/cookie") |
||||
|
async def cookies_model_with_extra(data: Model = Cookie()): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
def test_query_pass_extra_list(): |
||||
|
client = TestClient(app) |
||||
|
resp = client.get( |
||||
|
"/query", |
||||
|
params={ |
||||
|
"param": "123", |
||||
|
"param2": ["456", "789"], # Pass a list of values as extra parameter |
||||
|
}, |
||||
|
) |
||||
|
assert resp.status_code == 200 |
||||
|
assert resp.json() == { |
||||
|
"param": "123", |
||||
|
"param2": ["456", "789"], |
||||
|
} |
||||
|
|
||||
|
|
||||
|
def test_query_pass_extra_single(): |
||||
|
client = TestClient(app) |
||||
|
resp = client.get( |
||||
|
"/query", |
||||
|
params={ |
||||
|
"param": "123", |
||||
|
"param2": "456", |
||||
|
}, |
||||
|
) |
||||
|
assert resp.status_code == 200 |
||||
|
assert resp.json() == { |
||||
|
"param": "123", |
||||
|
"param2": "456", |
||||
|
} |
||||
|
|
||||
|
|
||||
|
def test_header_pass_extra_list(): |
||||
|
client = TestClient(app) |
||||
|
|
||||
|
resp = client.get( |
||||
|
"/header", |
||||
|
headers=[ |
||||
|
("param", "123"), |
||||
|
("param2", "456"), # Pass a list of values as extra parameter |
||||
|
("param2", "789"), |
||||
|
], |
||||
|
) |
||||
|
assert resp.status_code == 200 |
||||
|
resp_json = resp.json() |
||||
|
assert "param2" in resp_json |
||||
|
assert resp_json["param2"] == ["456", "789"] |
||||
|
|
||||
|
|
||||
|
def test_header_pass_extra_single(): |
||||
|
client = TestClient(app) |
||||
|
|
||||
|
resp = client.get( |
||||
|
"/header", |
||||
|
headers=[ |
||||
|
("param", "123"), |
||||
|
("param2", "456"), |
||||
|
], |
||||
|
) |
||||
|
assert resp.status_code == 200 |
||||
|
resp_json = resp.json() |
||||
|
assert "param2" in resp_json |
||||
|
assert resp_json["param2"] == "456" |
||||
|
|
||||
|
|
||||
|
def test_cookie_pass_extra_list(): |
||||
|
client = TestClient(app) |
||||
|
client.cookies = [ |
||||
|
("param", "123"), |
||||
|
("param2", "456"), # Pass a list of values as extra parameter |
||||
|
("param2", "789"), |
||||
|
] |
||||
|
resp = client.get("/cookie") |
||||
|
assert resp.status_code == 200 |
||||
|
resp_json = resp.json() |
||||
|
assert "param2" in resp_json |
||||
|
assert resp_json["param2"] == "789" # Cookies only keep the last value |
||||
@ -0,0 +1,76 @@ |
|||||
|
from dirty_equals import IsPartialDict |
||||
|
from fastapi import Cookie, FastAPI, Header, Query |
||||
|
from fastapi._compat import PYDANTIC_V2 |
||||
|
from fastapi.testclient import TestClient |
||||
|
from pydantic import BaseModel, Field |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class Model(BaseModel): |
||||
|
param: str = Field(alias="param_alias") |
||||
|
|
||||
|
|
||||
|
@app.get("/query") |
||||
|
async def query_model(data: Model = Query()): |
||||
|
return {"param": data.param} |
||||
|
|
||||
|
|
||||
|
@app.get("/header") |
||||
|
async def header_model(data: Model = Header()): |
||||
|
return {"param": data.param} |
||||
|
|
||||
|
|
||||
|
@app.get("/cookie") |
||||
|
async def cookie_model(data: Model = Cookie()): |
||||
|
return {"param": data.param} |
||||
|
|
||||
|
|
||||
|
def test_query_model_with_alias(): |
||||
|
client = TestClient(app) |
||||
|
response = client.get("/query", params={"param_alias": "value"}) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"param": "value"} |
||||
|
|
||||
|
|
||||
|
def test_header_model_with_alias(): |
||||
|
client = TestClient(app) |
||||
|
response = client.get("/header", headers={"param_alias": "value"}) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"param": "value"} |
||||
|
|
||||
|
|
||||
|
def test_cookie_model_with_alias(): |
||||
|
client = TestClient(app) |
||||
|
client.cookies.set("param_alias", "value") |
||||
|
response = client.get("/cookie") |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"param": "value"} |
||||
|
|
||||
|
|
||||
|
def test_query_model_with_alias_by_name(): |
||||
|
client = TestClient(app) |
||||
|
response = client.get("/query", params={"param": "value"}) |
||||
|
assert response.status_code == 422, response.text |
||||
|
details = response.json() |
||||
|
if PYDANTIC_V2: |
||||
|
assert details["detail"][0]["input"] == {"param": "value"} |
||||
|
|
||||
|
|
||||
|
def test_header_model_with_alias_by_name(): |
||||
|
client = TestClient(app) |
||||
|
response = client.get("/header", headers={"param": "value"}) |
||||
|
assert response.status_code == 422, response.text |
||||
|
details = response.json() |
||||
|
if PYDANTIC_V2: |
||||
|
assert details["detail"][0]["input"] == IsPartialDict({"param": "value"}) |
||||
|
|
||||
|
|
||||
|
def test_cookie_model_with_alias_by_name(): |
||||
|
client = TestClient(app) |
||||
|
client.cookies.set("param", "value") |
||||
|
response = client.get("/cookie") |
||||
|
assert response.status_code == 422, response.text |
||||
|
details = response.json() |
||||
|
if PYDANTIC_V2: |
||||
|
assert details["detail"][0]["input"] == {"param": "value"} |
||||
@ -0,0 +1,92 @@ |
|||||
|
import pytest |
||||
|
from fastapi import FastAPI |
||||
|
from fastapi.testclient import TestClient |
||||
|
from inline_snapshot import snapshot |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
from tests.utils import needs_py310, needs_pydanticv2 |
||||
|
|
||||
|
|
||||
|
@pytest.fixture(name="client") |
||||
|
def get_client(): |
||||
|
from enum import Enum |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
class PlatformRole(str, Enum): |
||||
|
admin = "admin" |
||||
|
user = "user" |
||||
|
|
||||
|
class OtherRole(str, Enum): ... |
||||
|
|
||||
|
class User(BaseModel): |
||||
|
username: str |
||||
|
role: PlatformRole | OtherRole |
||||
|
|
||||
|
@app.get("/users") |
||||
|
async def get_user() -> User: |
||||
|
return {"username": "alice", "role": "admin"} |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
return client |
||||
|
|
||||
|
|
||||
|
@needs_py310 |
||||
|
@needs_pydanticv2 |
||||
|
def test_get(client: TestClient): |
||||
|
response = client.get("/users") |
||||
|
assert response.json() == {"username": "alice", "role": "admin"} |
||||
|
|
||||
|
|
||||
|
@needs_py310 |
||||
|
@needs_pydanticv2 |
||||
|
def test_openapi_schema(client: TestClient): |
||||
|
response = client.get("openapi.json") |
||||
|
assert response.json() == snapshot( |
||||
|
{ |
||||
|
"openapi": "3.1.0", |
||||
|
"info": {"title": "FastAPI", "version": "0.1.0"}, |
||||
|
"paths": { |
||||
|
"/users": { |
||||
|
"get": { |
||||
|
"summary": "Get User", |
||||
|
"operationId": "get_user_users_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": { |
||||
|
"application/json": { |
||||
|
"schema": {"$ref": "#/components/schemas/User"} |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
"components": { |
||||
|
"schemas": { |
||||
|
"PlatformRole": { |
||||
|
"type": "string", |
||||
|
"enum": ["admin", "user"], |
||||
|
"title": "PlatformRole", |
||||
|
}, |
||||
|
"User": { |
||||
|
"properties": { |
||||
|
"username": {"type": "string", "title": "Username"}, |
||||
|
"role": { |
||||
|
"anyOf": [ |
||||
|
{"$ref": "#/components/schemas/PlatformRole"}, |
||||
|
{"enum": [], "title": "OtherRole"}, |
||||
|
], |
||||
|
"title": "Role", |
||||
|
}, |
||||
|
}, |
||||
|
"type": "object", |
||||
|
"required": ["username", "role"], |
||||
|
"title": "User", |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
) |
||||
Loading…
Reference in new issue