Browse Source

test: more coverage and fixes

pull/15477/head
Suren Khorenyan 2 months ago
parent
commit
49053271ae
  1. 64
      tests/test_annotated_body_depends_merge_body.py
  2. 25
      tests/test_annotated_body_depends_merge_file.py
  3. 34
      tests/test_annotated_body_depends_merge_form.py
  4. 65
      tests/test_annotated_body_depends_merge_query_plus_shape.py

64
tests/test_annotated_body_depends_merge_body.py

@ -36,10 +36,27 @@ class TestAnnotatedBodyDependsMergeBody:
"model_cls", "model_cls",
"expected_ref_suffix", "expected_ref_suffix",
"assert_no_query_params", "assert_no_query_params",
"payload",
), ),
[ [
("/a", Body(), Depends(FooPayload), FooPayload, "FooPayload", True), pytest.param(
("/b", Depends(BarPayload), Body(), BarPayload, "BarPayload", False), "/a",
Body(),
Depends(FooPayload),
FooPayload,
"FooPayload",
True,
{"kind": "foo", "extra_foo": "hit"},
),
pytest.param(
"/b",
Depends(BarPayload),
Body(),
BarPayload,
"BarPayload",
False,
{"kind": "bar", "extra_bar": "hit"},
),
], ],
) )
def test_openapi_json_body_depends_merge( def test_openapi_json_body_depends_merge(
@ -50,6 +67,7 @@ class TestAnnotatedBodyDependsMergeBody:
model_cls: type[BasePayload], model_cls: type[BasePayload],
expected_ref_suffix: str, expected_ref_suffix: str,
assert_no_query_params: bool, assert_no_query_params: bool,
payload: dict[str, str],
) -> None: ) -> None:
app = FastAPI() app = FastAPI()
@ -69,6 +87,9 @@ class TestAnnotatedBodyDependsMergeBody:
) )
assert ref.endswith(f"/{expected_ref_suffix}") assert ref.endswith(f"/{expected_ref_suffix}")
ok = client.post(path, json=payload)
assert ok.status_code == status.HTTP_200_OK
def test_runtime_json_validates_concrete_model(self) -> None: def test_runtime_json_validates_concrete_model(self) -> None:
app = FastAPI() app = FastAPI()
@ -79,9 +100,12 @@ class TestAnnotatedBodyDependsMergeBody:
return {"extra": data.extra_foo} return {"extra": data.extra_foo}
client = TestClient(app) client = TestClient(app)
r = client.post("/c", json={"kind": "foo", "extra_foo": "x"}) extra = "ricksanchez"
payload = {"kind": "foo", "extra_foo": extra}
expected_json = {"extra": extra}
r = client.post("/c", json=payload)
assert r.status_code == status.HTTP_200_OK assert r.status_code == status.HTTP_200_OK
assert r.json() == {"extra": "x"} assert r.json() == expected_json
bad = client.post("/c", json={"kind": "foo"}) bad = client.post("/c", json={"kind": "foo"})
# not status.*: Starlette confused HTTP_422_UNPROCESSABLE_CONTENT HTTP_422_UNPROCESSABLE_ENTITY # not status.*: Starlette confused HTTP_422_UNPROCESSABLE_CONTENT HTTP_422_UNPROCESSABLE_ENTITY
@ -97,9 +121,12 @@ class TestAnnotatedBodyDependsMergeBody:
return {"extra": data.extra_foo} return {"extra": data.extra_foo}
client = TestClient(app) client = TestClient(app)
r = client.post("/depends-empty", json={"kind": "foo", "extra_foo": "z"}) extra = "foobar"
payload = {"kind": "foo", "extra_foo": extra}
expected_json = {"extra": extra}
r = client.post("/depends-empty", json=payload)
assert r.status_code == status.HTTP_200_OK assert r.status_code == status.HTTP_200_OK
assert r.json() == {"extra": "z"} assert r.json() == expected_json
def test_merged_body_parameter_signature_default(self) -> None: def test_merged_body_parameter_signature_default(self) -> None:
app = FastAPI() app = FastAPI()
@ -115,13 +142,15 @@ class TestAnnotatedBodyDependsMergeBody:
client = TestClient(app) client = TestClient(app)
empty = client.post("/merged-default") empty = client.post("/merged-default")
assert empty.status_code == status.HTTP_200_OK assert empty.status_code == status.HTTP_200_OK
assert empty.json() == {"extra": "default"} expected_default = {"extra": "default"}
assert empty.json() == expected_default
overridden = client.post( override_extra = "hello world"
"/merged-default", json={"kind": "foo", "extra_foo": "ov"} override_payload = {"kind": "foo", "extra_foo": override_extra}
) expected_override = {"extra": override_extra}
overridden = client.post("/merged-default", json=override_payload)
assert overridden.status_code == status.HTTP_200_OK assert overridden.status_code == status.HTTP_200_OK
assert overridden.json() == {"extra": "ov"} assert overridden.json() == expected_override
def test_put_patch_json_body_depends_openapi(self) -> None: def test_put_patch_json_body_depends_openapi(self) -> None:
app = FastAPI() app = FastAPI()
@ -149,16 +178,20 @@ class TestAnnotatedBodyDependsMergeBody:
) )
assert ref.endswith("/FooPayload") assert ref.endswith("/FooPayload")
r = client.put("/items/1", json={"kind": "foo", "extra_foo": "a"}) put_extra = "fizz"
put_payload = {"kind": "foo", "extra_foo": put_extra}
r = client.put("/items/1", json=put_payload)
assert r.status_code == status.HTTP_200_OK assert r.status_code == status.HTTP_200_OK
r2 = client.patch("/items/1", json={"kind": "foo", "extra_foo": "b"}) patch_extra = "buzz"
patch_payload = {"kind": "foo", "extra_foo": patch_extra}
r2 = client.patch("/items/1", json=patch_payload)
assert r2.status_code == status.HTTP_200_OK assert r2.status_code == status.HTTP_200_OK
def test_rejects_body_with_callable_depends(self) -> None: def test_rejects_body_with_callable_depends(self) -> None:
app = FastAPI() app = FastAPI()
def not_a_model() -> None: def not_a_model() -> None:
return None pass # pragma: no cover
with pytest.raises(FastAPIError, match="Pydantic model class"): with pytest.raises(FastAPIError, match="Pydantic model class"):
@ -278,5 +311,4 @@ class TestAnnotatedBodyDependsMergeBody:
@app.post("/path/{data}") @app.post("/path/{data}")
def route_path( def route_path(
data: Annotated[BasePayload, Body(), Depends(FooPayload)], data: Annotated[BasePayload, Body(), Depends(FooPayload)],
) -> None: ) -> None: ...
pass # pragma: no cover

