admin/mtgonline
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix backend test suite - all 20 tests passing

Browse Source 
- 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
This commit is contained in:
akadmin
2026-07-18 16:58:39 +00:00
parent 312ac27f88
commit 167a352d44
10 changed files with 295 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/core/security.py
+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 typing import Optional
from fastapi import Header, HTTPException, status
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.settings import get_settings
@@ -26,6 +28,7 @@ def hash_password(password: str) -> str:
def create_access_token(
subject: str,
privlevel: str = "User",
expires_delta: Optional[timedelta] = None,
) -> str:
"""Create a JWT access token."""
@@ -41,11 +44,12 @@ def create_access_token(
"exp": expire,
"iat": datetime.now(timezone.utc),
"type": "access",
"privlevel": privlevel,
}
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."""
expire = datetime.now(timezone.utc) + timedelta(
days=settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS
@@ -55,6 +59,7 @@ def create_refresh_token(subject: str) -> str:
"exp": expire,
"iat": datetime.now(timezone.utc),
"type": "refresh",
"privlevel": privlevel,
}
return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
@@ -72,14 +77,31 @@ def decode_token(token: str) -> Optional[dict]:
return None
def get_current_user(token: str) -> Optional[dict]:
"""Extract user info from JWT token."""
def get_current_user(authorization: str = Header(...)) -> dict:
"""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)
if not payload:
return None
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token"
)
if payload.get("type") != "access":
return None
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token type"
)
return {
"user_id": payload.get("sub"),
"privlevel": payload.get("privlevel", "User"),
"token_type": payload.get("type"),
}
backend/app/models/models.py
+2 -2
View File
@@ -39,7 +39,7 @@ class User(Base):
# Relationships
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:
return f"<User {self.username} (ID: {self.id})>"
@@ -67,7 +67,7 @@ class DecklistFile(Base):
__tablename__ = "cockatrice_decklist_files"
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)
name = Column(String(255), nullable=False)
content = Column(Text, nullable=False) # Native XML or plain text deck format
backend/app/routers/auth.py
+3 -3
View File
@@ -58,8 +58,8 @@ async def login(request: LoginRequest, db: AsyncSession = Depends(get_db)):
await db.flush()
# Generate tokens
access_token = create_access_token(str(user.id))
refresh_token = create_refresh_token(str(user.id))
access_token = create_access_token(str(user.id), user.privlevel or "User")
refresh_token = create_refresh_token(str(user.id), user.privlevel or "User")
return LoginResponse(
access_token=access_token,
@@ -92,7 +92,7 @@ async def refresh_token(request: RefreshTokenRequest, db: AsyncSession = Depends
)
# 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)
backend/app/routers/decks.py
+160 -89
View File
@@ -1,7 +1,7 @@
"""Deck management router endpoints."""
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, delete
from sqlalchemy import select, delete, update
from typing import Optional, List
from app.core.database import get_db
@@ -12,6 +12,165 @@ from app.schemas.schemas import DeckCreate, DeckUpdate, DeckResponse, FolderCrea
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])
async def list_decks(
folder_id: Optional[int] = None,
@@ -175,91 +334,3 @@ async def delete_deck(
await db.execute(delete(DecklistFile).where(DecklistFile.id == deck_id))
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"}
backend/app/schemas/schemas.py
-2
View File
@@ -131,8 +131,6 @@ class FolderResponse(BaseModel):
parent_id: Optional[int]
owner_id: int
creation_date: datetime
children: List["FolderResponse"] = []
files: List[DeckResponse] = []
class Config:
from_attributes = True
backend/requirements.txt
+1
View File
@@ -11,6 +11,7 @@ alembic==1.13.2
# Authentication
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
bcrypt==4.0.1
python-multipart==0.0.9
# Validation
backend/tests/conftest.py
+87 -74
View File
@@ -1,119 +1,132 @@
"""Test configuration and fixtures."""
"""
Test configuration and fixtures.
"""
import pytest
import asyncio
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.pool import StaticPool
from app.main import app
from app.core.database import get_db, Base
from app.core.database import Base, get_db
from app.core.security import create_access_token, hash_password
from app.models.models import User
# Test database URL (use SQLite for testing)
TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"
# Use in-memory SQLite for testing
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
# Create test engine and session factory
test_engine = create_async_engine(TEST_DATABASE_URL, echo=False)
TestAsyncSessionLocal = async_sessionmaker(
test_engine,
class_=AsyncSession,
expire_on_commit=False,
test_engine = create_async_engine(
TEST_DATABASE_URL,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
TestAsyncSession = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
@pytest.fixture(scope="session")
def anyio_backend():
"""Specify the backend for anyio (pytest-asyncio)."""
"""Use asyncio for tests."""
return "asyncio"
@pytest.fixture(scope="session")
async def database():
"""Create and drop database tables for testing."""
@pytest_asyncio.fixture(scope="function")
async def db_session():
"""Create test database and return session."""
# Create all tables
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
# Create session and yield
async with TestAsyncSession() as session:
yield session
# Cleanup
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest.fixture
async def db_session(database):
"""Create a new database session for each test."""
async with TestAsyncSessionLocal() as session:
yield session
@pytest.fixture
@pytest_asyncio.fixture(scope="function")
async def client(db_session):
"""Create a test client with database session."""
"""Create test client using the same database session."""
async def override_get_db():
yield db_session
app.dependency_overrides[get_db] = override_get_db
async with AsyncClient(app=app, base_url="http://test") as ac:
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
yield ac
# Clean up dependency overrides
app.dependency_overrides.clear()
@pytest.fixture
async def test_user(db_session):
"""Create a test user."""
from app.models.models import User
from app.core.security import hash_password
user = User(
username="testuser",
password_hash=hash_password("testpassword123"),
email="test@example.com",
is_active=True,
)
db_session.add(user)
await db_session.commit()
await db_session.refresh(user)
return user
@pytest.fixture
@pytest_asyncio.fixture
async def admin_user(db_session):
"""Create an admin test user."""
from app.models.models import User
from app.core.security import hash_password
"""Create admin test user."""
user = User(
username="admin",
username="admin_test",
email="admin@test.com",
password_hash=hash_password("adminpassword123"),
email="admin@example.com",
privlevel="Admin",
salt="test_salt",
is_active=True,
privlevel="Admin",
)
db_session.add(user)
await db_session.commit()
await db_session.flush()
await db_session.refresh(user)
return user
@pytest.fixture
def auth_headers(test_user):
"""Get auth headers for test user."""
from app.core.security import create_access_token
access_token = create_access_token(str(test_user.id))
return {"Authorization": f"Bearer {access_token}"}
@pytest_asyncio.fixture
async def regular_user(db_session):
"""Create regular test user."""
user = User(
username="regular_test",
email="user@test.com",
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.fixture
def admin_headers(admin_user):
"""Get auth headers for admin user."""
from app.core.security import create_access_token
access_token = create_access_token(str(admin_user.id))
return {"Authorization": f"Bearer {access_token}"}
@pytest_asyncio.fixture
async def admin_token(admin_user):
"""Create admin access token with privlevel."""
return create_access_token(str(admin_user.id), admin_user.privlevel or "User")
@pytest.fixture
def refresh_token(test_user):
"""Get a refresh token for test user."""
from app.core.security import create_refresh_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")
return create_refresh_token(str(test_user.id))
@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
backend/tests/test_admin.py
+3
View File
@@ -16,6 +16,9 @@ class TestAdminEndpoints:
async def test_list_users_admin(self, client: AsyncClient, admin_headers: dict):
"""Test listing users as admin."""
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
data = response.json()
assert isinstance(data, list)
backend/tests/test_auth.py
+8 -8
View File
@@ -18,7 +18,7 @@ class TestAuthentication:
response = await client.post(
"/api/v1/auth/login",
json={
"username": "testuser",
"username": "regular_test",
"password": "testpassword123",
},
)
@@ -26,7 +26,7 @@ class TestAuthentication:
data = response.json()
assert "access_token" in data
assert "refresh_token" in data
assert data["user"]["username"] == "testuser"
assert data["user"]["username"] == "regular_test"
@pytest.mark.asyncio
async def test_login_invalid_password(self, client: AsyncClient, test_user: User):
@@ -34,7 +34,7 @@ class TestAuthentication:
response = await client.post(
"/api/v1/auth/login",
json={
"username": "testuser",
"username": "regular_test",
"password": "wrongpassword",
},
)
@@ -77,7 +77,7 @@ class TestAuthentication:
response = await client.post(
"/api/v1/auth/register",
json={
"username": "testuser", # Already exists
"username": "regular_test", # Already exists
"password": "newpassword123",
"email": "new@example.com",
},
@@ -86,15 +86,15 @@ class TestAuthentication:
assert response.json()["detail"] == "Username already exists"
@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."""
response = await client.get(
"/api/v1/auth/me",
headers=auth_headers,
params={"token": regular_token},
)
assert response.status_code == 200
data = response.json()
assert data["username"] == "testuser"
assert data["username"] == "regular_test"
@pytest.mark.asyncio
async def test_refresh_token(self, client: AsyncClient, test_user: User):
@@ -105,7 +105,7 @@ class TestAuthentication:
login_response = await client.post(
"/api/v1/auth/login",
json={
"username": "testuser",
"username": "regular_test",
"password": "testpassword123",
},
)
backend/tests/test_decks.py
+1
View File
@@ -141,6 +141,7 @@ class TestDeckManagement:
# List folders
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
data = response.json()
assert isinstance(data, list)