From 0bbb728cefd753eae138f39bbe1b022ce832562e Mon Sep 17 00:00:00 2001 From: Suren Khorenyan Date: Sun, 3 May 2026 19:30:14 +0300 Subject: [PATCH] test: add examples and coverage --- docs_src/app_testing/tutorial001_py310.py | 4 +- .../tutorial001_an_py310.py | 53 ++++++ .../tutorial001_an_py310.py | 57 +++++++ .../tutorial001_an_py310.py | 53 ++++++ .../tutorial001_an_py310.py | 59 +++++++ .../tutorial001_an_py310.py | 77 +++++++++ tests/_annotated_body_depends_merge_common.py | 42 +++++ tests/docs_src/__init__.py | 0 tests/docs_src/_loader.py | 24 +++ ...st_body_depends_model_merge_tutorial001.py | 34 ++++ ...st_file_depends_model_merge_tutorial001.py | 53 ++++++ ...st_form_depends_model_merge_tutorial001.py | 34 ++++ ...t_query_depends_model_merge_tutorial001.py | 42 +++++ .../test_query_plus_body_tutorial001.py | 39 +++++ .../test_annotated_body_depends_merge_body.py | 161 ++++++++++++++++++ .../test_annotated_body_depends_merge_file.py | 100 +++++++++++ .../test_annotated_body_depends_merge_form.py | 64 +++++++ ...ted_body_depends_merge_query_plus_shape.py | 144 ++++++++++++++++ 18 files changed, 1038 insertions(+), 2 deletions(-) create mode 100644 docs_src/body_depends_model_merge_body/tutorial001_an_py310.py create mode 100644 docs_src/body_depends_model_merge_file/tutorial001_an_py310.py create mode 100644 docs_src/body_depends_model_merge_form/tutorial001_an_py310.py create mode 100644 docs_src/body_depends_model_merge_query/tutorial001_an_py310.py create mode 100644 docs_src/body_depends_model_merge_query_plus_body/tutorial001_an_py310.py create mode 100644 tests/_annotated_body_depends_merge_common.py create mode 100644 tests/docs_src/__init__.py create mode 100644 tests/docs_src/_loader.py create mode 100644 tests/docs_src/body_depends_model_merge_body/test_body_depends_model_merge_tutorial001.py create mode 100644 tests/docs_src/body_depends_model_merge_file/test_file_depends_model_merge_tutorial001.py create mode 100644 tests/docs_src/body_depends_model_merge_form/test_form_depends_model_merge_tutorial001.py create mode 100644 tests/docs_src/body_depends_model_merge_query/test_query_depends_model_merge_tutorial001.py create mode 100644 tests/docs_src/body_depends_model_merge_query_plus_body/test_query_plus_body_tutorial001.py create mode 100644 tests/test_annotated_body_depends_merge_body.py create mode 100644 tests/test_annotated_body_depends_merge_file.py create mode 100644 tests/test_annotated_body_depends_merge_form.py create mode 100644 tests/test_annotated_body_depends_merge_query_plus_shape.py diff --git a/docs_src/app_testing/tutorial001_py310.py b/docs_src/app_testing/tutorial001_py310.py index 79a853b487..a2aa080c79 100644 --- a/docs_src/app_testing/tutorial001_py310.py +++ b/docs_src/app_testing/tutorial001_py310.py @@ -1,4 +1,4 @@ -from fastapi import FastAPI +from fastapi import FastAPI, status from fastapi.testclient import TestClient app = FastAPI() @@ -14,5 +14,5 @@ client = TestClient(app) def test_read_main(): response = client.get("/") - assert response.status_code == 200 + assert response.status_code == status.HTTP_200_OK, response.text assert response.json() == {"msg": "Hello World"} diff --git a/docs_src/body_depends_model_merge_body/tutorial001_an_py310.py b/docs_src/body_depends_model_merge_body/tutorial001_an_py310.py new file mode 100644 index 0000000000..dffbc33faa --- /dev/null +++ b/docs_src/body_depends_model_merge_body/tutorial001_an_py310.py @@ -0,0 +1,53 @@ +from typing import Annotated + +from fastapi import APIRouter, Body, Depends, FastAPI +from pydantic import BaseModel + +app = FastAPI() +items_router = APIRouter() + + +class ItemBase(BaseModel): + name: str + + +class Gadget(ItemBase): + description: str + + +class Part(ItemBase): + sku: str + + +def register_post_route( + router: APIRouter, + path: str, + schema: type[ItemBase], +): + @router.post(path) + def create_entity( + entity: Annotated[ + ItemBase, + Body(), + Depends(schema), + ], + ): + return entity + + return create_entity + + +register_post_route( + items_router, + "/objects/gadgets/", + Gadget, +) +register_post_route( + items_router, + "/objects/parts/", + Part, +) +app.include_router( + items_router, + prefix="/items", +) diff --git a/docs_src/body_depends_model_merge_file/tutorial001_an_py310.py b/docs_src/body_depends_model_merge_file/tutorial001_an_py310.py new file mode 100644 index 0000000000..d78a31e2a1 --- /dev/null +++ b/docs_src/body_depends_model_merge_file/tutorial001_an_py310.py @@ -0,0 +1,57 @@ +from typing import Annotated, Any + +from fastapi import APIRouter, Depends, FastAPI, File, UploadFile +from pydantic import BaseModel + +app = FastAPI() +files_router = APIRouter() + + +class AttachmentBase(BaseModel): + file: UploadFile + + +class CommentedAttachment(AttachmentBase): + comment: str = "" + + +class NamedAttachment(AttachmentBase): + name: str = "" + + +def register_upload_route( + router: APIRouter, + path: str, + schema: type[AttachmentBase], +): + @router.post(path) + async def upload( + attachment: Annotated[ + AttachmentBase, + File(), + Depends(schema), + ], + ) -> dict[str, Any]: + return { + "filename": attachment.file.filename, + "content_type": attachment.file.content_type, + "data": attachment.model_dump(exclude={"file"}), + } + + return upload + + +register_upload_route( + files_router, + "/attachments/commented/", + CommentedAttachment, +) +register_upload_route( + files_router, + "/attachments/named/", + NamedAttachment, +) +app.include_router( + files_router, + prefix="/files", +) diff --git a/docs_src/body_depends_model_merge_form/tutorial001_an_py310.py b/docs_src/body_depends_model_merge_form/tutorial001_an_py310.py new file mode 100644 index 0000000000..e7c1fc1764 --- /dev/null +++ b/docs_src/body_depends_model_merge_form/tutorial001_an_py310.py @@ -0,0 +1,53 @@ +from typing import Annotated + +from fastapi import APIRouter, Depends, FastAPI, Form +from pydantic import BaseModel + +app = FastAPI() +auth_router = APIRouter() + + +class AccountBase(BaseModel): + username: str + + +class PasswordLogin(AccountBase): + password: str + + +class TokenLogin(AccountBase): + token: str + + +def register_form_post( + router: APIRouter, + path: str, + schema: type[AccountBase], +): + @router.post(path) + def authenticate( + data: Annotated[ + AccountBase, + Form(), + Depends(schema), + ], + ): + return data + + return authenticate + + +register_form_post( + auth_router, + "/session/password/", + PasswordLogin, +) +register_form_post( + auth_router, + "/session/token/", + TokenLogin, +) +app.include_router( + auth_router, + prefix="/auth", +) diff --git a/docs_src/body_depends_model_merge_query/tutorial001_an_py310.py b/docs_src/body_depends_model_merge_query/tutorial001_an_py310.py new file mode 100644 index 0000000000..0c289f6a0e --- /dev/null +++ b/docs_src/body_depends_model_merge_query/tutorial001_an_py310.py @@ -0,0 +1,59 @@ +from typing import Annotated + +from fastapi import APIRouter, Depends, FastAPI, Query +from pydantic import BaseModel + +app = FastAPI() +catalog_router = APIRouter() + + +class ProductFiltersBase(BaseModel): + category: str | None = None + + +class ProductFiltersFull(ProductFiltersBase): + in_stock: bool = True + + +class ProductFiltersPaginated(ProductFiltersBase): + page: int = 1 + per_page: int = 10 + + +def register_product_list( + router: APIRouter, + path: str, + schema: type[ProductFiltersBase], +): + @router.get(path) + def list_products( + params: Annotated[ + ProductFiltersBase, + Query(), # optional + Depends(schema), + ], + ): + return params + + return list_products + + +register_product_list( + catalog_router, + "/items/", + ProductFiltersFull, +) +register_product_list( + catalog_router, + "/items-paginated/", + ProductFiltersPaginated, +) +register_product_list( + catalog_router, + "/basics/", + ProductFiltersBase, +) +app.include_router( + catalog_router, + prefix="/catalog", +) diff --git a/docs_src/body_depends_model_merge_query_plus_body/tutorial001_an_py310.py b/docs_src/body_depends_model_merge_query_plus_body/tutorial001_an_py310.py new file mode 100644 index 0000000000..ae647e437d --- /dev/null +++ b/docs_src/body_depends_model_merge_query_plus_body/tutorial001_an_py310.py @@ -0,0 +1,77 @@ +from typing import Annotated + +from fastapi import APIRouter, Body, Depends, FastAPI, Query +from pydantic import BaseModel + +app = FastAPI() +records_router = APIRouter() + + +class ClientInfoBase(BaseModel): + client_id: str + + +class RetailClientInfo(ClientInfoBase): + region: str | None = None + + +class PartnerClientInfo(ClientInfoBase): + contract_ref: str + + +class RecordBase(BaseModel): + title: str + + +class CaseFileRecord(RecordBase): + case_number: str + + +class ContractRecord(RecordBase): + contract_id: str + + +def register_record_route( + router: APIRouter, + path: str, + record_schema: type[RecordBase], + client_schema: type[ClientInfoBase], +): + @router.post(path) + def create_record( + client_info: Annotated[ + ClientInfoBase, + Query(), + Depends(client_schema), + ], + record: Annotated[ + RecordBase, + Body(), + Depends(record_schema), + ], + ): + print(f"processing client #{client_info.client_id} data {record.title!r}") + return { + "client_info": client_info.model_dump(), + "record": record.model_dump(), + } + + return create_record + + +register_record_route( + records_router, + "/case-files/", + CaseFileRecord, + RetailClientInfo, +) +register_record_route( + records_router, + "/contracts/", + ContractRecord, + PartnerClientInfo, +) +app.include_router( + records_router, + prefix="/clients", +) diff --git a/tests/_annotated_body_depends_merge_common.py b/tests/_annotated_body_depends_merge_common.py new file mode 100644 index 0000000000..73d16c749c --- /dev/null +++ b/tests/_annotated_body_depends_merge_common.py @@ -0,0 +1,42 @@ +from typing import Any + +from fastapi import UploadFile +from pydantic import BaseModel + + +class BasePayload(BaseModel): + kind: str + + +class FooPayload(BasePayload): + kind: str = "foo" + extra_foo: str + + +class BarPayload(BasePayload): + kind: str = "bar" + extra_bar: str + + +class FooFilePayload(BasePayload): + kind: str = "foo" + extra_foo: str + blob: UploadFile + + +class BarFilePayload(BasePayload): + kind: str = "bar" + extra_bar: str + blob: UploadFile + + +def openapi_request_body_schema_ref( + schema: dict[str, Any], + *, + path: str, + method: str = "post", + content_type: str, +) -> str: + return schema["paths"][path][method]["requestBody"]["content"][content_type][ + "schema" + ]["$ref"] diff --git a/tests/docs_src/__init__.py b/tests/docs_src/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/docs_src/_loader.py b/tests/docs_src/_loader.py new file mode 100644 index 0000000000..42e517671f --- /dev/null +++ b/tests/docs_src/_loader.py @@ -0,0 +1,24 @@ +"""Load `docs_src` tutorial modules by path for smoke tests.""" + +import importlib.util +import sys +from pathlib import Path + +from fastapi.testclient import TestClient + +_DOCS_SRC = Path(__file__).resolve().parent.parent.parent / "docs_src" + + +def load_docs_src_module(unique_name: str, *relative_parts: str): + path = _DOCS_SRC.joinpath(*relative_parts) + spec = importlib.util.spec_from_file_location(unique_name, path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + sys.modules[unique_name] = module + spec.loader.exec_module(module) + return module + + +def docs_src_test_client(unique_name: str, *relative_parts: str) -> TestClient: + mod = load_docs_src_module(unique_name, *relative_parts) + return TestClient(mod.app) diff --git a/tests/docs_src/body_depends_model_merge_body/test_body_depends_model_merge_tutorial001.py b/tests/docs_src/body_depends_model_merge_body/test_body_depends_model_merge_tutorial001.py new file mode 100644 index 0000000000..dd41ebc04d --- /dev/null +++ b/tests/docs_src/body_depends_model_merge_body/test_body_depends_model_merge_tutorial001.py @@ -0,0 +1,34 @@ +import pytest +from fastapi import status + +from tests.docs_src._loader import docs_src_test_client + + +@pytest.fixture(scope="module") +def client(): + return docs_src_test_client( + "docs_src_body_depends_model_merge_body_tutorial001", + "body_depends_model_merge_body", + "tutorial001_an_py310.py", + ) + + +@pytest.mark.parametrize( + ("path", "payload"), + [ + pytest.param( + "/items/objects/gadgets/", + {"name": "G1", "description": "d1"}, + id="gadget", + ), + pytest.param( + "/items/objects/parts/", + {"name": "P1", "sku": "S1"}, + id="part", + ), + ], +) +def test_json_body_depends_model_merge(client, path, payload): + response = client.post(path, json=payload) + assert response.status_code == status.HTTP_200_OK, response.text + assert response.json() == payload diff --git a/tests/docs_src/body_depends_model_merge_file/test_file_depends_model_merge_tutorial001.py b/tests/docs_src/body_depends_model_merge_file/test_file_depends_model_merge_tutorial001.py new file mode 100644 index 0000000000..0ab4712adc --- /dev/null +++ b/tests/docs_src/body_depends_model_merge_file/test_file_depends_model_merge_tutorial001.py @@ -0,0 +1,53 @@ +import pytest +from fastapi import status + +from tests.docs_src._loader import docs_src_test_client + + +@pytest.fixture(scope="module") +def client(): + return docs_src_test_client( + "docs_src_body_depends_model_merge_file_tutorial001", + "body_depends_model_merge_file", + "tutorial001_an_py310.py", + ) + + +@pytest.mark.parametrize( + ("path", "form_fields", "file_name", "file_bytes", "file_ct"), + [ + pytest.param( + "/files/attachments/commented/", + {"comment": "Spec"}, + "doc.pdf", + b"%PDF-1.4\n", + "application/pdf", + id="commented", + ), + pytest.param( + "/files/attachments/named/", + {"name": "My text file"}, + "file.txt", + b"hello", + "text/plain", + id="named", + ), + ], +) +def test_file_depends_model_merge( + client, + path, + form_fields, + file_name, + file_bytes, + file_ct, +): + upload = (file_name, file_bytes, file_ct) + response = client.post(path, data=form_fields, files={"file": upload}) + assert response.status_code == status.HTTP_200_OK, response.text + expected = { + "filename": file_name, + "content_type": file_ct, + "data": form_fields, + } + assert response.json() == expected diff --git a/tests/docs_src/body_depends_model_merge_form/test_form_depends_model_merge_tutorial001.py b/tests/docs_src/body_depends_model_merge_form/test_form_depends_model_merge_tutorial001.py new file mode 100644 index 0000000000..84454ccfe7 --- /dev/null +++ b/tests/docs_src/body_depends_model_merge_form/test_form_depends_model_merge_tutorial001.py @@ -0,0 +1,34 @@ +import pytest +from fastapi import status + +from tests.docs_src._loader import docs_src_test_client + + +@pytest.fixture(scope="module") +def client(): + return docs_src_test_client( + "docs_src_body_depends_model_merge_form_tutorial001", + "body_depends_model_merge_form", + "tutorial001_an_py310.py", + ) + + +@pytest.mark.parametrize( + ("path", "form_data"), + [ + pytest.param( + "/auth/session/password/", + {"username": "alice", "password": "secret"}, + id="password", + ), + pytest.param( + "/auth/session/token/", + {"username": "bob", "token": "tok-123"}, + id="token", + ), + ], +) +def test_form_depends_model_merge(client, path, form_data): + response = client.post(path, data=form_data) + assert response.status_code == status.HTTP_200_OK, response.text + assert response.json() == form_data diff --git a/tests/docs_src/body_depends_model_merge_query/test_query_depends_model_merge_tutorial001.py b/tests/docs_src/body_depends_model_merge_query/test_query_depends_model_merge_tutorial001.py new file mode 100644 index 0000000000..26be99bec6 --- /dev/null +++ b/tests/docs_src/body_depends_model_merge_query/test_query_depends_model_merge_tutorial001.py @@ -0,0 +1,42 @@ +import pytest +from fastapi import status + +from tests.docs_src._loader import docs_src_test_client + + +@pytest.fixture(scope="module") +def client(): + return docs_src_test_client( + "docs_src_body_depends_model_merge_query_tutorial001", + "body_depends_model_merge_query", + "tutorial001_an_py310.py", + ) + + +_CATEGORY = "books" + + +@pytest.mark.parametrize( + ("path", "params"), + [ + pytest.param( + "/catalog/items/", + {"category": _CATEGORY, "in_stock": False}, + id="items_full", + ), + pytest.param( + "/catalog/basics/", + {"category": _CATEGORY}, + id="basics", + ), + pytest.param( + "/catalog/items-paginated/", + {"category": _CATEGORY, "page": 2, "per_page": 5}, + id="paginated", + ), + ], +) +def test_query_depends_model_merge(client, path, params): + response = client.get(path, params=params) + assert response.status_code == status.HTTP_200_OK, response.text + assert response.json() == params diff --git a/tests/docs_src/body_depends_model_merge_query_plus_body/test_query_plus_body_tutorial001.py b/tests/docs_src/body_depends_model_merge_query_plus_body/test_query_plus_body_tutorial001.py new file mode 100644 index 0000000000..21d1f39993 --- /dev/null +++ b/tests/docs_src/body_depends_model_merge_query_plus_body/test_query_plus_body_tutorial001.py @@ -0,0 +1,39 @@ +import pytest +from fastapi import status + +from tests.docs_src._loader import docs_src_test_client + + +@pytest.fixture(scope="module") +def client(): + return docs_src_test_client( + "docs_src_body_depends_model_merge_query_plus_body_tutorial001", + "body_depends_model_merge_query_plus_body", + "tutorial001_an_py310.py", + ) + + +@pytest.mark.parametrize( + ("path", "query", "json_body"), + [ + ( + "/clients/case-files/", + {"client_id": "rick", "region": "west"}, + {"title": "Q1", "case_number": "C-9"}, + ), + ( + "/clients/contracts/", + {"client_id": "morty", "contract_ref": "R-9"}, + {"title": "Partner deal", "contract_id": "Z-1"}, + ), + ], + ids=["case_file", "contract"], +) +def test_query_plus_merged_json_body(client, path, query, json_body): + response = client.post(path, params=query, json=json_body) + assert response.status_code == status.HTTP_200_OK, response.text + expected = { + "client_info": query, + "record": json_body, + } + assert response.json() == expected diff --git a/tests/test_annotated_body_depends_merge_body.py b/tests/test_annotated_body_depends_merge_body.py new file mode 100644 index 0000000000..9530cd5aa8 --- /dev/null +++ b/tests/test_annotated_body_depends_merge_body.py @@ -0,0 +1,161 @@ +from typing import Annotated, Any + +import pytest +from fastapi import Body, Depends, FastAPI, Form, status +from fastapi.exceptions import FastAPIError +from fastapi.testclient import TestClient + +from tests._annotated_body_depends_merge_common import ( + BarPayload, + BasePayload, + FooPayload, + openapi_request_body_schema_ref, +) + + +class TestAnnotatedBodyDependsMergeBody: + @pytest.mark.parametrize( + ( + "path", + "ann1", + "ann2", + "model_cls", + "expected_ref_suffix", + "assert_no_query_params", + ), + [ + ("/a", Body(), Depends(FooPayload), FooPayload, "FooPayload", True), + ("/b", Depends(BarPayload), Body(), BarPayload, "BarPayload", False), + ], + ) + def test_openapi_json_body_depends_merge( + self, + path: str, + ann1: Any, + ann2: Any, + model_cls: type[BasePayload], + expected_ref_suffix: str, + assert_no_query_params: bool, + ) -> None: + app = FastAPI() + + @app.post(path) + def route( + data: Annotated[BasePayload, ann1, ann2], + ) -> None: + assert isinstance(data, model_cls) + + client = TestClient(app) + schema = client.get("/openapi.json").json() + post = schema["paths"][path]["post"] + if assert_no_query_params: + assert post.get("parameters") in (None, []) + ref = openapi_request_body_schema_ref( + schema, path=path, method="post", content_type="application/json" + ) + assert ref.endswith(f"/{expected_ref_suffix}") + + def test_runtime_json_validates_concrete_model(self) -> None: + app = FastAPI() + + @app.post("/c") + def route( + data: Annotated[BasePayload, Body(), Depends(FooPayload)], + ) -> dict[str, str]: + return {"extra": data.extra_foo} + + client = TestClient(app) + r = client.post("/c", json={"kind": "foo", "extra_foo": "x"}) + assert r.status_code == status.HTTP_200_OK + assert r.json() == {"extra": "x"} + + bad = client.post("/c", json={"kind": "foo"}) + assert bad.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT + + def test_put_patch_json_body_depends_openapi(self) -> None: + app = FastAPI() + path = "/items/{item_id}" + + @app.put(path) + def route_put( + item_id: str, + data: Annotated[BasePayload, Body(), Depends(FooPayload)], + ) -> None: + assert isinstance(data, FooPayload) + + @app.patch(path) + def route_patch( + item_id: str, + data: Annotated[BasePayload, Depends(FooPayload), Body()], + ) -> None: + assert isinstance(data, FooPayload) + + client = TestClient(app) + schema = client.get("/openapi.json").json() + for method in ("put", "patch"): + ref = openapi_request_body_schema_ref( + schema, path=path, method=method, content_type="application/json" + ) + assert ref.endswith("/FooPayload") + + r = client.put("/items/1", json={"kind": "foo", "extra_foo": "a"}) + assert r.status_code == status.HTTP_200_OK + r2 = client.patch("/items/1", json={"kind": "foo", "extra_foo": "b"}) + assert r2.status_code == status.HTTP_200_OK + + def test_rejects_body_with_callable_depends(self) -> None: + app = FastAPI() + + def not_a_model() -> None: + return None + + with pytest.raises(FastAPIError, match="Pydantic model class"): + + @app.post("/d") + def route_d( + data: Annotated[BasePayload, Body(), Depends(not_a_model)], + ) -> None: + pass # pragma: no cover + + def test_rejects_multiple_depends_with_body(self) -> None: + app = FastAPI() + + with pytest.raises(FastAPIError, match="multiple `Depends`"): + + @app.post("/e") + def route_e( + data: Annotated[ + BasePayload, + Body(), + Depends(FooPayload), + Depends(BarPayload), + ], + ) -> None: + pass # pragma: no cover + + def test_rejects_body_and_form_together(self) -> None: + app = FastAPI() + + with pytest.raises(FastAPIError, match="multiple `Body`"): + + @app.post("/conflict") + def route_conflict( + data: Annotated[ + BasePayload, + Body(), + Form(), + Depends(FooPayload), + ], + ) -> None: + pass # pragma: no cover + + def test_rejects_merge_on_path_parameter(self) -> None: + app = FastAPI() + + with pytest.raises(FastAPIError, match="path parameter"): + + @app.post("/path/{data}") + def route_path( + data: Annotated[BasePayload, Body(), Depends(FooPayload)], + ) -> None: + pass # pragma: no cover diff --git a/tests/test_annotated_body_depends_merge_file.py b/tests/test_annotated_body_depends_merge_file.py new file mode 100644 index 0000000000..c1ebdb19d1 --- /dev/null +++ b/tests/test_annotated_body_depends_merge_file.py @@ -0,0 +1,100 @@ +from io import BytesIO +from typing import Annotated, Any + +import pytest +from fastapi import Depends, FastAPI, File, Form, status +from fastapi.exceptions import FastAPIError +from fastapi.testclient import TestClient + +from tests._annotated_body_depends_merge_common import ( + BarFilePayload, + BasePayload, + FooFilePayload, + FooPayload, +) + + +class TestAnnotatedBodyDependsMergeFile: + @pytest.mark.parametrize( + ("path", "ann1", "ann2", "model_cls", "expected_ref_suffix"), + [ + ( + "/file-a", + File(), + Depends(FooFilePayload), + FooFilePayload, + "FooFilePayload", + ), + ( + "/file-b", + Depends(BarFilePayload), + File(), + BarFilePayload, + "BarFilePayload", + ), + ], + ) + def test_openapi_file_depends_merge( + self, + path: str, + ann1: Any, + ann2: Any, + model_cls: type[BasePayload], + expected_ref_suffix: str, + ) -> None: + app = FastAPI() + + @app.post(path) + def route_file( + data: Annotated[BasePayload, ann1, ann2], + ) -> None: + assert isinstance(data, model_cls) + + client = TestClient(app) + schema = client.get("/openapi.json").json() + rb = schema["paths"][path]["post"]["requestBody"] + content = rb["content"] + assert "multipart/form-data" in content + ref = content["multipart/form-data"]["schema"]["$ref"] + assert ref.endswith(f"/{expected_ref_suffix}") + + def test_runtime_file_validates_concrete_model(self) -> None: + app = FastAPI() + + @app.post("/file-c") + def route_file( + data: Annotated[BasePayload, File(), Depends(FooFilePayload)], + ) -> dict[str, str]: + return {"extra": data.extra_foo, "fn": data.blob.filename or ""} + + client = TestClient(app) + r = client.post( + "/file-c", + data={"kind": "foo", "extra_foo": "u"}, + files={"blob": ("up.txt", BytesIO(b"xyz"), "text/plain")}, + ) + assert r.status_code == status.HTTP_200_OK + assert r.json() == {"extra": "u", "fn": "up.txt"} + + bad = client.post( + "/file-c", + data={"kind": "foo"}, + files={"blob": ("up.txt", BytesIO(b"x"), "text/plain")}, + ) + assert bad.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT + + def test_rejects_file_and_form_together(self) -> None: + app = FastAPI() + + with pytest.raises(FastAPIError, match="multiple `Body`"): + + @app.post("/file-conflict") + def route_conflict( + data: Annotated[ + BasePayload, + File(), + Form(), + Depends(FooPayload), + ], + ) -> None: + pass # pragma: no cover diff --git a/tests/test_annotated_body_depends_merge_form.py b/tests/test_annotated_body_depends_merge_form.py new file mode 100644 index 0000000000..8a67d30f3a --- /dev/null +++ b/tests/test_annotated_body_depends_merge_form.py @@ -0,0 +1,64 @@ +from typing import Annotated, Any + +import pytest +from fastapi import Depends, FastAPI, Form, status +from fastapi.testclient import TestClient + +from tests._annotated_body_depends_merge_common import ( + BarPayload, + BasePayload, + FooPayload, +) + + +class TestAnnotatedBodyDependsMergeForm: + @pytest.mark.parametrize( + ("path", "ann1", "ann2", "model_cls", "expected_ref_suffix"), + [ + ("/form-a", Form(), Depends(FooPayload), FooPayload, "FooPayload"), + ("/form-b", Depends(BarPayload), Form(), BarPayload, "BarPayload"), + ], + ) + def test_openapi_form_depends_merge( + self, + path: str, + ann1: Any, + ann2: Any, + model_cls: type[BasePayload], + expected_ref_suffix: str, + ) -> None: + app = FastAPI() + + @app.post(path) + def route_form( + data: Annotated[BasePayload, ann1, ann2], + ) -> None: + assert isinstance(data, model_cls) + + client = TestClient(app) + schema = client.get("/openapi.json").json() + rb = schema["paths"][path]["post"]["requestBody"] + content = rb["content"] + assert "application/x-www-form-urlencoded" in content + ref = content["application/x-www-form-urlencoded"]["schema"]["$ref"] + assert ref.endswith(f"/{expected_ref_suffix}") + + def test_runtime_form_validates_concrete_model(self) -> None: + app = FastAPI() + + @app.post("/form-c") + def route_form( + data: Annotated[BasePayload, Form(), Depends(FooPayload)], + ) -> dict[str, str]: + return {"extra": data.extra_foo} + + client = TestClient(app) + r = client.post( + "/form-c", + data={"kind": "foo", "extra_foo": "z"}, + ) + assert r.status_code == status.HTTP_200_OK + assert r.json() == {"extra": "z"} + + bad = client.post("/form-c", data={"kind": "foo"}) + assert bad.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT diff --git a/tests/test_annotated_body_depends_merge_query_plus_shape.py b/tests/test_annotated_body_depends_merge_query_plus_shape.py new file mode 100644 index 0000000000..024df417c1 --- /dev/null +++ b/tests/test_annotated_body_depends_merge_query_plus_shape.py @@ -0,0 +1,144 @@ +"""Query() alongside a second parameter that uses Body/Form/File + Depends(model).""" + +from io import BytesIO +from typing import Annotated, Any + +from fastapi import Body, Depends, FastAPI, File, Form, Query +from fastapi.testclient import TestClient + +from tests._annotated_body_depends_merge_common import ( + BarFilePayload, + BarPayload, + BasePayload, + FooFilePayload, + FooPayload, + openapi_request_body_schema_ref, +) + + +def _param_names(post_schema: dict[str, Any]) -> list[str]: + params = post_schema.get("parameters") or [] + return [p["name"] for p in params] + + +class TestQueryPlusMergedShape: + def test_openapi_query_plus_json_body(self) -> None: + app = FastAPI() + + @app.post("/mix-json") + def route( + client_id: Annotated[str, Query()], + data: Annotated[BasePayload, Body(), Depends(FooPayload)], + ) -> None: + assert isinstance(data, FooPayload) + + client = TestClient(app) + schema = client.get("/openapi.json").json() + post = schema["paths"]["/mix-json"]["post"] + assert "client_id" in _param_names(post) + ref = openapi_request_body_schema_ref( + schema, + path="/mix-json", + method="post", + content_type="application/json", + ) + assert ref.endswith("/FooPayload") + + def test_runtime_query_plus_json_body(self) -> None: + app = FastAPI() + + @app.post("/r-json") + def route( + client_id: Annotated[str, Query()], + data: Annotated[BasePayload, Body(), Depends(FooPayload)], + ) -> dict[str, str]: + return {"client": client_id, "extra": data.extra_foo} + + client = TestClient(app) + r = client.post( + "/r-json?client_id=c1", + json={"kind": "foo", "extra_foo": "e"}, + ) + assert r.status_code == 200 + assert r.json() == {"client": "c1", "extra": "e"} + + def test_openapi_query_plus_form(self) -> None: + app = FastAPI() + + @app.post("/mix-form") + def route( + client_id: Annotated[str, Query()], + data: Annotated[BasePayload, Form(), Depends(BarPayload)], + ) -> None: + assert isinstance(data, BarPayload) + + client = TestClient(app) + schema = client.get("/openapi.json").json() + post = schema["paths"]["/mix-form"]["post"] + assert "client_id" in _param_names(post) + rb = post["requestBody"]["content"] + assert "application/x-www-form-urlencoded" in rb + ref = rb["application/x-www-form-urlencoded"]["schema"]["$ref"] + assert ref.endswith("/BarPayload") + + def test_runtime_query_plus_form(self) -> None: + app = FastAPI() + + @app.post("/r-form") + def route( + client_id: Annotated[str, Query()], + data: Annotated[BasePayload, Form(), Depends(FooPayload)], + ) -> dict[str, str]: + return {"client": client_id, "extra": data.extra_foo} + + client = TestClient(app) + r = client.post( + "/r-form", + params={"client_id": "c2"}, + data={"kind": "foo", "extra_foo": "f"}, + ) + assert r.status_code == 200 + assert r.json() == {"client": "c2", "extra": "f"} + + def test_openapi_query_plus_file_multipart(self) -> None: + app = FastAPI() + + @app.post("/mix-file") + def route( + client_id: Annotated[str, Query()], + data: Annotated[BasePayload, File(), Depends(FooFilePayload)], + ) -> None: + assert isinstance(data, FooFilePayload) + + client = TestClient(app) + schema = client.get("/openapi.json").json() + post = schema["paths"]["/mix-file"]["post"] + assert "client_id" in _param_names(post) + rb = post["requestBody"]["content"] + assert "multipart/form-data" in rb + ref = rb["multipart/form-data"]["schema"]["$ref"] + assert ref.endswith("/FooFilePayload") + + def test_runtime_query_plus_file_multipart(self) -> None: + app = FastAPI() + + @app.post("/r-file") + def route( + client_id: Annotated[str, Query()], + data: Annotated[BasePayload, File(), Depends(BarFilePayload)], + ) -> dict[str, str]: + return { + "client": client_id, + "extra": data.extra_bar, + "fn": data.blob.filename or "", + } + + client = TestClient(app) + r = client.post( + "/r-file", + params={"client_id": "c3"}, + data={"kind": "bar", "extra_bar": "b"}, + files={"blob": ("up.bin", BytesIO(b"abc"), "application/octet-stream")}, + ) + assert r.status_code == 200 + assert r.json() == {"client": "c3", "extra": "b", "fn": "up.bin"}