mirror of
https://github.com/d3vyce/fastapi-toolsets.git
synced 2026-08-14 19:58:39 +00:00
wip4
This commit is contained in:
+100
-236
@@ -6,12 +6,11 @@ from fastapi.testclient import TestClient
|
||||
|
||||
from fastapi_toolsets.exceptions import UnauthorizedError, init_exceptions_handlers
|
||||
from fastapi_toolsets.security import (
|
||||
APIKeyHeaderAuth,
|
||||
AuthSource,
|
||||
BearerTokenAuth,
|
||||
CookieAuth,
|
||||
MultiAuth,
|
||||
OAuth2Auth,
|
||||
OpenIDAuth,
|
||||
)
|
||||
|
||||
|
||||
@@ -316,73 +315,85 @@ class TestCookieAuth:
|
||||
assert await auth.extract(request) == "abc"
|
||||
|
||||
|
||||
class TestOAuth2Auth:
|
||||
def test_valid_token_returns_identity(self):
|
||||
oauth = OAuth2Auth(token_url="/token", validator=simple_validator)
|
||||
class TestAPIKeyHeaderAuth:
|
||||
def test_valid_key_returns_identity(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oauth)):
|
||||
async def me(user=Security(auth)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"})
|
||||
response = client.get("/me", headers={"X-API-Key": VALID_TOKEN})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice"}
|
||||
|
||||
def test_missing_token_returns_401(self):
|
||||
oauth = OAuth2Auth(token_url="/token", validator=simple_validator)
|
||||
def test_missing_header_returns_401(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oauth)):
|
||||
async def me(user=Security(auth)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me")
|
||||
assert response.status_code == 401
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_no_header(self):
|
||||
from starlette.requests import Request
|
||||
def test_invalid_key_returns_401(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
|
||||
auth = OAuth2Auth("/token", simple_validator)
|
||||
scope = {"type": "http", "method": "GET", "path": "/", "headers": []}
|
||||
request = Request(scope)
|
||||
assert await auth.extract(request) is None
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(auth)):
|
||||
return user
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_empty_token(self):
|
||||
from starlette.requests import Request
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"X-API-Key": "wrong"})
|
||||
assert response.status_code == 401
|
||||
|
||||
auth = OAuth2Auth("/token", simple_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"authorization", b"Bearer ")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await auth.extract(request) is None
|
||||
def test_kwargs_forwarded_to_validator(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", role_validator, role="admin")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_token(self):
|
||||
from starlette.requests import Request
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(auth)):
|
||||
return user
|
||||
|
||||
auth = OAuth2Auth("/token", simple_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"authorization", b"Bearer mytoken")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await auth.extract(request) == "mytoken"
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"X-API-Key": VALID_TOKEN})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice", "role": "admin"}
|
||||
|
||||
def test_require_forwards_kwargs(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", role_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/admin")
|
||||
async def admin(user=Security(auth.require(role="admin"))):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/admin", headers={"X-API-Key": VALID_TOKEN})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice", "role": "admin"}
|
||||
|
||||
def test_require_preserves_name(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
derived = auth.require(role="admin")
|
||||
assert derived._name == "X-API-Key"
|
||||
|
||||
def test_require_does_not_mutate_original(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", role_validator, role="user")
|
||||
auth.require(role="admin")
|
||||
assert auth._kwargs == {"role": "user"}
|
||||
|
||||
def test_in_multi_auth(self):
|
||||
"""OAuth2Auth.authenticate() is exercised when used inside MultiAuth."""
|
||||
oauth = OAuth2Auth(token_url="/token", validator=simple_validator)
|
||||
multi = MultiAuth(oauth)
|
||||
"""APIKeyHeaderAuth.authenticate() is exercised inside MultiAuth."""
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
api_key = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
multi = MultiAuth(bearer, api_key)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
@@ -390,10 +401,52 @@ class TestOAuth2Auth:
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"})
|
||||
# No bearer → falls through to API key header
|
||||
response = client.get("/me", headers={"X-API-Key": VALID_TOKEN})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice"}
|
||||
|
||||
def test_is_auth_source(self):
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
assert isinstance(auth, AuthSource)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_no_header(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
scope = {"type": "http", "method": "GET", "path": "/", "headers": []}
|
||||
request = Request(scope)
|
||||
assert await auth.extract(request) is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_empty_header(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"x-api-key", b"")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await auth.extract(request) is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_key_present(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
auth = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"x-api-key", b"mykey")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await auth.extract(request) == "mykey"
|
||||
|
||||
|
||||
class TestMultiAuth:
|
||||
def test_first_source_matches(self):
|
||||
@@ -626,26 +679,6 @@ class TestRequire:
|
||||
derived = cookie.require(scope="admin")
|
||||
assert derived._name == "session"
|
||||
|
||||
def test_oauth2_require_forwards_kwargs(self):
|
||||
oauth = OAuth2Auth("/token", role_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/admin")
|
||||
async def admin(user=Security(oauth.require(role="admin"))):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get(
|
||||
"/admin", headers={"Authorization": f"Bearer {VALID_TOKEN}"}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice", "role": "admin"}
|
||||
|
||||
def test_oauth2_require_preserves_token_url(self):
|
||||
oauth = OAuth2Auth("/token", simple_validator)
|
||||
derived = oauth.require(role="admin")
|
||||
assert derived._token_url == "/token"
|
||||
|
||||
def test_bearer_require_in_multi_auth(self):
|
||||
"""require() instances work seamlessly inside MultiAuth."""
|
||||
PREFIXED_TOKEN = f"user_{VALID_TOKEN}"
|
||||
@@ -671,141 +704,6 @@ class TestRequire:
|
||||
assert response.json() == {"user": "alice", "role": "admin"}
|
||||
|
||||
|
||||
class TestOpenIDAuth:
|
||||
DISCOVERY_URL = "https://accounts.example.com/.well-known/openid-configuration"
|
||||
|
||||
def test_valid_token_returns_identity(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oidc)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice"}
|
||||
|
||||
def test_missing_token_returns_401(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oidc)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me")
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_invalid_token_returns_401(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oidc)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": "Bearer wrong"})
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_kwargs_forwarded_to_validator(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, role_validator, role="admin")
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oidc)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice", "role": "admin"}
|
||||
|
||||
def test_require_forwards_kwargs(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, role_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/admin")
|
||||
async def admin(user=Security(oidc.require(role="admin"))):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get(
|
||||
"/admin", headers={"Authorization": f"Bearer {VALID_TOKEN}"}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice", "role": "admin"}
|
||||
|
||||
def test_require_preserves_discovery_url(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
derived = oidc.require(role="admin")
|
||||
assert derived._openid_connect_url == self.DISCOVERY_URL
|
||||
|
||||
def test_require_does_not_mutate_original(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, role_validator, role="user")
|
||||
oidc.require(role="admin")
|
||||
assert oidc._kwargs == {"role": "user"}
|
||||
|
||||
def test_is_auth_source(self):
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
assert isinstance(oidc, AuthSource)
|
||||
|
||||
def test_in_multi_auth(self):
|
||||
"""OpenIDAuth works seamlessly inside MultiAuth."""
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
multi = MultiAuth(oidc)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(multi)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"})
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"user": "alice"}
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_no_header(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
scope = {"type": "http", "method": "GET", "path": "/", "headers": []}
|
||||
request = Request(scope)
|
||||
assert await oidc.extract(request) is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_empty_token(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"authorization", b"Bearer ")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await oidc.extract(request) is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_extract_token(self):
|
||||
from starlette.requests import Request
|
||||
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"headers": [(b"authorization", b"Bearer mytoken")],
|
||||
}
|
||||
request = Request(scope)
|
||||
assert await oidc.extract(request) == "mytoken"
|
||||
|
||||
|
||||
class TestSyncValidators:
|
||||
"""Sync (non-async) validators — covers the sync path in _call_validator."""
|
||||
|
||||
@@ -849,36 +747,6 @@ class TestSyncValidators:
|
||||
assert response.json() == {"user": "alice"}
|
||||
|
||||
|
||||
class TestOpenIDAuthEdgeCases:
|
||||
DISCOVERY_URL = "https://accounts.example.com/.well-known/openid-configuration"
|
||||
|
||||
def test_non_bearer_authorization_returns_401(self):
|
||||
"""Authorization header present but not Bearer scheme."""
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oidc)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": "Basic dXNlcjpwYXNz"})
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_empty_bearer_token_returns_401(self):
|
||||
"""Authorization: Bearer with no token after the scheme prefix."""
|
||||
oidc = OpenIDAuth(self.DISCOVERY_URL, simple_validator)
|
||||
|
||||
def setup(app: FastAPI):
|
||||
@app.get("/me")
|
||||
async def me(user=Security(oidc)):
|
||||
return user
|
||||
|
||||
client = TestClient(_app(setup))
|
||||
response = client.get("/me", headers={"Authorization": "Bearer "})
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
class TestCookieAuthSigned:
|
||||
"""CookieAuth with HMAC-SHA256 signed cookies (secret_key path)."""
|
||||
|
||||
@@ -1029,14 +897,10 @@ class TestAuthSource:
|
||||
def test_builtin_classes_are_auth_sources(self):
|
||||
bearer = BearerTokenAuth(simple_validator)
|
||||
cookie = CookieAuth("session", cookie_validator)
|
||||
oauth = OAuth2Auth("/token", simple_validator)
|
||||
oidc = OpenIDAuth(
|
||||
"https://example.com/.well-known/openid-configuration", simple_validator
|
||||
)
|
||||
api_key = APIKeyHeaderAuth("X-API-Key", simple_validator)
|
||||
assert isinstance(bearer, AuthSource)
|
||||
assert isinstance(cookie, AuthSource)
|
||||
assert isinstance(oauth, AuthSource)
|
||||
assert isinstance(oidc, AuthSource)
|
||||
assert isinstance(api_key, AuthSource)
|
||||
|
||||
def test_custom_source_standalone_valid(self):
|
||||
"""Default __call__ wires extract + authenticate via Request injection."""
|
||||
|
||||
Reference in New Issue
Block a user