Initial commit of mtgonline project
This commit is contained in:
@@ -0,0 +1,122 @@
|
||||
"""
|
||||
Pytest configuration and fixtures for Cockatrice Web Application tests.
|
||||
"""
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from httpx import AsyncClient, ASGITransport
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.main import app
|
||||
from app.core.database import Base, get_db
|
||||
from app.core.security import hash_password
|
||||
from app.models.models import User
|
||||
|
||||
# Test database URL (in-memory SQLite for tests)
|
||||
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
|
||||
|
||||
# Create test engine and session
|
||||
test_engine = create_async_engine(
|
||||
TEST_DATABASE_URL,
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
|
||||
TestSessionLocal = async_sessionmaker(
|
||||
bind=test_engine,
|
||||
class_=AsyncSession,
|
||||
expire_on_commit=False,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def event_loop():
|
||||
"""Create an instance of the default event loop for the test session."""
|
||||
import asyncio
|
||||
loop = asyncio.new_event_loop()
|
||||
yield loop
|
||||
loop.close()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
async def db_session():
|
||||
"""Create a new database session for each test."""
|
||||
async with test_engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
|
||||
async with TestSessionLocal() as session:
|
||||
yield session
|
||||
|
||||
async with test_engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.drop_all)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
async def client(db_session):
|
||||
"""Create a test client with database dependency override."""
|
||||
async def override_get_db():
|
||||
try:
|
||||
yield db_session
|
||||
finally:
|
||||
pass
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app),
|
||||
base_url="http://test",
|
||||
) as ac:
|
||||
yield ac
|
||||
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
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):
|
||||
"""Create an admin test user."""
|
||||
user = User(
|
||||
username="adminuser",
|
||||
password_sha512=hash_password("adminpassword123"),
|
||||
salt="admin_salt",
|
||||
email="admin@example.com",
|
||||
country="US",
|
||||
privlevel="Admin",
|
||||
)
|
||||
db_session.add(user)
|
||||
await db_session.flush()
|
||||
return user
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
async def auth_headers(client, test_user):
|
||||
"""Get authentication headers for test user."""
|
||||
from app.core.security import create_access_token, create_refresh_token
|
||||
|
||||
access_token = create_access_token(str(test_user.id))
|
||||
return {"Authorization": f"Bearer {access_token}"}
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
async def admin_headers(client, admin_user):
|
||||
"""Get authentication 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}"}
|
||||
@@ -0,0 +1,85 @@
|
||||
"""
|
||||
Tests for admin endpoints.
|
||||
"""
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.models import User, Ban
|
||||
from app.core.security import hash_password
|
||||
|
||||
|
||||
class TestAdminEndpoints:
|
||||
"""Test admin endpoints."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
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)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert isinstance(data, list)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_users_non_admin(self, client: AsyncClient, auth_headers: dict):
|
||||
"""Test listing users as non-admin."""
|
||||
response = await client.get("/api/v1/admin/users", headers=auth_headers)
|
||||
assert response.status_code == 403
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_ban(self, client: AsyncClient, admin_headers: dict, test_user: User):
|
||||
"""Test creating a ban."""
|
||||
response = await client.post(
|
||||
"/api/v1/admin/bans",
|
||||
json={
|
||||
"user_id": test_user.id,
|
||||
"reason": "Test ban reason",
|
||||
},
|
||||
headers=admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["reason"] == "Test ban reason"
|
||||
assert data["active"] == True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_bans(self, client: AsyncClient, admin_headers: dict, test_user: User):
|
||||
"""Test listing bans."""
|
||||
# Create a ban first
|
||||
await client.post(
|
||||
"/api/v1/admin/bans",
|
||||
json={
|
||||
"user_id": test_user.id,
|
||||
"reason": "Test ban",
|
||||
},
|
||||
headers=admin_headers,
|
||||
)
|
||||
|
||||
# List bans
|
||||
response = await client.get("/api/v1/admin/bans", headers=admin_headers)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert isinstance(data, list)
|
||||
assert len(data) > 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unban_user(self, client: AsyncClient, admin_headers: dict, test_user: User):
|
||||
"""Test unbanning a user."""
|
||||
# Create a ban first
|
||||
ban_response = await client.post(
|
||||
"/api/v1/admin/bans",
|
||||
json={
|
||||
"user_id": test_user.id,
|
||||
"reason": "Test ban",
|
||||
},
|
||||
headers=admin_headers,
|
||||
)
|
||||
ban_id = ban_response.json()["id"]
|
||||
|
||||
# Unban user
|
||||
response = await client.post(
|
||||
f"/api/v1/admin/bans/{ban_id}/unban",
|
||||
headers=admin_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["message"] == "User unbanned successfully"
|
||||
@@ -0,0 +1,123 @@
|
||||
"""
|
||||
Tests for authentication endpoints.
|
||||
"""
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.models import User
|
||||
from app.core.security import hash_password, create_access_token
|
||||
|
||||
|
||||
class TestAuthentication:
|
||||
"""Test authentication endpoints."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_success(self, client: AsyncClient, test_user: User):
|
||||
"""Test successful login."""
|
||||
response = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
json={
|
||||
"username": "testuser",
|
||||
"password": "testpassword123",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "access_token" in data
|
||||
assert "refresh_token" in data
|
||||
assert data["user"]["username"] == "testuser"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_invalid_password(self, client: AsyncClient, test_user: User):
|
||||
"""Test login with invalid password."""
|
||||
response = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
json={
|
||||
"username": "testuser",
|
||||
"password": "wrongpassword",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 401
|
||||
assert response.json()["detail"] == "Invalid username or password"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_nonexistent_user(self, client: AsyncClient):
|
||||
"""Test login with non-existent user."""
|
||||
response = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
json={
|
||||
"username": "nonexistent",
|
||||
"password": "password123",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 401
|
||||
assert response.json()["detail"] == "Invalid username or password"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_register_success(self, client: AsyncClient):
|
||||
"""Test successful user registration."""
|
||||
response = await client.post(
|
||||
"/api/v1/auth/register",
|
||||
json={
|
||||
"username": "newuser",
|
||||
"password": "newpassword123",
|
||||
"email": "new@example.com",
|
||||
"country": "US",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["username"] == "newuser"
|
||||
assert data["email"] == "new@example.com"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_register_duplicate_username(self, client: AsyncClient, test_user: User):
|
||||
"""Test registration with duplicate username."""
|
||||
response = await client.post(
|
||||
"/api/v1/auth/register",
|
||||
json={
|
||||
"username": "testuser", # Already exists
|
||||
"password": "newpassword123",
|
||||
"email": "new@example.com",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 409
|
||||
assert response.json()["detail"] == "Username already exists"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_current_user(self, client: AsyncClient, auth_headers: dict):
|
||||
"""Test getting current user."""
|
||||
response = await client.get(
|
||||
"/api/v1/auth/me",
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["username"] == "testuser"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refresh_token(self, client: AsyncClient, test_user: User):
|
||||
"""Test refreshing access token."""
|
||||
from app.core.security import create_refresh_token
|
||||
|
||||
# First login to get refresh token
|
||||
login_response = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
json={
|
||||
"username": "testuser",
|
||||
"password": "testpassword123",
|
||||
},
|
||||
)
|
||||
refresh_token = login_response.json()["refresh_token"]
|
||||
|
||||
# Refresh token
|
||||
response = await client.post(
|
||||
"/api/v1/auth/refresh",
|
||||
json={
|
||||
"refresh_token": refresh_token,
|
||||
},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "access_token" in data
|
||||
@@ -0,0 +1,162 @@
|
||||
"""
|
||||
Tests for deck management endpoints.
|
||||
"""
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.models import DecklistFile, DecklistFolder
|
||||
from app.core.security import hash_password
|
||||
|
||||
|
||||
class TestDeckManagement:
|
||||
"""Test deck management endpoints."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_deck(self, client: AsyncClient, auth_headers: dict):
|
||||
"""Test creating a new deck."""
|
||||
response = await client.post(
|
||||
"/api/v1/decks/",
|
||||
json={
|
||||
"name": "Test Deck",
|
||||
"content": "4 Lightning Bolt\n2 Searing Blaze",
|
||||
"format": "plain",
|
||||
},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["name"] == "Test Deck"
|
||||
assert data["format"] == "plain"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_decks(self, client: AsyncClient, auth_headers: dict):
|
||||
"""Test listing decks."""
|
||||
# Create a deck first
|
||||
await client.post(
|
||||
"/api/v1/decks/",
|
||||
json={
|
||||
"name": "List Test Deck",
|
||||
"content": "4 Lightning Bolt",
|
||||
"format": "plain",
|
||||
},
|
||||
headers=auth_headers,
|
||||
)
|
||||
|
||||
# List decks
|
||||
response = await client.get("/api/v1/decks/", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert isinstance(data, list)
|
||||
assert len(data) > 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_deck(self, client: AsyncClient, auth_headers: dict):
|
||||
"""Test getting a specific deck."""
|
||||
# Create a deck first
|
||||
create_response = await client.post(
|
||||
"/api/v1/decks/",
|
||||
json={
|
||||
"name": "Get Test Deck",
|
||||
"content": "4 Lightning Bolt",
|
||||
"format": "plain",
|
||||
},
|
||||
headers=auth_headers,
|
||||
)
|
||||
deck_id = create_response.json()["id"]
|
||||
|
||||
# Get deck
|
||||
response = await client.get(f"/api/v1/decks/{deck_id}", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["name"] == "Get Test Deck"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_deck(self, client: AsyncClient, auth_headers: dict):
|
||||
"""Test updating a deck."""
|
||||
# Create a deck first
|
||||
create_response = await client.post(
|
||||
"/api/v1/decks/",
|
||||
json={
|
||||
"name": "Update Test Deck",
|
||||
"content": "4 Lightning Bolt",
|
||||
"format": "plain",
|
||||
},
|
||||
headers=auth_headers,
|
||||
)
|
||||
deck_id = create_response.json()["id"]
|
||||
|
||||
# Update deck
|
||||
response = await client.patch(
|
||||
f"/api/v1/decks/{deck_id}",
|
||||
json={"name": "Updated Deck Name"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["name"] == "Updated Deck Name"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_deck(self, client: AsyncClient, auth_headers: dict):
|
||||
"""Test deleting a deck."""
|
||||
# Create a deck first
|
||||
create_response = await client.post(
|
||||
"/api/v1/decks/",
|
||||
json={
|
||||
"name": "Delete Test Deck",
|
||||
"content": "4 Lightning Bolt",
|
||||
"format": "plain",
|
||||
},
|
||||
headers=auth_headers,
|
||||
)
|
||||
deck_id = create_response.json()["id"]
|
||||
|
||||
# Delete deck
|
||||
response = await client.delete(f"/api/v1/decks/{deck_id}", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_folder(self, client: AsyncClient, auth_headers: dict):
|
||||
"""Test creating a folder."""
|
||||
response = await client.post(
|
||||
"/api/v1/decks/folders",
|
||||
json={
|
||||
"name": "Test Folder",
|
||||
},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["name"] == "Test Folder"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_folders(self, client: AsyncClient, auth_headers: dict):
|
||||
"""Test listing folders."""
|
||||
# Create a folder first
|
||||
await client.post(
|
||||
"/api/v1/decks/folders",
|
||||
json={"name": "List Test Folder"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
|
||||
# List folders
|
||||
response = await client.get("/api/v1/decks/folders", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert isinstance(data, list)
|
||||
assert len(data) > 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_folder(self, client: AsyncClient, auth_headers: dict):
|
||||
"""Test deleting a folder."""
|
||||
# Create a folder first
|
||||
create_response = await client.post(
|
||||
"/api/v1/decks/folders",
|
||||
json={"name": "Delete Test Folder"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
folder_id = create_response.json()["id"]
|
||||
|
||||
# Delete folder
|
||||
response = await client.delete(f"/api/v1/decks/folders/{folder_id}", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
Reference in New Issue
Block a user