RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: Consolidate get methods with JWT token resolution

This commit is contained in:
Vineeth Voruganti 2025-04-07 17:06:53 -04:00
parent 253b8109cb
commit 8d2f3db6ba
14 changed files with 1024 additions and 787 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
src/crud.py
View File

@ -348,7 +348,22 @@ async def get_session(
app_id: str,
session_id: str,
user_id: Optional[str] = None,
) -> Optional[models.Session]:
) -> models.Session:
"""
Get a session by ID for a specific user and app.
Args:
db: Database session
app_id: Public ID of the app
session_id: Public ID of the session
user_id: Optional public ID of the user
Returns:
The session if found
Raises:
ResourceNotFoundException: If the session does not exist or doesn't belong to the user
"""
stmt = (
select(models.Session)
.join(models.User, models.User.public_id == models.Session.user_id)
@ -359,6 +374,11 @@ async def get_session(
stmt = stmt.where(models.Session.user_id == user_id)
result = await db.execute(stmt)
session = result.scalar_one_or_none()
if session is None:
logger.warning(
f"Session with ID '{session_id}' not found for user {user_id}"
)
raise ResourceNotFoundException("Session not found or does not belong to user")
return session
@ -750,10 +770,9 @@ async def get_message(
select(models.Message)
.join(models.Session, models.Session.public_id == models.Message.session_id)
.join(models.User, models.User.public_id == models.Session.user_id)
.join(models.App, models.App.public_id == models.User.app_id)
.where(models.App.public_id == app_id)
.where(models.User.app_id == app_id)
.where(models.User.public_id == user_id)
.where(models.Message.session_id == session_id)
.where(models.Session.public_id == session_id)
.where(models.Message.public_id == message_id)
)
result = await db.execute(stmt)
@ -902,22 +921,20 @@ async def get_metamessage(
stmt = (
select(models.Metamessage)
.join(models.User, models.User.public_id == models.Metamessage.user_id)
.join(models.App, models.App.public_id == models.User.app_id)
.where(models.App.public_id == app_id)
.where(models.User.app_id == app_id)
.where(models.User.public_id == user_id)
.where(models.Metamessage.public_id == metamessage_id)
)
# Add session filter if provided
# Add optional filters
if session_id is not None:
stmt = stmt.where(models.Metamessage.session_id == session_id)
# Add message filter if provided
if message_id is not None:
stmt = stmt.where(models.Metamessage.message_id == message_id)
result = await db.execute(stmt)
return result.scalar_one_or_none()
metamessage = result.scalar_one_or_none()
return metamessage
async def update_metamessage(
@ -927,25 +944,19 @@ async def update_metamessage(
metamessage_id: str,
) -> bool:
# First retrieve the metamessage
honcho_metamessage = await get_metamessage(
db,
app_id=app_id,
user_id=metamessage.user_id,
metamessage_id=metamessage_id,
session_id=metamessage.session_id,
message_id=metamessage.message_id,
metamessage_obj = await get_metamessage(
db, app_id=app_id, user_id=metamessage.user_id, metamessage_id=metamessage_id
)
if honcho_metamessage is None:
if metamessage_obj is None:
raise ResourceNotFoundException(
"Metamessage not found or does not belong to user"
f"Metamessage with ID {metamessage_id} not found"
)
# Validate the consistency of relationships if they're being changed
# If we're setting message_id, we must have a session_id
if metamessage.message_id is not None and metamessage.session_id is None:
# If updating message_id but not session_id, use the existing session_id
metamessage.session_id = honcho_metamessage.session_id
metamessage.session_id = metamessage_obj.session_id
if metamessage.session_id is None:
raise ValidationException("Cannot specify message_id without session_id")
@ -965,19 +976,19 @@ async def update_metamessage(
# Update fields
if metamessage.session_id is not None:
honcho_metamessage.session_id = metamessage.session_id
metamessage_obj.session_id = metamessage.session_id
if metamessage.message_id is not None:
honcho_metamessage.message_id = metamessage.message_id
metamessage_obj.message_id = metamessage.message_id
if metamessage.metadata is not None:
honcho_metamessage.h_metadata = metamessage.metadata
metamessage_obj.h_metadata = metamessage.metadata
if metamessage.metamessage_type is not None:
honcho_metamessage.metamessage_type = metamessage.metamessage_type
metamessage_obj.metamessage_type = metamessage.metamessage_type
await db.commit()
return honcho_metamessage
return metamessage_obj
########################################################

4
src/main.py
View File

@ -10,7 +10,7 @@ from fastapi_pagination import add_pagination
from sentry_sdk.integrations.fastapi import FastApiIntegration
from sentry_sdk.integrations.starlette import StarletteIntegration
from src.db import engine, scaffold_db
from src.db import engine
from src.exceptions import HonchoException
from src.routers import (
apps,
@ -25,7 +25,6 @@ from src.routers import (
from src.security import create_admin_jwt
def get_log_level(env_var="LOG_LEVEL", default="INFO"):
"""
Convert log level string from environment variable to logging module constant.
@ -70,7 +69,6 @@ SENTRY_ENABLED = os.getenv("SENTRY_ENABLED", "False").lower() == "true"
if SENTRY_ENABLED:
sentry_sdk.init(
dsn=os.getenv("SENTRY_DSN"),
enable_tracing=True,
traces_sample_rate=0.4,
profiles_sample_rate=0.4,
integrations=[

76
src/routers/apps.py
View File

@ -1,14 +1,14 @@
import logging
from typing import Optional
from fastapi import APIRouter, Depends
from fastapi import APIRouter, Depends, Query
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
from src import crud, schemas
from src.dependencies import db
from src.exceptions import AuthenticationException, ResourceNotFoundException
from src.security import require_auth
from src.security import JWTParams, require_auth
logger = logging.getLogger(__name__)
@ -17,21 +17,60 @@ router = APIRouter(
tags=["apps"],
)
jwt_params = Depends(require_auth(app_id="app_id"))
# jwt_params = Depends(require_auth(app_id="app_id"))
@router.get(
"",
response_model=schemas.App,
)
async def get_app_from_token(jwt_params=jwt_params, db=db):
@router.get("", response_model=schemas.App)
async def get_app(
app_id: Optional[str] = Query(
None, description="App ID to retrieve. If not provided, uses JWT token"
),
jwt_params: JWTParams = Depends(require_auth()),
db=db,
):
"""
Get an App by ID from the app_id provided in the JWT.
If no app_id is provided, return a 401 Unauthorized error.
Get an App by ID.
If app_id is provided as a query parameter, it uses that (must match JWT app_id).
Otherwise, it uses the app_id from the JWT token.
"""
if jwt_params.ap is None:
raise AuthenticationException("App not found in JWT")
return await crud.get_app(db, app_id=jwt_params.ap)
# If app_id provided in query, check if it matches jwt or user is admin
if app_id:
if not jwt_params.ad and jwt_params.ap != app_id:
raise AuthenticationException("Unauthorized access to resource")
target_app_id = app_id
else:
# Use app_id from JWT
if not jwt_params.ap:
raise AuthenticationException("App ID not found in query parameter or JWT")
target_app_id = jwt_params.ap
return await crud.get_app(db, app_id=target_app_id)
# @router.get(
# "",
# response_model=schemas.App,
# )
# async def get_app_from_token(jwt_params=jwt_params, db=db):
# """
# Get an App by ID from the app_id provided in the JWT.
# If no app_id is provided, return a 401 Unauthorized error.
# """
# if jwt_params.ap is None:
# raise AuthenticationException("App not found in JWT")
# return await crud.get_app(db, app_id=jwt_params.ap)
#
#
# @router.get(
# "/{app_id}",
# response_model=schemas.App,
# dependencies=[Depends(require_auth(app_id="app_id"))],
# )
# async def get_app(app_id: str, db=db):
# """Get an App by ID"""
# app = await crud.get_app(db, app_id=app_id)
# return app
@router.post(
@ -55,17 +94,6 @@ async def get_all_apps(
)
@router.get(
"/{app_id}",
response_model=schemas.App,
dependencies=[Depends(require_auth(app_id="app_id"))],
)
async def get_app(app_id: str, db=db):
"""Get an App by ID"""
app = await crud.get_app(db, app_id=app_id)
return app
@router.get(
"/name/{name}",
response_model=schemas.App,

73
src/routers/collections.py
View File

@ -1,40 +1,65 @@
from typing import Optional
from fastapi import APIRouter, Depends
from fastapi import APIRouter, Depends, Query
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
from src import crud, schemas
from src.dependencies import db
from src.exceptions import AuthenticationException
from src.security import require_auth
from src.exceptions import AuthenticationException, ResourceNotFoundException
from src.security import JWTParams, require_auth
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/collections",
tags=["collections"],
)
jwt_params = Depends(require_auth(collection_id="collection_id"))
@router.get(
"",
response_model=schemas.Collection,
)
async def get_collection_from_token(
async def get_collection(
app_id: str,
user_id: str,
jwt_params=jwt_params,
collection_id: Optional[str] = Query(
None, description="Collection ID to retrieve. If not provided, uses JWT token"
),
jwt_params: JWTParams = Depends(require_auth()),
db=db,
):
"""
Get a specific collection for a user by collection_id provided in the JWT.
If no collection_id is provided, return a 401 Unauthorized error.
Get a specific collection for a user.
If collection_id is provided as a query parameter, it uses that (must match JWT collection_id).
Otherwise, it uses the collection_id from the JWT token.
"""
if jwt_params.co is None:
raise AuthenticationException("Collection not found in JWT")
# Verify JWT has access to the requested resource
if not jwt_params.ad:
if jwt_params.ap is not None and jwt_params.ap != app_id:
raise AuthenticationException("Unauthorized access to resource")
if jwt_params.us is not None and jwt_params.us != user_id:
raise AuthenticationException("Unauthorized access to resource")
# If collection_id provided in query, check if it matches jwt or user is admin
if collection_id:
if (
not jwt_params.ad
and jwt_params.co is not None
and jwt_params.co != collection_id
):
raise AuthenticationException("Unauthorized access to resource")
target_collection_id = collection_id
else:
# Use collection_id from JWT
if not jwt_params.co:
raise AuthenticationException(
"Collection ID not found in query parameter or JWT"
)
target_collection_id = jwt_params.co
# Let crud function handle the ResourceNotFoundException
return await crud.get_collection_by_id(
db, app_id=app_id, collection_id=jwt_params.co, user_id=user_id
db, app_id=app_id, collection_id=target_collection_id, user_id=user_id
)
@ -77,30 +102,6 @@ async def get_collection_by_name(
return honcho_collection
@router.get(
"/{collection_id}",
response_model=schemas.Collection,
dependencies=[
Depends(
require_auth(
app_id="app_id", user_id="user_id", collection_id="collection_id"
)
)
],
)
async def get_collection_by_id(
app_id: str,
user_id: str,
collection_id: str,
db=db,
) -> schemas.Collection:
"""Get a Collection by ID"""
honcho_collection = await crud.get_collection_by_id(
db, app_id=app_id, user_id=user_id, collection_id=collection_id
)
return honcho_collection
@router.post(
"",
response_model=schemas.Collection,

72
src/routers/sessions.py
View File

@ -2,7 +2,7 @@ import logging
from typing import Optional
from anthropic import MessageStreamManager
from fastapi import APIRouter, Depends
from fastapi import APIRouter, Depends, Query
from fastapi.responses import StreamingResponse
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
@ -14,7 +14,7 @@ from src.exceptions import (
ResourceNotFoundException,
ValidationException,
)
from src.security import require_auth
from src.security import JWTParams, require_auth
logger = logging.getLogger(__name__)
@ -23,27 +23,52 @@ router = APIRouter(
tags=["sessions"],
)
jwt_params = Depends(require_auth(session_id="session_id"))
@router.get(
"",
response_model=schemas.Session,
)
async def get_session_from_token(
async def get_session(
app_id: str,
user_id: str,
jwt_params=jwt_params,
session_id: Optional[str] = Query(
None, description="Session ID to retrieve. If not provided, uses JWT token"
),
jwt_params: JWTParams = Depends(require_auth()),
db=db,
):
"""
Get a specific session for a user by session_id provided in the JWT.
If no session_id is provided, return a 401 Unauthorized error.
Get a specific session for a user.
If session_id is provided as a query parameter, it uses that (must match JWT session_id).
Otherwise, it uses the session_id from the JWT token.
"""
if jwt_params.se is None:
raise AuthenticationException("Session not found in JWT")
# Verify JWT has access to the requested resource
if not jwt_params.ad:
if jwt_params.ap is not None and jwt_params.ap != app_id:
raise AuthenticationException("Unauthorized access to resource")
if jwt_params.us is not None and jwt_params.us != user_id:
raise AuthenticationException("Unauthorized access to resource")
# If session_id provided in query, check if it matches jwt or user is admin
if session_id:
if (
not jwt_params.ad
and jwt_params.se is not None
and jwt_params.se != session_id
):
raise AuthenticationException("Unauthorized access to resource")
target_session_id = session_id
else:
# Use session_id from JWT
if not jwt_params.se:
raise AuthenticationException(
"Session ID not found in query parameter or JWT"
)
target_session_id = jwt_params.se
# Let crud function handle the ResourceNotFoundException
return await crud.get_session(
db, app_id=app_id, session_id=jwt_params.se, user_id=user_id
db, app_id=app_id, session_id=target_session_id, user_id=user_id
)
@ -150,31 +175,6 @@ async def delete_session(
raise ResourceNotFoundException("Session not found") from e
@router.get(
"/{session_id}",
response_model=schemas.Session,
dependencies=[
Depends(
require_auth(app_id="app_id", user_id="user_id", session_id="session_id")
)
],
)
async def get_session(
app_id: str,
user_id: str,
session_id: str,
db=db,
):
"""Get a specific session for a user by ID"""
honcho_session = await crud.get_session(
db, app_id=app_id, session_id=session_id, user_id=user_id
)
if honcho_session is None:
logger.warning(f"Session {session_id} not found for user {user_id}")
raise ResourceNotFoundException(f"Session with ID {session_id} not found")
return honcho_session
@router.post(
"/{session_id}/chat",
response_model=schemas.AgentChat,

101
src/routers/users.py
View File

@ -1,6 +1,7 @@
import logging
from typing import Optional
from fastapi import APIRouter, Depends
from fastapi import APIRouter, Depends, Query
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
@ -10,7 +11,7 @@ from src.exceptions import (
AuthenticationException,
ResourceNotFoundException,
)
from src.security import require_auth
from src.security import JWTParams, require_auth
logger = logging.getLogger(__name__)
@ -19,22 +20,6 @@ router = APIRouter(
tags=["users"],
)
jwt_params = Depends(require_auth(user_id="user_id"))
@router.get(
"",
response_model=schemas.User,
)
async def get_user_from_token(app_id: str, jwt_params=jwt_params, db=db):
"""
Get a User by ID from the user_id provided in the JWT.
If no user_id is provided, return a 401 Unauthorized error.
"""
if jwt_params.us is None:
raise AuthenticationException("User not found in JWT")
return await crud.get_user(db, app_id=app_id, user_id=jwt_params.us)
@router.post(
"",
@ -69,6 +54,71 @@ async def get_users(
)
@router.get(
"",
response_model=schemas.User,
)
async def get_user(
app_id: str,
user_id: Optional[str] = Query(
None, description="User ID to retrieve. If not provided, users JWT token"
),
jwt_params: JWTParams = Depends(require_auth()),
db=db,
):
"""
Get a User by ID
If user_id is provided as a query parameter, it uses that (must match JWT app_id).
Otherwise, it uses the user_id from the JWT token.
"""
# validate app query param
if not jwt_params.ad and jwt_params.ap is not None and jwt_params.ap != app_id:
raise AuthenticationException("Unauthorized access to resource")
if user_id:
if not jwt_params.ad and jwt_params.us is not None and jwt_params.us != user_id:
raise AuthenticationException("Unauthorized access to resource")
target_user_id = user_id
else:
# Use user_id from JWT
if not jwt_params.us:
raise AuthenticationException("User ID not found in query parameter or JWT")
target_user_id = jwt_params.us
user = await crud.get_user(db, app_id=app_id, user_id=target_user_id)
return user
# @router.get(
# "",
# response_model=schemas.User,
# )
# async def get_user_from_token(app_id: str, jwt_params=jwt_params, db=db):
# """
# Get a User by ID from the user_id provided in the JWT.
# If no user_id is provided, return a 401 Unauthorized error.
# """
# if jwt_params.us is None:
# raise AuthenticationException("User not found in JWT")
# return await crud.get_user(db, app_id=app_id, user_id=jwt_params.us)
#
#
# @router.get(
# "/{user_id}",
# response_model=schemas.User,
# dependencies=[Depends(require_auth(app_id="app_id", user_id="user_id"))],
# )
# async def get_user(
# app_id: str,
# user_id: str,
# db=db,
# ):
# """Get a User by ID"""
# user = await crud.get_user(db, app_id=app_id, user_id=user_id)
# return user
#
@router.get(
"/name/{name}",
response_model=schemas.User,
@ -90,21 +140,6 @@ async def get_user_by_name(
return user
@router.get(
"/{user_id}",
response_model=schemas.User,
dependencies=[Depends(require_auth(app_id="app_id", user_id="user_id"))],
)
async def get_user(
app_id: str,
user_id: str,
db=db,
):
"""Get a User by ID"""
user = await crud.get_user(db, app_id=app_id, user_id=user_id)
return user
@router.get(
"/get_or_create/{name}",
response_model=schemas.User,

25
src/security.py
View File

@ -123,11 +123,26 @@ def require_auth(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: AsyncSession = Depends(get_db),
):
app_id_param = request.path_params.get(app_id) if app_id else None
user_id_param = request.path_params.get(user_id) if user_id else None
session_id_param = request.path_params.get(session_id) if session_id else None
app_id_param = (
request.path_params.get(app_id) or request.query_params.get(app_id)
if app_id
else None
)
user_id_param = (
request.path_params.get(user_id) or request.query_params.get(user_id)
if user_id
else None
)
session_id_param = (
request.path_params.get(session_id) or request.query_params.get(session_id)
if session_id
else None
)
collection_id_param = (
request.path_params.get(collection_id) if collection_id else None
request.path_params.get(collection_id)
or request.query_params.get(collection_id)
if collection_id
else None
)
return await auth(
@ -181,6 +196,8 @@ async def auth(
return jwt_params
if any([session_id, collection_id, user_id, app_id]):
print([session_id, collection_id, user_id, app_id])
print(jwt_params)
raise AuthenticationException("JWT not permissioned for this resource")
# Route did not specify any parameters, so it should parse parameters itself

14
tests/conftest.py
View File

@ -3,7 +3,7 @@ import os
import sys
import jwt
from nanoid import generate as generate_nanoid
from unittest.mock import patch, MagicMock
from unittest.mock import patch, MagicMock, AsyncMock
import pytest
import pytest_asyncio
@ -235,4 +235,14 @@ def mock_langfuse():
for handler in logging.getLogger().handlers[:]:
if isinstance(handler, TestHandler):
handler.close()
logging.getLogger().removeHandler(handler)
logging.getLogger().removeHandler(handler)
@pytest.fixture(autouse=True)
def mock_openai_embeddings():
"""Mock OpenAI embeddings API calls for testing"""
with patch("src.crud.openai_client.embeddings.create") as mock_create:
mock_response = AsyncMock()
mock_response.data = [MagicMock(embedding=[0.1] * 1536)]
mock_create.return_value = mock_response
yield mock_create

2
tests/routes/test_apps.py
View File

@ -74,7 +74,7 @@ def test_get_or_create_existing_app(client):
def test_get_app_by_id(client, sample_data):
test_app, _ = sample_data
response = client.get(f"/v1/apps/{test_app.public_id}")
response = client.get(f"/v1/apps?app_id={test_app.public_id}")
assert response.status_code == 200
data = response.json()
assert data["name"] == test_app.name

4
tests/routes/test_collections.py
View File

@ -25,7 +25,7 @@ def test_get_collection_by_id(client, sample_data) -> None:
data = response.json()
# Get the collection
response = client.get(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/collections/{data['id']}"
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/collections?collection_id={data['id']}"
)
assert response.status_code == 200
data = response.json()
@ -127,6 +127,6 @@ def test_delete_collection(client, sample_data) -> None:
)
assert response.status_code == 200
response = client.get(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/collections/{data['id']}"
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/collections?collection_id={data['id']}"
)
assert response.status_code == 404

37
tests/routes/test_scoped_api.py
View File

@ -72,7 +72,7 @@ def test_get_app_by_id_with_auth(auth_client, sample_data):
f"Bearer {create_jwt(JWTParams(ap=test_app.public_id))}"
)
response = auth_client.get(f"/v1/apps/{test_app.public_id}")
response = auth_client.get(f"/v1/apps?app_id={test_app.public_id}")
# Admin JWT or JWT with matching app_id should be allowed
if auth_client.auth_type in ["admin", "empty"]:
@ -198,7 +198,7 @@ def test_get_user_by_id_with_auth(auth_client, sample_data):
)
response = auth_client.get(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}"
f"/v1/apps/{test_app.public_id}/users?user_id={test_user.public_id}"
)
# Admin JWT or JWT with matching app_id should be allowed
@ -214,11 +214,19 @@ def test_get_user_by_id_with_auth(auth_client, sample_data):
)
response = auth_client.get(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}"
f"/v1/apps/{test_app.public_id}/users?user_id={test_user.public_id}"
)
assert response.status_code == 200
response2 = auth_client.get(f"/v1/apps/{test_app.public_id}/users")
assert response2.status_code == 200
print(response2.json())
assert response2.json()["id"] == test_user.public_id
def test_get_user_by_name_with_auth(auth_client, sample_data):
test_app, test_user = sample_data
@ -335,7 +343,7 @@ def test_get_session_by_id_with_auth(auth_client, sample_data):
# Test with app and user scoped JWT
response = auth_client.get(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{session_id}"
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions?session_id={session_id}"
)
assert response.status_code == 200
@ -346,10 +354,18 @@ def test_get_session_by_id_with_auth(auth_client, sample_data):
)
response = auth_client.get(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{session_id}"
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions?session_id={session_id}"
)
assert response.status_code == 200
response2 = auth_client.get(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions"
)
assert response2.status_code == 200
assert response2.json()["id"] == session_id
def test_create_collection(auth_client, sample_data) -> None:
test_app, test_user = sample_data
@ -421,7 +437,7 @@ def test_get_collection_by_id_with_auth(auth_client, sample_data) -> None:
# Test with app and user scoped JWT
response = auth_client.get(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/collections/{collection_id}"
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/collections?collection_id={collection_id}"
)
assert response.status_code == 200
@ -432,10 +448,17 @@ def test_get_collection_by_id_with_auth(auth_client, sample_data) -> None:
)
response = auth_client.get(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/collections/{collection_id}"
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/collections?collection_id={collection_id}"
)
assert response.status_code == 200
# Test auto resolution of ID
response2 = auth_client.get(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/collections"
)
assert response2.status_code == 200
assert response2.json()["id"] == collection_id
def test_get_collection_by_name_with_auth(auth_client, sample_data) -> None:
test_app, test_user = sample_data

2
tests/routes/test_sessions.py
View File

@ -118,7 +118,7 @@ async def test_delete_session(client, db_session, sample_data):
)
assert response.status_code == 200
response = client.get(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{test_session.public_id}"
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions?session_id={test_session.public_id}"
)
data = response.json()
assert data["is_active"] is False

2
tests/routes/test_users.py
View File

@ -17,7 +17,7 @@ def test_create_user(client, sample_data):
def test_get_user_by_id(client, sample_data):
test_app, test_user = sample_data
response = client.get(f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}")
response = client.get(f"/v1/apps/{test_app.public_id}/users?user_id={test_user.public_id}")
assert response.status_code == 200
data = response.json()
assert data["name"] == test_user.name

1336
uv.lock
View File

File diff suppressed because it is too large Load Diff