Compare commits

...
2 Commits
Author SHA1 Message Date
akadmin 167a352d44 Fix backend test suite - all 20 tests passing
- Updated JWT tokens to include privlevel for authorization
- Fixed test fixtures to properly hash passwords with bcrypt
- Fixed client fixture to share database session with test fixtures
- Reordered deck router routes to prevent conflicts
- Removed relationship fields from FolderResponse schema
- Updated conftest.py with proper async session management
2026-07-18 16:58:39 +00:00
akadmin 312ac27f88 Fix backend bugs: password hash column rename, dynamic import, AuditLog response_model, websocket reference, missing exports, bcrypt version pin 2026-07-18 05:14:37 +00:00
15 changed files with 301 additions and 185 deletions
+3
View File
@@ -29,6 +29,9 @@ class Base(DeclarativeBase):
pass pass
__all__ = ["Base", "get_db", "async_session", "engine"]
async def get_db() -> AsyncSession: async def get_db() -> AsyncSession:
"""FastAPI dependency that provides a database session.""" """FastAPI dependency that provides a database session."""
async with async_session() as session: async with async_session() as session:
+27 -5
View File
@@ -5,6 +5,8 @@ Implements JWT token management and bcrypt password hashing with salt.
""" """
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from typing import Optional from typing import Optional
from fastapi import Header, HTTPException, status
from jose import JWTError, jwt from jose import JWTError, jwt
from passlib.context import CryptContext from passlib.context import CryptContext
from app.core.settings import get_settings from app.core.settings import get_settings
@@ -26,6 +28,7 @@ def hash_password(password: str) -> str:
def create_access_token( def create_access_token(
subject: str, subject: str,
privlevel: str = "User",
expires_delta: Optional[timedelta] = None, expires_delta: Optional[timedelta] = None,
) -> str: ) -> str:
"""Create a JWT access token.""" """Create a JWT access token."""
@@ -41,11 +44,12 @@ def create_access_token(
"exp": expire, "exp": expire,
"iat": datetime.now(timezone.utc), "iat": datetime.now(timezone.utc),
"type": "access", "type": "access",
"privlevel": privlevel,
} }
return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.JWT_ALGORITHM) return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
def create_refresh_token(subject: str) -> str: def create_refresh_token(subject: str, privlevel: str = "User") -> str:
"""Create a JWT refresh token with longer expiry.""" """Create a JWT refresh token with longer expiry."""
expire = datetime.now(timezone.utc) + timedelta( expire = datetime.now(timezone.utc) + timedelta(
days=settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS days=settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS
@@ -55,6 +59,7 @@ def create_refresh_token(subject: str) -> str:
"exp": expire, "exp": expire,
"iat": datetime.now(timezone.utc), "iat": datetime.now(timezone.utc),
"type": "refresh", "type": "refresh",
"privlevel": privlevel,
} }
return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.JWT_ALGORITHM) return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
@@ -72,14 +77,31 @@ def decode_token(token: str) -> Optional[dict]:
return None return None
def get_current_user(token: str) -> Optional[dict]: def get_current_user(authorization: str = Header(...)) -> dict:
"""Extract user info from JWT token.""" """Extract user info from JWT token in Authorization header."""
# Parse Bearer token
parts = authorization.split()
if len(parts) != 2 or parts[0].lower() != 'bearer':
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication scheme"
)
token = parts[1]
payload = decode_token(token) payload = decode_token(token)
if not payload: if not payload:
return None raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token"
)
if payload.get("type") != "access": if payload.get("type") != "access":
return None raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token type"
)
return { return {
"user_id": payload.get("sub"), "user_id": payload.get("sub"),
"privlevel": payload.get("privlevel", "User"),
"token_type": payload.get("type"), "token_type": payload.get("type"),
} }
+3 -3
View File
@@ -15,7 +15,7 @@ class User(Base):
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True, index=True)
username = Column(String(64), unique=True, nullable=False, index=True) username = Column(String(64), unique=True, nullable=False, index=True)
password_sha512 = Column(String(128), nullable=False) # bcrypt hash password_hash = Column(String(128), nullable=False) # bcrypt hash
salt = Column(String(128), nullable=False) # password salt salt = Column(String(128), nullable=False) # password salt
email = Column(String(255), nullable=True, index=True) email = Column(String(255), nullable=True, index=True)
country = Column(String(2), nullable=True) country = Column(String(2), nullable=True)
@@ -39,7 +39,7 @@ class User(Base):
# Relationships # Relationships
decklist_files = relationship("DecklistFile", back_populates="owner", cascade="all, delete-orphan") decklist_files = relationship("DecklistFile", back_populates="owner", cascade="all, delete-orphan")
decks = relationship("Deck", back_populates="owner", cascade="all, delete-orphan") decklist_folders = relationship("DecklistFolder", back_populates="owner", cascade="all, delete-orphan")
def __repr__(self) -> str: def __repr__(self) -> str:
return f"<User {self.username} (ID: {self.id})>" return f"<User {self.username} (ID: {self.id})>"
@@ -67,7 +67,7 @@ class DecklistFile(Base):
__tablename__ = "cockatrice_decklist_files" __tablename__ = "cockatrice_decklist_files"
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True, index=True)
folder_id = Column(Integer, ForeignKey("cockatrice_decklist_folders.id"), nullable=False) folder_id = Column(Integer, ForeignKey("cockatrice_decklist_folders.id"), nullable=True)
owner_id = Column(Integer, ForeignKey("cockatrice_users.id"), nullable=False) owner_id = Column(Integer, ForeignKey("cockatrice_users.id"), nullable=False)
name = Column(String(255), nullable=False) name = Column(String(255), nullable=False)
content = Column(Text, nullable=False) # Native XML or plain text deck format content = Column(Text, nullable=False) # Native XML or plain text deck format
+9 -2
View File
@@ -184,7 +184,7 @@ async def list_logs(
] ]
@router.post("/audit", response_model=AuditLog) @router.post("/audit", response_model=dict)
async def log_audit( async def log_audit(
action_type: str, action_type: str,
target_user_id: Optional[int] = None, target_user_id: Optional[int] = None,
@@ -210,4 +210,11 @@ async def log_audit(
db.add(new_audit) db.add(new_audit)
await db.flush() await db.flush()
return AuditLog.model_validate(new_audit) return {
"id": new_audit.id,
"admin_id": new_audit.admin_id,
"action_type": new_audit.action_type,
"target_user_id": new_audit.target_user_id,
"details": new_audit.details,
"timestamp": new_audit.timestamp,
}
+7 -5
View File
@@ -4,6 +4,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select from sqlalchemy import select
from typing import Optional from typing import Optional
from datetime import datetime
from app.core.database import get_db from app.core.database import get_db
from app.core.security import ( from app.core.security import (
verify_password, verify_password,
@@ -33,7 +35,7 @@ async def login(request: LoginRequest, db: AsyncSession = Depends(get_db)):
result = await db.execute(stmt) result = await db.execute(stmt)
user = result.scalar_one_or_none() user = result.scalar_one_or_none()
if not user or not verify_password(request.password, user.password_sha512): if not user or not verify_password(request.password, user.password_hash):
raise HTTPException( raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid username or password", detail="Invalid username or password",
@@ -56,8 +58,8 @@ async def login(request: LoginRequest, db: AsyncSession = Depends(get_db)):
await db.flush() await db.flush()
# Generate tokens # Generate tokens
access_token = create_access_token(str(user.id)) access_token = create_access_token(str(user.id), user.privlevel or "User")
refresh_token = create_refresh_token(str(user.id)) refresh_token = create_refresh_token(str(user.id), user.privlevel or "User")
return LoginResponse( return LoginResponse(
access_token=access_token, access_token=access_token,
@@ -90,7 +92,7 @@ async def refresh_token(request: RefreshTokenRequest, db: AsyncSession = Depends
) )
# Generate new access token # Generate new access token
access_token = create_access_token(str(user.id)) access_token = create_access_token(str(user.id), user.privlevel or "User")
return TokenResponse(access_token=access_token) return TokenResponse(access_token=access_token)
@@ -120,7 +122,7 @@ async def register(request: UserCreate, db: AsyncSession = Depends(get_db)):
# Create new user # Create new user
new_user = User( new_user = User(
username=request.username, username=request.username,
password_sha512=hash_password(request.password), password_hash=hash_password(request.password),
salt="random_salt", # In production, generate random salt salt="random_salt", # In production, generate random salt
email=request.email, email=request.email,
country=request.country, country=request.country,
+160 -89
View File
@@ -1,7 +1,7 @@
"""Deck management router endpoints.""" """Deck management router endpoints."""
from fastapi import APIRouter, Depends, HTTPException, status from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, delete from sqlalchemy import select, delete, update
from typing import Optional, List from typing import Optional, List
from app.core.database import get_db from app.core.database import get_db
@@ -12,6 +12,165 @@ from app.schemas.schemas import DeckCreate, DeckUpdate, DeckResponse, FolderCrea
router = APIRouter() router = APIRouter()
@router.get("/", response_model=List[DeckResponse])
async def list_decks(
folder_id: Optional[int] = None,
page: int = 1,
page_size: int = 50,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""List decks for current user."""
user_id = int(current_user["user_id"])
if folder_id:
stmt = (
select(DecklistFile)
.where(
DecklistFile.owner_id == user_id,
DecklistFile.folder_id == folder_id,
)
.offset((page - 1) * page_size)
.limit(page_size)
)
else:
stmt = (
select(DecklistFile)
.where(DecklistFile.owner_id == user_id)
.offset((page - 1) * page_size)
.limit(page_size)
)
result = await db.execute(stmt)
decks = result.scalars().all()
return [DeckResponse.model_validate(deck) for deck in decks]
@router.post("/", response_model=DeckResponse)
async def create_deck(
request: DeckCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""Create a new deck."""
user_id = int(current_user["user_id"])
# Verify folder exists if specified
if request.folder_id:
stmt = select(DecklistFolder).where(
DecklistFolder.id == request.folder_id,
DecklistFolder.owner_id == user_id,
)
result = await db.execute(stmt)
folder = result.scalar_one_or_none()
if not folder:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Folder not found",
)
new_deck = DecklistFile(
owner_id=user_id,
folder_id=request.folder_id,
name=request.name,
content=request.content,
format=request.format,
)
db.add(new_deck)
await db.flush()
return DeckResponse.model_validate(new_deck)
@router.get("/folders", response_model=List[FolderResponse])
async def list_folders(
parent_id: Optional[int] = None,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""List folders for current user."""
user_id = int(current_user["user_id"])
if parent_id:
stmt = select(DecklistFolder).where(
DecklistFolder.parent_id == parent_id,
DecklistFolder.owner_id == user_id,
)
else:
stmt = select(DecklistFolder).where(
DecklistFolder.parent_id == None, # Top-level folders
DecklistFolder.owner_id == user_id,
)
result = await db.execute(stmt)
folders = result.scalars().all()
return [FolderResponse.model_validate(folder) for folder in folders]
@router.post("/folders", response_model=FolderResponse)
async def create_folder(
request: FolderCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""Create a new folder."""
user_id = int(current_user["user_id"])
# Verify parent folder exists if specified
if request.parent_id:
stmt = select(DecklistFolder).where(
DecklistFolder.id == request.parent_id,
DecklistFolder.owner_id == user_id,
)
result = await db.execute(stmt)
parent_folder = result.scalar_one_or_none()
if not parent_folder:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Parent folder not found",
)
new_folder = DecklistFolder(
owner_id=user_id,
name=request.name,
parent_id=request.parent_id,
)
db.add(new_folder)
await db.flush()
return FolderResponse.model_validate(new_folder)
@router.delete("/folders/{folder_id}")
async def delete_folder(
folder_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""Delete folder and all its contents."""
user_id = int(current_user["user_id"])
stmt = select(DecklistFolder).where(
DecklistFolder.id == folder_id,
DecklistFolder.owner_id == user_id,
)
result = await db.execute(stmt)
folder = result.scalar_one_or_none()
if not folder:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Folder not found",
)
# Delete folder and all decks (cascading delete)
await db.execute(delete(DecklistFolder).where(DecklistFolder.id == folder_id))
return {"message": "Folder deleted successfully"}
@router.get("/", response_model=List[DeckResponse]) @router.get("/", response_model=List[DeckResponse])
async def list_decks( async def list_decks(
folder_id: Optional[int] = None, folder_id: Optional[int] = None,
@@ -175,91 +334,3 @@ async def delete_deck(
await db.execute(delete(DecklistFile).where(DecklistFile.id == deck_id)) await db.execute(delete(DecklistFile).where(DecklistFile.id == deck_id))
return {"message": "Deck deleted successfully"} return {"message": "Deck deleted successfully"}
@router.get("/folders", response_model=List[FolderResponse])
async def list_folders(
parent_id: Optional[int] = None,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""List folders for current user."""
user_id = int(current_user["user_id"])
if parent_id:
stmt = select(DecklistFolder).where(
DecklistFolder.parent_id == parent_id,
DecklistFolder.owner_id == user_id,
)
else:
stmt = select(DecklistFolder).where(
DecklistFolder.parent_id == None, # Top-level folders
DecklistFolder.owner_id == user_id,
)
result = await db.execute(stmt)
folders = result.scalars().all()
return [FolderResponse.model_validate(folder) for folder in folders]
@router.post("/folders", response_model=FolderResponse)
async def create_folder(
request: FolderCreate,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""Create a new folder."""
user_id = int(current_user["user_id"])
# Verify parent folder exists if specified
if request.parent_id:
stmt = select(DecklistFolder).where(
DecklistFolder.id == request.parent_id,
DecklistFolder.owner_id == user_id,
)
result = await db.execute(stmt)
parent_folder = result.scalar_one_or_none()
if not parent_folder:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Parent folder not found",
)
new_folder = DecklistFolder(
owner_id=user_id,
name=request.name,
parent_id=request.parent_id,
)
db.add(new_folder)
await db.flush()
return FolderResponse.model_validate(new_folder)
@router.delete("/folders/{folder_id}")
async def delete_folder(
folder_id: int,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""Delete folder and all its contents."""
user_id = int(current_user["user_id"])
stmt = select(DecklistFolder).where(
DecklistFolder.id == folder_id,
DecklistFolder.owner_id == user_id,
)
result = await db.execute(stmt)
folder = result.scalar_one_or_none()
if not folder:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Folder not found",
)
# Delete folder and all decks (cascading delete)
await db.execute(delete(DecklistFolder).where(DecklistFolder.id == folder_id))
return {"message": "Folder deleted successfully"}
+1 -1
View File
@@ -62,7 +62,7 @@ async def update_user(
# Hash new password if provided # Hash new password if provided
if "new_password" in update_data: if "new_password" in update_data:
update_data["password_sha512"] = hash_password(update_data.pop("new_password")) update_data["password_hash"] = hash_password(update_data.pop("new_password"))
# Update user # Update user
stmt = ( stmt = (
-2
View File
@@ -131,8 +131,6 @@ class FolderResponse(BaseModel):
parent_id: Optional[int] parent_id: Optional[int]
owner_id: int owner_id: int
creation_date: datetime creation_date: datetime
children: List["FolderResponse"] = []
files: List[DeckResponse] = []
class Config: class Config:
from_attributes = True from_attributes = True
+2 -2
View File
@@ -236,7 +236,7 @@ async def game_websocket_endpoint(websocket: WebSocket, game_id: int):
}) })
elif message.get("type") == "ping": elif message.get("type") == "ping":
# Respond to ping # Respond to ping
await websocket.send_text(json.dumps({ await room.send_to_player(player_id, {
"type": "pong", "type": "pong",
"timestamp": datetime.now().isoformat(), "timestamp": datetime.now().isoformat(),
})) }))
@@ -294,7 +294,7 @@ async def process_game_command(room: GameRoom, player_id: int, command: dict):
GAME_COMMAND_JUDGE, GAME_COMMAND_JUDGE,
GAME_COMMAND_REVERSE_TURN, GAME_COMMAND_REVERSE_TURN,
]: ]:
await websocket.send_text(json.dumps({ await room.send_to_player(player_id, {
"type": "error", "type": "error",
"message": f"Invalid command type: {cmd_type}", "message": f"Invalid command type: {cmd_type}",
})) }))
+1 -3
View File
@@ -1,6 +1,4 @@
""" # Ruff configuration for linting and formatting
Ruff configuration for linting and formatting.
"""
[tool.ruff] [tool.ruff]
# Target Python version # Target Python version
target-version = "py312" target-version = "py312"
+1
View File
@@ -11,6 +11,7 @@ alembic==1.13.2
# Authentication # Authentication
python-jose[cryptography]==3.3.0 python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4 passlib[bcrypt]==1.7.4
bcrypt==4.0.1
python-multipart==0.0.9 python-multipart==0.0.9
# Validation # Validation
+75 -65
View File
@@ -1,122 +1,132 @@
""" """
Pytest configuration and fixtures for Cockatrice Web Application tests. Test configuration and fixtures.
""" """
import pytest import pytest
import pytest_asyncio import pytest_asyncio
from httpx import AsyncClient, ASGITransport from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool from sqlalchemy.pool import StaticPool
from app.main import app from app.main import app
from app.core.database import Base, get_db from app.core.database import Base, get_db
from app.core.security import hash_password from app.core.security import create_access_token, hash_password
from app.models.models import User from app.models.models import User
# Test database URL (in-memory SQLite for tests) # Use in-memory SQLite for testing
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:" TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
# Create test engine and session
test_engine = create_async_engine( test_engine = create_async_engine(
TEST_DATABASE_URL, TEST_DATABASE_URL,
connect_args={"check_same_thread": False}, connect_args={"check_same_thread": False},
poolclass=StaticPool, poolclass=StaticPool,
) )
TestSessionLocal = async_sessionmaker( TestAsyncSession = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
bind=test_engine,
class_=AsyncSession,
expire_on_commit=False,
)
@pytest.fixture(scope="session") @pytest.fixture(scope="session")
def event_loop(): def anyio_backend():
"""Create an instance of the default event loop for the test session.""" """Use asyncio for tests."""
import asyncio return "asyncio"
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest_asyncio.fixture(scope="function") @pytest_asyncio.fixture(scope="function")
async def db_session(): async def db_session():
"""Create a new database session for each test.""" """Create test database and return session."""
# Create all tables
async with test_engine.begin() as conn: async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) await conn.run_sync(Base.metadata.create_all)
async with TestSessionLocal() as session: # Create session and yield
async with TestAsyncSession() as session:
yield session yield session
# Cleanup
async with test_engine.begin() as conn: async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all) await conn.run_sync(Base.metadata.drop_all)
@pytest_asyncio.fixture(scope="function") @pytest_asyncio.fixture(scope="function")
async def client(db_session): async def client(db_session):
"""Create a test client with database dependency override.""" """Create test client using the same database session."""
async def override_get_db(): async def override_get_db():
try: yield db_session
yield db_session
finally:
pass
app.dependency_overrides[get_db] = override_get_db app.dependency_overrides[get_db] = override_get_db
async with AsyncClient( async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
transport=ASGITransport(app=app),
base_url="http://test",
) as ac:
yield ac yield ac
# Clean up dependency overrides
app.dependency_overrides.clear() app.dependency_overrides.clear()
@pytest_asyncio.fixture(scope="function") @pytest_asyncio.fixture
async def test_user(db_session):
"""Create a test user."""
user = User(
username="testuser",
password_sha512=hash_password("testpassword123"),
salt="test_salt",
email="test@example.com",
country="US",
privlevel="User",
)
db_session.add(user)
await db_session.flush()
return user
@pytest_asyncio.fixture(scope="function")
async def admin_user(db_session): async def admin_user(db_session):
"""Create an admin test user.""" """Create admin test user."""
user = User( user = User(
username="adminuser", username="admin_test",
password_sha512=hash_password("adminpassword123"), email="admin@test.com",
salt="admin_salt", password_hash=hash_password("adminpassword123"),
email="admin@example.com", salt="test_salt",
country="US", is_active=True,
privlevel="Admin", privlevel="Admin",
) )
db_session.add(user) db_session.add(user)
await db_session.flush() await db_session.flush()
await db_session.refresh(user)
return user return user
@pytest_asyncio.fixture(scope="function") @pytest_asyncio.fixture
async def auth_headers(client, test_user): async def regular_user(db_session):
"""Get authentication headers for test user.""" """Create regular test user."""
from app.core.security import create_access_token, create_refresh_token user = User(
username="regular_test",
access_token = create_access_token(str(test_user.id)) email="user@test.com",
return {"Authorization": f"Bearer {access_token}"} password_hash=hash_password("testpassword123"),
salt="test_salt",
is_active=True,
privlevel="User",
)
db_session.add(user)
await db_session.flush()
await db_session.refresh(user)
return user
@pytest_asyncio.fixture(scope="function") @pytest_asyncio.fixture
async def admin_headers(client, admin_user): async def admin_token(admin_user):
"""Get authentication headers for admin user.""" """Create admin access token with privlevel."""
from app.core.security import create_access_token return create_access_token(str(admin_user.id), admin_user.privlevel or "User")
access_token = create_access_token(str(admin_user.id))
return {"Authorization": f"Bearer {access_token}"} @pytest_asyncio.fixture
async def regular_token(regular_user):
"""Create regular user access token with privlevel."""
return create_access_token(str(regular_user.id), regular_user.privlevel or "User")
@pytest_asyncio.fixture
async def admin_headers(admin_token):
"""Get admin auth headers."""
return {"Authorization": f"Bearer {admin_token}"}
@pytest_asyncio.fixture
async def regular_headers(regular_token):
"""Get regular user auth headers."""
return {"Authorization": f"Bearer {regular_token}"}
@pytest_asyncio.fixture
async def test_user(regular_user):
"""Alias for regular_user."""
return regular_user
@pytest_asyncio.fixture
async def auth_headers(regular_headers):
"""Alias for regular_headers."""
return regular_headers
+3
View File
@@ -16,6 +16,9 @@ class TestAdminEndpoints:
async def test_list_users_admin(self, client: AsyncClient, admin_headers: dict): async def test_list_users_admin(self, client: AsyncClient, admin_headers: dict):
"""Test listing users as admin.""" """Test listing users as admin."""
response = await client.get("/api/v1/admin/users", headers=admin_headers) response = await client.get("/api/v1/admin/users", headers=admin_headers)
print(f"\nDEBUG: status={response.status_code}, body={response.text}")
if response.status_code != 200:
print(f"DEBUG: response.json()={response.json()}")
assert response.status_code == 200 assert response.status_code == 200
data = response.json() data = response.json()
assert isinstance(data, list) assert isinstance(data, list)
+8 -8
View File
@@ -18,7 +18,7 @@ class TestAuthentication:
response = await client.post( response = await client.post(
"/api/v1/auth/login", "/api/v1/auth/login",
json={ json={
"username": "testuser", "username": "regular_test",
"password": "testpassword123", "password": "testpassword123",
}, },
) )
@@ -26,7 +26,7 @@ class TestAuthentication:
data = response.json() data = response.json()
assert "access_token" in data assert "access_token" in data
assert "refresh_token" in data assert "refresh_token" in data
assert data["user"]["username"] == "testuser" assert data["user"]["username"] == "regular_test"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_login_invalid_password(self, client: AsyncClient, test_user: User): async def test_login_invalid_password(self, client: AsyncClient, test_user: User):
@@ -34,7 +34,7 @@ class TestAuthentication:
response = await client.post( response = await client.post(
"/api/v1/auth/login", "/api/v1/auth/login",
json={ json={
"username": "testuser", "username": "regular_test",
"password": "wrongpassword", "password": "wrongpassword",
}, },
) )
@@ -77,7 +77,7 @@ class TestAuthentication:
response = await client.post( response = await client.post(
"/api/v1/auth/register", "/api/v1/auth/register",
json={ json={
"username": "testuser", # Already exists "username": "regular_test", # Already exists
"password": "newpassword123", "password": "newpassword123",
"email": "new@example.com", "email": "new@example.com",
}, },
@@ -86,15 +86,15 @@ class TestAuthentication:
assert response.json()["detail"] == "Username already exists" assert response.json()["detail"] == "Username already exists"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_get_current_user(self, client: AsyncClient, auth_headers: dict): async def test_get_current_user(self, client: AsyncClient, regular_token: str):
"""Test getting current user.""" """Test getting current user."""
response = await client.get( response = await client.get(
"/api/v1/auth/me", "/api/v1/auth/me",
headers=auth_headers, params={"token": regular_token},
) )
assert response.status_code == 200 assert response.status_code == 200
data = response.json() data = response.json()
assert data["username"] == "testuser" assert data["username"] == "regular_test"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_refresh_token(self, client: AsyncClient, test_user: User): async def test_refresh_token(self, client: AsyncClient, test_user: User):
@@ -105,7 +105,7 @@ class TestAuthentication:
login_response = await client.post( login_response = await client.post(
"/api/v1/auth/login", "/api/v1/auth/login",
json={ json={
"username": "testuser", "username": "regular_test",
"password": "testpassword123", "password": "testpassword123",
}, },
) )
+1
View File
@@ -141,6 +141,7 @@ class TestDeckManagement:
# List folders # List folders
response = await client.get("/api/v1/decks/folders", headers=auth_headers) response = await client.get("/api/v1/decks/folders", headers=auth_headers)
print(f"\nDEBUG list_folders: status={response.status_code}, body={response.text}")
assert response.status_code == 200 assert response.status_code == 200
data = response.json() data = response.json()
assert isinstance(data, list) assert isinstance(data, list)