mirror of
https://github.com/hay-kot/mealie.git
synced 2025-08-22 14:33:33 -07:00
basic token tests
This commit is contained in:
parent
81ae051308
commit
7a7b25904a
6 changed files with 60 additions and 8 deletions
|
@ -8,8 +8,8 @@ class LongLiveToken(SqlAlchemyBase, BaseMixins):
|
|||
__tablename__ = "long_live_tokens"
|
||||
id = Column(Integer, primary_key=True)
|
||||
parent_id = Column(Integer, ForeignKey("users.id"))
|
||||
name = Column(String)
|
||||
token = Column(String, unique=True, nullable=False)
|
||||
name = Column(String, nullable=False)
|
||||
token = Column(String, nullable=False)
|
||||
user = orm.relationship("User")
|
||||
|
||||
def __init__(self, session, name, token, parent_id) -> None:
|
||||
|
|
|
@ -8,7 +8,8 @@ from mealie.core.config import settings
|
|||
from mealie.db.database import db
|
||||
from mealie.db.db_setup import generate_session
|
||||
from mealie.schema.auth import TokenData
|
||||
from mealie.schema.user import UserInDB
|
||||
from mealie.schema.user import LongLiveTokenInDB, UserInDB
|
||||
from sqlalchemy.orm.session import Session
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/token")
|
||||
ALGORITHM = "HS256"
|
||||
|
@ -23,8 +24,14 @@ async def get_current_user(token: str = Depends(oauth2_scheme), session=Depends(
|
|||
try:
|
||||
payload = jwt.decode(token, settings.SECRET, algorithms=[ALGORITHM])
|
||||
username: str = payload.get("sub")
|
||||
long_token: str = payload.get("long_token")
|
||||
|
||||
if long_token is not None:
|
||||
return validate_long_live_token(session, token, payload.get("id"))
|
||||
|
||||
if username is None:
|
||||
raise credentials_exception
|
||||
|
||||
token_data = TokenData(username=username)
|
||||
except JWTError:
|
||||
raise credentials_exception
|
||||
|
@ -35,6 +42,16 @@ async def get_current_user(token: str = Depends(oauth2_scheme), session=Depends(
|
|||
return user
|
||||
|
||||
|
||||
def validate_long_live_token(session: Session, client_token: str, id: int) -> UserInDB:
|
||||
|
||||
tokens: list[LongLiveTokenInDB] = db.api_tokens.get(session, id, "parent_id", limit=9999)
|
||||
|
||||
for token in tokens:
|
||||
token: LongLiveTokenInDB
|
||||
if token.token == client_token:
|
||||
return token.user
|
||||
|
||||
|
||||
async def validate_file_token(token: Optional[str] = None) -> Path:
|
||||
credentials_exception = HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
|
|
|
@ -12,7 +12,7 @@ from sqlalchemy.orm.session import Session
|
|||
router = APIRouter(prefix="/api/users", tags=["User API Tokens"])
|
||||
|
||||
|
||||
@router.post("/api-tokens")
|
||||
@router.post("/api-tokens", status_code=status.HTTP_201_CREATED)
|
||||
async def create_api_token(
|
||||
token_name: LoingLiveTokenIn,
|
||||
current_user: UserInDB = Depends(get_current_user),
|
||||
|
@ -20,7 +20,7 @@ async def create_api_token(
|
|||
):
|
||||
""" Create api_token in the Database """
|
||||
|
||||
token_data = {"long_token": True, "user": current_user.email}
|
||||
token_data = {"long_token": True, "id": current_user.id}
|
||||
|
||||
five_years = timedelta(1825)
|
||||
token = create_access_token(token_data, five_years)
|
||||
|
|
|
@ -25,6 +25,9 @@ class CreateToken(LoingLiveTokenIn):
|
|||
parent_id: int
|
||||
token: str
|
||||
|
||||
class Config:
|
||||
orm_mode = True
|
||||
|
||||
|
||||
class ChangePassword(CamelModel):
|
||||
current_password: str
|
||||
|
@ -115,7 +118,7 @@ class GroupInDB(UpdateGroup):
|
|||
}
|
||||
|
||||
|
||||
class LongLiveTokenInDB(LoingLiveTokenIn):
|
||||
class LongLiveTokenInDB(CreateToken):
|
||||
id: int
|
||||
user: UserInDB
|
||||
|
||||
|
|
|
@ -8,7 +8,7 @@ class AppRoutes:
|
|||
self.users_sign_ups = "/api/users/sign-ups"
|
||||
self.users = "/api/users"
|
||||
self.users_self = "/api/users/self"
|
||||
self.users_api_tokens = "/api/users-tokens"
|
||||
self.users_api_tokens = "/api/users/api-tokens"
|
||||
self.groups = "/api/groups"
|
||||
self.groups_self = "/api/groups/self"
|
||||
self.recipes_summary = "/api/recipes/summary"
|
||||
|
@ -60,7 +60,7 @@ class AppRoutes:
|
|||
return f"{self.prefix}/users/{id}/password"
|
||||
|
||||
def users_api_tokens_token_id(self, token_id):
|
||||
return f"{self.prefix}/users-tokens/{token_id}"
|
||||
return f"{self.prefix}/users/api-tokens/{token_id}"
|
||||
|
||||
def groups_id(self, id):
|
||||
return f"{self.prefix}/groups/{id}"
|
||||
|
|
|
@ -0,0 +1,32 @@
|
|||
import json
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
from pytest import fixture
|
||||
from tests.app_routes import AppRoutes
|
||||
|
||||
|
||||
@fixture
|
||||
def long_live_token(api_client: TestClient, api_routes: AppRoutes, token):
|
||||
response = api_client.post(api_routes.users_api_tokens, json={"name": "Test Fixture Token"}, headers=token)
|
||||
assert response.status_code == 201
|
||||
|
||||
return {"Authorization": f"Bearer {json.loads(response.text).get('token')}"}
|
||||
|
||||
|
||||
def test_api_token_creation(api_client: TestClient, api_routes: AppRoutes, token):
|
||||
response = api_client.post(api_routes.users_api_tokens, json={"name": "Test API Token"}, headers=token)
|
||||
assert response.status_code == 201
|
||||
|
||||
|
||||
def test_use_token(api_client: TestClient, api_routes: AppRoutes, long_live_token):
|
||||
response = api_client.get(api_routes.users, headers=long_live_token)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_delete_token(api_client: TestClient, api_routes: AppRoutes, token):
|
||||
response = api_client.delete(api_routes.users_api_tokens_token_id(1), headers=token)
|
||||
assert response.status_code == 200
|
||||
|
||||
response = api_client.delete(api_routes.users_api_tokens_token_id(2), headers=token)
|
||||
assert response.status_code == 200
|
Loading…
Add table
Add a link
Reference in a new issue