25
tests/test_annotated_body_depends_merge_file.py

@ -17,7 +17,7 @@ from tests._annotated_body_depends_merge_common import (
class TestAnnotatedBodyDependsMergeFile: class TestAnnotatedBodyDependsMergeFile:
@pytest.mark.parametrize( @pytest.mark.parametrize(
("path", "ann1", "ann2", "model_cls", "expected_ref_suffix"), ("path", "ann1", "ann2", "model_cls", "expected_ref_suffix", "payload"),
[ [
( (
"/file-a", "/file-a",
@ -25,6 +25,7 @@ class TestAnnotatedBodyDependsMergeFile:
Depends(FooFilePayload), Depends(FooFilePayload),
FooFilePayload, FooFilePayload,
"FooFilePayload", "FooFilePayload",
{"kind": "foo", "extra_foo": "hit"},
), ),
( (
"/file-b", "/file-b",
@ -32,6 +33,7 @@ class TestAnnotatedBodyDependsMergeFile:
File(), File(),
BarFilePayload, BarFilePayload,
"BarFilePayload", "BarFilePayload",
{"kind": "bar", "extra_bar": "hit"},
), ),
], ],
) )
@ -42,6 +44,7 @@ class TestAnnotatedBodyDependsMergeFile:
ann2: Any, ann2: Any,
model_cls: type[BasePayload], model_cls: type[BasePayload],
expected_ref_suffix: str, expected_ref_suffix: str,
payload: dict[str, str],
) -> None: ) -> None:
app = FastAPI() app = FastAPI()
@ -59,6 +62,10 @@ class TestAnnotatedBodyDependsMergeFile:
ref = content["multipart/form-data"]["schema"]["$ref"] ref = content["multipart/form-data"]["schema"]["$ref"]
assert ref.endswith(f"/{expected_ref_suffix}") assert ref.endswith(f"/{expected_ref_suffix}")
blob = ("blob.bin", BytesIO(b"x"), "application/octet-stream")
ok = client.post(path, data=payload, files={"blob": blob})
assert ok.status_code == status.HTTP_200_OK
def test_runtime_file_validates_concrete_model(self) -> None: def test_runtime_file_validates_concrete_model(self) -> None:
app = FastAPI() app = FastAPI()
@ -69,13 +76,14 @@ class TestAnnotatedBodyDependsMergeFile:
return {"extra": data.extra_foo, "fn": data.blob.filename or ""} return {"extra": data.extra_foo, "fn": data.blob.filename or ""}
client = TestClient(app) client = TestClient(app)
r = client.post( extra = "info"
"/file-c", filename = "file.txt"
data={"kind": "foo", "extra_foo": "u"}, payload = {"kind": "foo", "extra_foo": extra}
files={"blob": ("up.txt", BytesIO(b"xyz"), "text/plain")}, file_field = (filename, BytesIO(b"xyz"), "text/plain")
) expected_json = {"extra": extra, "fn": filename}
r = client.post("/file-c", data=payload, files={"blob": file_field})
assert r.status_code == status.HTTP_200_OK assert r.status_code == status.HTTP_200_OK
assert r.json() == {"extra": "u", "fn": "up.txt"} assert r.json() == expected_json
bad = client.post( bad = client.post(
"/file-c", "/file-c",
@ -98,5 +106,4 @@ class TestAnnotatedBodyDependsMergeFile:
Form(), Form(),
Depends(FooPayload), Depends(FooPayload),
], ],
) -> None: ) -> None: ...
pass # pragma: no cover

