You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
108 lines
3.0 KiB
Python
108 lines
3.0 KiB
Python
from fastapi import Depends
|
|
from fastapi.testclient import TestClient
|
|
|
|
from iti import create_app
|
|
from iti.auth import (
|
|
Actor,
|
|
Principal,
|
|
StaticPermissionProvider,
|
|
create_access_token,
|
|
require_actor,
|
|
require_permission,
|
|
require_service,
|
|
require_user,
|
|
)
|
|
from iti.config import BaseConfig
|
|
from iti.responses import ok
|
|
|
|
|
|
class Provider(StaticPermissionProvider):
|
|
def load_principal(self, principal_id, request):
|
|
return Principal(
|
|
id=principal_id,
|
|
permissions=frozenset({"demo.read"}),
|
|
scopes=frozenset({"svc:read"}),
|
|
)
|
|
|
|
|
|
class AuthModule:
|
|
name = "auth"
|
|
|
|
def register_routes(self, app):
|
|
@app.get("/me")
|
|
def me(principal=Depends(require_user)):
|
|
return ok({"id": principal.id})
|
|
|
|
@app.get("/allowed")
|
|
def allowed(principal=Depends(require_permission("demo.read"))):
|
|
return ok({"id": principal.id})
|
|
|
|
@app.get("/denied")
|
|
def denied(principal=Depends(require_permission("demo.write"))):
|
|
return ok({"id": principal.id})
|
|
|
|
@app.get("/service")
|
|
def service(actor: Actor = Depends(require_service())):
|
|
return ok({"id": actor.id, "type": actor.type})
|
|
|
|
@app.get("/actor")
|
|
def actor(
|
|
actor_value: Actor = Depends(
|
|
require_actor(permissions=["demo.read"], allow_service=True)
|
|
),
|
|
):
|
|
return ok({"id": actor_value.id, "type": actor_value.type})
|
|
|
|
|
|
def make_app():
|
|
config = BaseConfig(
|
|
database_url="sqlite+pysqlite:///:memory:",
|
|
testing=True,
|
|
jwt_secret_key="test-secret",
|
|
service_tokens={"erp": "svc-token"},
|
|
)
|
|
return create_app(
|
|
modules=[AuthModule()],
|
|
config_mapping=config,
|
|
permission_provider=Provider(),
|
|
)
|
|
|
|
|
|
def test_jwt_auth_and_permission_dependencies():
|
|
app = make_app()
|
|
token = create_access_token("u1", app.state.config)
|
|
client = TestClient(app)
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
assert client.get("/me", headers=headers).json()["data"]["id"] == "u1"
|
|
assert client.get("/allowed", headers=headers).json()["code"] == 200
|
|
assert client.get("/denied", headers=headers).json()["code"] == 403
|
|
|
|
|
|
def test_missing_auth_returns_envelope():
|
|
client = TestClient(make_app())
|
|
|
|
response = client.get("/me")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["code"] == 401
|
|
|
|
|
|
def test_service_token_and_actor_dependencies():
|
|
app = make_app()
|
|
token = create_access_token("u1", app.state.config)
|
|
client = TestClient(app)
|
|
|
|
assert client.get(
|
|
"/service",
|
|
headers={"Authorization": "Bearer svc-token"},
|
|
).json()["data"] == {"id": "erp", "type": "service"}
|
|
assert client.get(
|
|
"/actor",
|
|
headers={"Authorization": "Bearer svc-token"},
|
|
).json()["data"]["type"] == "service"
|
|
assert client.get(
|
|
"/actor",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
).json()["data"]["type"] == "user"
|