Motov Yurii
5 months ago
committed by
GitHub
GitHub
12 changed files with 329 additions and 14 deletions
@ -0,0 +1,27 @@ |
|||||
|
from typing import Annotated |
||||
|
|
||||
|
from fastapi import Depends, FastAPI |
||||
|
from fastapi.testclient import TestClient |
||||
|
from typing_extensions import TypeAliasType |
||||
|
|
||||
|
|
||||
|
async def some_value() -> int: |
||||
|
return 123 |
||||
|
|
||||
|
|
||||
|
DependedValue = TypeAliasType( |
||||
|
"DependedValue", Annotated[int, Depends(some_value)], type_params=() |
||||
|
) |
||||
|
|
||||
|
|
||||
|
def test_pep695_type_dependencies(): |
||||
|
app = FastAPI() |
||||
|
|
||||
|
@app.get("/") |
||||
|
async def get_with_dep(value: DependedValue) -> str: # noqa |
||||
|
return f"value: {value}" |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
response = client.get("/") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.text == '"value: 123"' |
||||
@ -0,0 +1,63 @@ |
|||||
|
import json |
||||
|
from typing import Annotated |
||||
|
|
||||
|
from fastapi import Cookie, FastAPI, Form, Header, Query |
||||
|
from fastapi.testclient import TestClient |
||||
|
from pydantic import Json |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.post("/form-json-list") |
||||
|
def form_json_list(items: Annotated[Json[list[str]], Form()]) -> list[str]: |
||||
|
return items |
||||
|
|
||||
|
|
||||
|
@app.get("/query-json-list") |
||||
|
def query_json_list(items: Annotated[Json[list[str]], Query()]) -> list[str]: |
||||
|
return items |
||||
|
|
||||
|
|
||||
|
@app.get("/header-json-list") |
||||
|
def header_json_list(x_items: Annotated[Json[list[str]], Header()]) -> list[str]: |
||||
|
return x_items |
||||
|
|
||||
|
|
||||
|
@app.get("/cookie-json-list") |
||||
|
def cookie_json_list(items: Annotated[Json[list[str]], Cookie()]) -> list[str]: |
||||
|
return items |
||||
|
|
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_form_json_list(): |
||||
|
response = client.post( |
||||
|
"/form-json-list", data={"items": json.dumps(["abc", "def"])} |
||||
|
) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == ["abc", "def"] |
||||
|
|
||||
|
|
||||
|
def test_query_json_list(): |
||||
|
response = client.get( |
||||
|
"/query-json-list", params={"items": json.dumps(["abc", "def"])} |
||||
|
) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == ["abc", "def"] |
||||
|
|
||||
|
|
||||
|
def test_header_json_list(): |
||||
|
response = client.get( |
||||
|
"/header-json-list", headers={"x-items": json.dumps(["abc", "def"])} |
||||
|
) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == ["abc", "def"] |
||||
|
|
||||
|
|
||||
|
def test_cookie_json_list(): |
||||
|
client.cookies.set("items", json.dumps(["abc", "def"])) |
||||
|
response = client.get("/cookie-json-list") |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == ["abc", "def"] |
||||
|
client.cookies.clear() |
||||
@ -0,0 +1,173 @@ |
|||||
|
"""Test using special types (Response, Request, BackgroundTasks) as dependency annotations. |
||||
|
|
||||
|
These tests verify that special FastAPI types can be used with Depends() annotations |
||||
|
and that the dependency injection system properly handles them. |
||||
|
""" |
||||
|
|
||||
|
from typing import Annotated |
||||
|
|
||||
|
from fastapi import BackgroundTasks, Depends, FastAPI, Request, Response |
||||
|
from fastapi.responses import JSONResponse |
||||
|
from fastapi.testclient import TestClient |
||||
|
|
||||
|
|
||||
|
def test_response_with_depends_annotated(): |
||||
|
"""Response type hint should work with Annotated[Response, Depends(...)].""" |
||||
|
app = FastAPI() |
||||
|
|
||||
|
def modify_response(response: Response) -> Response: |
||||
|
response.headers["X-Custom"] = "modified" |
||||
|
return response |
||||
|
|
||||
|
@app.get("/") |
||||
|
def endpoint(response: Annotated[Response, Depends(modify_response)]): |
||||
|
return {"status": "ok"} |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
resp = client.get("/") |
||||
|
|
||||
|
assert resp.status_code == 200 |
||||
|
assert resp.json() == {"status": "ok"} |
||||
|
assert resp.headers.get("X-Custom") == "modified" |
||||
|
|
||||
|
|
||||
|
def test_response_with_depends_default(): |
||||
|
"""Response type hint should work with Response = Depends(...).""" |
||||
|
app = FastAPI() |
||||
|
|
||||
|
def modify_response(response: Response) -> Response: |
||||
|
response.headers["X-Custom"] = "modified" |
||||
|
return response |
||||
|
|
||||
|
@app.get("/") |
||||
|
def endpoint(response: Response = Depends(modify_response)): |
||||
|
return {"status": "ok"} |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
resp = client.get("/") |
||||
|
|
||||
|
assert resp.status_code == 200 |
||||
|
assert resp.json() == {"status": "ok"} |
||||
|
assert resp.headers.get("X-Custom") == "modified" |
||||
|
|
||||
|
|
||||
|
def test_response_without_depends(): |
||||
|
"""Regular Response injection should still work.""" |
||||
|
app = FastAPI() |
||||
|
|
||||
|
@app.get("/") |
||||
|
def endpoint(response: Response): |
||||
|
response.headers["X-Direct"] = "set" |
||||
|
return {"status": "ok"} |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
resp = client.get("/") |
||||
|
|
||||
|
assert resp.status_code == 200 |
||||
|
assert resp.json() == {"status": "ok"} |
||||
|
assert resp.headers.get("X-Direct") == "set" |
||||
|
|
||||
|
|
||||
|
def test_response_dependency_chain(): |
||||
|
"""Response dependency should work in a chain of dependencies.""" |
||||
|
app = FastAPI() |
||||
|
|
||||
|
def first_modifier(response: Response) -> Response: |
||||
|
response.headers["X-First"] = "1" |
||||
|
return response |
||||
|
|
||||
|
def second_modifier( |
||||
|
response: Annotated[Response, Depends(first_modifier)], |
||||
|
) -> Response: |
||||
|
response.headers["X-Second"] = "2" |
||||
|
return response |
||||
|
|
||||
|
@app.get("/") |
||||
|
def endpoint(response: Annotated[Response, Depends(second_modifier)]): |
||||
|
return {"status": "ok"} |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
resp = client.get("/") |
||||
|
|
||||
|
assert resp.status_code == 200 |
||||
|
assert resp.headers.get("X-First") == "1" |
||||
|
assert resp.headers.get("X-Second") == "2" |
||||
|
|
||||
|
|
||||
|
def test_response_dependency_returns_different_response_instance(): |
||||
|
"""Dependency that returns a different Response instance should work. |
||||
|
|
||||
|
When a dependency returns a new Response object (e.g., JSONResponse) instead |
||||
|
of modifying the injected one, the returned response should be used and any |
||||
|
modifications to it in the endpoint should be preserved. |
||||
|
""" |
||||
|
app = FastAPI() |
||||
|
|
||||
|
def default_response() -> Response: |
||||
|
response = JSONResponse(content={"status": "ok"}) |
||||
|
response.headers["X-Custom"] = "initial" |
||||
|
return response |
||||
|
|
||||
|
@app.get("/") |
||||
|
def endpoint(response: Annotated[Response, Depends(default_response)]): |
||||
|
response.headers["X-Custom"] = "modified" |
||||
|
return response |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
resp = client.get("/") |
||||
|
|
||||
|
assert resp.status_code == 200 |
||||
|
assert resp.json() == {"status": "ok"} |
||||
|
assert resp.headers.get("X-Custom") == "modified" |
||||
|
|
||||
|
|
||||
|
# Tests for Request type hint with Depends |
||||
|
def test_request_with_depends_annotated(): |
||||
|
"""Request type hint should work in dependency chain.""" |
||||
|
app = FastAPI() |
||||
|
|
||||
|
def extract_request_info(request: Request) -> dict: |
||||
|
return { |
||||
|
"path": request.url.path, |
||||
|
"user_agent": request.headers.get("user-agent", "unknown"), |
||||
|
} |
||||
|
|
||||
|
@app.get("/") |
||||
|
def endpoint( |
||||
|
info: Annotated[dict, Depends(extract_request_info)], |
||||
|
): |
||||
|
return info |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
resp = client.get("/", headers={"user-agent": "test-agent"}) |
||||
|
|
||||
|
assert resp.status_code == 200 |
||||
|
assert resp.json() == {"path": "/", "user_agent": "test-agent"} |
||||
|
|
||||
|
|
||||
|
# Tests for BackgroundTasks type hint with Depends |
||||
|
def test_background_tasks_with_depends_annotated(): |
||||
|
"""BackgroundTasks type hint should work with Annotated[BackgroundTasks, Depends(...)].""" |
||||
|
app = FastAPI() |
||||
|
task_results = [] |
||||
|
|
||||
|
def background_task(message: str): |
||||
|
task_results.append(message) |
||||
|
|
||||
|
def add_background_task(background_tasks: BackgroundTasks) -> BackgroundTasks: |
||||
|
background_tasks.add_task(background_task, "from dependency") |
||||
|
return background_tasks |
||||
|
|
||||
|
@app.get("/") |
||||
|
def endpoint( |
||||
|
background_tasks: Annotated[BackgroundTasks, Depends(add_background_task)], |
||||
|
): |
||||
|
background_tasks.add_task(background_task, "from endpoint") |
||||
|
return {"status": "ok"} |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
resp = client.get("/") |
||||
|
|
||||
|
assert resp.status_code == 200 |
||||
|
assert "from dependency" in task_results |
||||
|
assert "from endpoint" in task_results |
||||
Loading…
Reference in new issue