34
tests/test_annotated_body_depends_merge_form.py

@ -14,10 +14,24 @@ from tests._annotated_body_depends_merge_common import (
class TestAnnotatedBodyDependsMergeForm: class TestAnnotatedBodyDependsMergeForm:
@pytest.mark.parametrize( @pytest.mark.parametrize(
("path", "ann1", "ann2", "model_cls", "expected_ref_suffix"), ("path", "ann1", "ann2", "model_cls", "expected_ref_suffix", "payload"),
[ [
("/form-a", Form(), Depends(FooPayload), FooPayload, "FooPayload"), (
("/form-b", Depends(BarPayload), Form(), BarPayload, "BarPayload"), "/form-a",
Form(),
Depends(FooPayload),
FooPayload,
"FooPayload",
{"kind": "foo", "extra_foo": "hit"},
),
(
"/form-b",
Depends(BarPayload),
Form(),
BarPayload,
"BarPayload",
{"kind": "bar", "extra_bar": "hit"},
),
], ],
) )
def test_openapi_form_depends_merge( def test_openapi_form_depends_merge(
@ -27,6 +41,7 @@ class TestAnnotatedBodyDependsMergeForm:
ann2: Any, ann2: Any,
model_cls: type[BasePayload], model_cls: type[BasePayload],
expected_ref_suffix: str, expected_ref_suffix: str,
payload: dict[str, str],
) -> None: ) -> None:
app = FastAPI() app = FastAPI()
@ -44,6 +59,9 @@ class TestAnnotatedBodyDependsMergeForm:
ref = content["application/x-www-form-urlencoded"]["schema"]["$ref"] ref = content["application/x-www-form-urlencoded"]["schema"]["$ref"]
assert ref.endswith(f"/{expected_ref_suffix}") assert ref.endswith(f"/{expected_ref_suffix}")
ok = client.post(path, data=payload)
assert ok.status_code == status.HTTP_200_OK
def test_runtime_form_validates_concrete_model(self) -> None: def test_runtime_form_validates_concrete_model(self) -> None:
app = FastAPI() app = FastAPI()
@ -54,12 +72,12 @@ class TestAnnotatedBodyDependsMergeForm:
return {"extra": data.extra_foo} return {"extra": data.extra_foo}
client = TestClient(app) client = TestClient(app)
r = client.post( extra = "z"
"/form-c", payload = {"kind": "foo", "extra_foo": extra}
data={"kind": "foo", "extra_foo": "z"}, expected_json = {"extra": extra}
) r = client.post("/form-c", data=payload)
assert r.status_code == status.HTTP_200_OK assert r.status_code == status.HTTP_200_OK
assert r.json() == {"extra": "z"} assert r.json() == expected_json
bad = client.post("/form-c", data={"kind": "foo"}) bad = client.post("/form-c", data={"kind": "foo"})
# not status.*: Starlette confused HTTP_422_UNPROCESSABLE_CONTENT HTTP_422_UNPROCESSABLE_ENTITY # not status.*: Starlette confused HTTP_422_UNPROCESSABLE_CONTENT HTTP_422_UNPROCESSABLE_ENTITY

65
tests/test_annotated_body_depends_merge_query_plus_shape.py

@ -3,7 +3,7 @@
from io import BytesIO from io import BytesIO
from typing import Annotated, Any from typing import Annotated, Any
from fastapi import Body, Depends, FastAPI, File, Form, Query from fastapi import Body, Depends, FastAPI, File, Form, Query, status
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from tests._annotated_body_depends_merge_common import ( from tests._annotated_body_depends_merge_common import (
@ -44,6 +44,10 @@ class TestQueryPlusMergedShape:
) )
assert ref.endswith("/FooPayload") assert ref.endswith("/FooPayload")
hit_payload = {"kind": "foo", "extra_foo": "x"}
hit = client.post("/mix-json?client_id=hit", json=hit_payload)
assert hit.status_code == status.HTTP_200_OK
def test_runtime_query_plus_json_body(self) -> None: def test_runtime_query_plus_json_body(self) -> None:
app = FastAPI() app = FastAPI()
@ -55,12 +59,13 @@ class TestQueryPlusMergedShape:
return {"client": client_id, "extra": data.extra_foo} return {"client": client_id, "extra": data.extra_foo}
client = TestClient(app) client = TestClient(app)
r = client.post( client_id = "c1"
"/r-json?client_id=c1", extra = "helloworld"
json={"kind": "foo", "extra_foo": "e"}, payload = {"kind": "foo", "extra_foo": extra}
) expected_json = {"client": client_id, "extra": extra}
assert r.status_code == 200 r = client.post(f"/r-json?client_id={client_id}", json=payload)
assert r.json() == {"client": "c1", "extra": "e"} assert r.status_code == status.HTTP_200_OK
assert r.json() == expected_json
def test_openapi_query_plus_form(self) -> None: def test_openapi_query_plus_form(self) -> None:
app = FastAPI() app = FastAPI()
@ -81,6 +86,10 @@ class TestQueryPlusMergedShape:
ref = rb["application/x-www-form-urlencoded"]["schema"]["$ref"] ref = rb["application/x-www-form-urlencoded"]["schema"]["$ref"]
assert ref.endswith("/BarPayload") assert ref.endswith("/BarPayload")
hit_payload = {"kind": "bar", "extra_bar": "y"}
hit = client.post("/mix-form", params={"client_id": "hit"}, data=hit_payload)
assert hit.status_code == status.HTTP_200_OK
def test_runtime_query_plus_form(self) -> None: def test_runtime_query_plus_form(self) -> None:
app = FastAPI() app = FastAPI()
@ -92,13 +101,13 @@ class TestQueryPlusMergedShape:
return {"client": client_id, "extra": data.extra_foo} return {"client": client_id, "extra": data.extra_foo}
client = TestClient(app) client = TestClient(app)
r = client.post( client_id = "c2"
"/r-form", extra = "foo"
params={"client_id": "c2"}, payload = {"kind": "foo", "extra_foo": extra}
data={"kind": "foo", "extra_foo": "f"}, expected_json = {"client": client_id, "extra": extra}
) r = client.post("/r-form", params={"client_id": client_id}, data=payload)
assert r.status_code == 200 assert r.status_code == status.HTTP_200_OK
assert r.json() == {"client": "c2", "extra": "f"} assert r.json() == expected_json
def test_openapi_query_plus_file_multipart(self) -> None: def test_openapi_query_plus_file_multipart(self) -> None:
app = FastAPI() app = FastAPI()
@ -119,6 +128,18 @@ class TestQueryPlusMergedShape:
ref = rb["multipart/form-data"]["schema"]["$ref"] ref = rb["multipart/form-data"]["schema"]["$ref"]
assert ref.endswith("/FooFilePayload") assert ref.endswith("/FooFilePayload")
hit_form = {"kind": "foo", "extra_foo": "u"}
hit_files = {
"blob": ("up.bin", BytesIO(b"abc"), "application/octet-stream"),
}
hit = client.post(
"/mix-file",
params={"client_id": "hit"},
data=hit_form,
files=hit_files,
)
assert hit.status_code == status.HTTP_200_OK
def test_runtime_query_plus_file_multipart(self) -> None: def test_runtime_query_plus_file_multipart(self) -> None:
app = FastAPI() app = FastAPI()
@ -134,11 +155,17 @@ class TestQueryPlusMergedShape:
} }
client = TestClient(app) client = TestClient(app)
client_id = "c3"
extra = "bar"
filename = "up.bin"
payload = {"kind": "bar", "extra_bar": extra}
file_field = (filename, BytesIO(b"abc"), "application/octet-stream")
expected_json = {"client": client_id, "extra": extra, "fn": filename}
r = client.post( r = client.post(
"/r-file", "/r-file",
params={"client_id": "c3"}, params={"client_id": client_id},
data={"kind": "bar", "extra_bar": "b"}, data=payload,
files={"blob": ("up.bin", BytesIO(b"abc"), "application/octet-stream")}, files={"blob": file_field},
) )
assert r.status_code == 200 assert r.status_code == status.HTTP_200_OK
assert r.json() == {"client": "c3", "extra": "b", "fn": "up.bin"} assert r.json() == expected_json

Loading…
Cancel
Save