committed by
GitHub
9 changed files with 270 additions and 21 deletions
@ -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