Sebastián Ramírez
8 months ago
committed by
GitHub
GitHub
24 changed files with 738 additions and 372 deletions
|
After Width: | Height: | Size: 7.1 KiB |
|
After Width: | Height: | Size: 7.2 KiB |
@ -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", |
|||
}, |
|||
} |
|||
}, |
|||
} |
|||
) |
|||
@ -0,0 +1,46 @@ |
|||
from typing import Dict |
|||
|
|||
import pytest |
|||
from fastapi import Depends, FastAPI, Security |
|||
from fastapi.testclient import TestClient |
|||
from typing_extensions import Annotated |
|||
|
|||
|
|||
@pytest.fixture(name="call_counter") |
|||
def call_counter_fixture(): |
|||
return {"count": 0} |
|||
|
|||
|
|||
@pytest.fixture(name="app") |
|||
def app_fixture(call_counter: Dict[str, int]): |
|||
def get_db(): |
|||
call_counter["count"] += 1 |
|||
return f"db_{call_counter['count']}" |
|||
|
|||
def get_user(db: Annotated[str, Depends(get_db)]): |
|||
return "user" |
|||
|
|||
app = FastAPI() |
|||
|
|||
@app.get("/") |
|||
def endpoint( |
|||
db: Annotated[str, Depends(get_db)], |
|||
user: Annotated[str, Security(get_user, scopes=["read"])], |
|||
): |
|||
return {"db": db} |
|||
|
|||
return app |
|||
|
|||
|
|||
@pytest.fixture(name="client") |
|||
def client_fixture(app: FastAPI): |
|||
return TestClient(app) |
|||
|
|||
|
|||
def test_security_scopes_dependency_called_once( |
|||
client: TestClient, call_counter: Dict[str, int] |
|||
): |
|||
response = client.get("/") |
|||
|
|||
assert response.status_code == 200 |
|||
assert call_counter["count"] == 1 |
|||
@ -0,0 +1,45 @@ |
|||
# Ref: https://github.com/tiangolo/fastapi/issues/5623 |
|||
|
|||
from typing import Any, Dict, List |
|||
|
|||
from fastapi import FastAPI, Security |
|||
from fastapi.security import SecurityScopes |
|||
from fastapi.testclient import TestClient |
|||
from typing_extensions import Annotated |
|||
|
|||
|
|||
async def security1(scopes: SecurityScopes): |
|||
return scopes.scopes |
|||
|
|||
|
|||
async def security2(scopes: SecurityScopes): |
|||
return scopes.scopes |
|||
|
|||
|
|||
async def dep3( |
|||
dep1: Annotated[List[str], Security(security1, scopes=["scope1"])], |
|||
dep2: Annotated[List[str], Security(security2, scopes=["scope2"])], |
|||
): |
|||
return {"dep1": dep1, "dep2": dep2} |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/scopes") |
|||
def get_scopes( |
|||
dep3: Annotated[Dict[str, Any], Security(dep3, scopes=["scope3"])], |
|||
): |
|||
return dep3 |
|||
|
|||
|
|||
client = TestClient(app) |
|||
|
|||
|
|||
def test_security_scopes_dont_propagate(): |
|||
response = client.get("/scopes") |
|||
assert response.status_code == 200 |
|||
assert response.json() == { |
|||
"dep1": ["scope3", "scope1"], |
|||
"dep2": ["scope3", "scope2"], |
|||
} |
|||
@ -0,0 +1,107 @@ |
|||
# Ref: https://github.com/fastapi/fastapi/discussions/6024#discussioncomment-8541913 |
|||
|
|||
from typing import Dict |
|||
|
|||
import pytest |
|||
from fastapi import Depends, FastAPI, Security |
|||
from fastapi.security import SecurityScopes |
|||
from fastapi.testclient import TestClient |
|||
from typing_extensions import Annotated |
|||
|
|||
|
|||
@pytest.fixture(name="call_counts") |
|||
def call_counts_fixture(): |
|||
return { |
|||
"get_db_session": 0, |
|||
"get_current_user": 0, |
|||
"get_user_me": 0, |
|||
"get_user_items": 0, |
|||
} |
|||
|
|||
|
|||
@pytest.fixture(name="app") |
|||
def app_fixture(call_counts: Dict[str, int]): |
|||
def get_db_session(): |
|||
call_counts["get_db_session"] += 1 |
|||
return f"db_session_{call_counts['get_db_session']}" |
|||
|
|||
def get_current_user( |
|||
security_scopes: SecurityScopes, |
|||
db_session: Annotated[str, Depends(get_db_session)], |
|||
): |
|||
call_counts["get_current_user"] += 1 |
|||
return { |
|||
"user": f"user_{call_counts['get_current_user']}", |
|||
"scopes": security_scopes.scopes, |
|||
"db_session": db_session, |
|||
} |
|||
|
|||
def get_user_me( |
|||
current_user: Annotated[dict, Security(get_current_user, scopes=["me"])], |
|||
): |
|||
call_counts["get_user_me"] += 1 |
|||
return { |
|||
"user_me": f"user_me_{call_counts['get_user_me']}", |
|||
"current_user": current_user, |
|||
} |
|||
|
|||
def get_user_items( |
|||
user_me: Annotated[dict, Depends(get_user_me)], |
|||
): |
|||
call_counts["get_user_items"] += 1 |
|||
return { |
|||
"user_items": f"user_items_{call_counts['get_user_items']}", |
|||
"user_me": user_me, |
|||
} |
|||
|
|||
app = FastAPI() |
|||
|
|||
@app.get("/") |
|||
def path_operation( |
|||
user_me: Annotated[dict, Depends(get_user_me)], |
|||
user_items: Annotated[dict, Security(get_user_items, scopes=["items"])], |
|||
): |
|||
return { |
|||
"user_me": user_me, |
|||
"user_items": user_items, |
|||
} |
|||
|
|||
return app |
|||
|
|||
|
|||
@pytest.fixture(name="client") |
|||
def client_fixture(app: FastAPI): |
|||
return TestClient(app) |
|||
|
|||
|
|||
def test_security_scopes_sub_dependency_caching( |
|||
client: TestClient, call_counts: Dict[str, int] |
|||
): |
|||
response = client.get("/") |
|||
|
|||
assert response.status_code == 200 |
|||
assert call_counts["get_db_session"] == 1 |
|||
assert call_counts["get_current_user"] == 2 |
|||
assert call_counts["get_user_me"] == 2 |
|||
assert call_counts["get_user_items"] == 1 |
|||
assert response.json() == { |
|||
"user_me": { |
|||
"user_me": "user_me_1", |
|||
"current_user": { |
|||
"user": "user_1", |
|||
"scopes": ["me"], |
|||
"db_session": "db_session_1", |
|||
}, |
|||
}, |
|||
"user_items": { |
|||
"user_items": "user_items_1", |
|||
"user_me": { |
|||
"user_me": "user_me_2", |
|||
"current_user": { |
|||
"user": "user_2", |
|||
"scopes": ["items", "me"], |
|||
"db_session": "db_session_1", |
|||
}, |
|||
}, |
|||
}, |
|||
} |
|||
Loading…
Reference in new issue