3861b027b2
Some checks failed
K8S Fission Deployment / Deployment fission functions (push) Failing after 12s
435 lines
11 KiB
Markdown
435 lines
11 KiB
Markdown
# API Testing
|
|
|
|
## Table of Contents
|
|
- [FastAPI Testing](#fastapi-testing)
|
|
- [Flask Testing](#flask-testing)
|
|
- [Test Factories](#test-factories)
|
|
- [Authentication Testing](#authentication-testing)
|
|
- [Database Testing](#database-testing)
|
|
|
|
## FastAPI Testing
|
|
|
|
### Setup
|
|
|
|
```python
|
|
# tests/conftest.py
|
|
import pytest
|
|
from httpx import AsyncClient, ASGITransport
|
|
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
|
|
from sqlmodel import SQLModel
|
|
|
|
from app.main import app
|
|
from app.core.database import get_session
|
|
|
|
TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"
|
|
|
|
test_engine = create_async_engine(TEST_DATABASE_URL)
|
|
TestSession = async_sessionmaker(test_engine, class_=AsyncSession)
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
async def db_session():
|
|
"""Fresh database for each test."""
|
|
async with test_engine.begin() as conn:
|
|
await conn.run_sync(SQLModel.metadata.create_all)
|
|
|
|
async with TestSession() as session:
|
|
yield session
|
|
|
|
async with test_engine.begin() as conn:
|
|
await conn.run_sync(SQLModel.metadata.drop_all)
|
|
|
|
|
|
@pytest.fixture
|
|
async def client(db_session):
|
|
"""Test client with dependency override."""
|
|
async def override_session():
|
|
yield db_session
|
|
|
|
app.dependency_overrides[get_session] = override_session
|
|
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app),
|
|
base_url="http://test",
|
|
) as ac:
|
|
yield ac
|
|
|
|
app.dependency_overrides.clear()
|
|
```
|
|
|
|
### Basic Endpoint Tests
|
|
|
|
```python
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
class TestUsers:
|
|
@pytest.mark.asyncio
|
|
async def test_create_user(self, client: AsyncClient):
|
|
response = await client.post(
|
|
"/api/users",
|
|
json={"email": "test@example.com", "name": "Test User"},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["email"] == "test@example.com"
|
|
assert "id" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user(self, client: AsyncClient, test_user):
|
|
response = await client.get(f"/api/users/{test_user.id}")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["id"] == test_user.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_not_found(self, client: AsyncClient):
|
|
response = await client.get("/api/users/99999")
|
|
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_user(self, client: AsyncClient, test_user, auth_headers):
|
|
response = await client.patch(
|
|
f"/api/users/{test_user.id}",
|
|
json={"name": "Updated Name"},
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["name"] == "Updated Name"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_user(self, client: AsyncClient, test_user, admin_headers):
|
|
response = await client.delete(
|
|
f"/api/users/{test_user.id}",
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 204
|
|
```
|
|
|
|
### Testing Query Parameters
|
|
|
|
```python
|
|
@pytest.mark.asyncio
|
|
async def test_list_users_pagination(self, client: AsyncClient, auth_headers):
|
|
response = await client.get(
|
|
"/api/users",
|
|
params={"page": 1, "per_page": 10, "sort": "-created_at"},
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "items" in data
|
|
assert "total" in data
|
|
assert data["page"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_users(self, client: AsyncClient, auth_headers):
|
|
response = await client.get(
|
|
"/api/users",
|
|
params={"q": "john", "active": True},
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
```
|
|
|
|
### Testing File Uploads
|
|
|
|
```python
|
|
import io
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_file(self, client: AsyncClient, auth_headers):
|
|
file_content = b"Hello, World!"
|
|
files = {"file": ("test.txt", io.BytesIO(file_content), "text/plain")}
|
|
|
|
response = await client.post(
|
|
"/api/files/upload",
|
|
files=files,
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert response.json()["filename"] == "test.txt"
|
|
```
|
|
|
|
## Flask Testing
|
|
|
|
### Setup
|
|
|
|
```python
|
|
import pytest
|
|
from app import create_app
|
|
from app.extensions import db
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
"""Create application for testing."""
|
|
app = create_app("testing")
|
|
app.config["TESTING"] = True
|
|
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
|
|
|
|
with app.app_context():
|
|
db.create_all()
|
|
yield app
|
|
db.drop_all()
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
"""Test client."""
|
|
return app.test_client()
|
|
|
|
@pytest.fixture
|
|
def runner(app):
|
|
"""CLI test runner."""
|
|
return app.test_cli_runner()
|
|
```
|
|
|
|
### Flask Test Examples
|
|
|
|
```python
|
|
def test_create_user(client):
|
|
response = client.post(
|
|
"/api/users",
|
|
json={"email": "test@example.com", "name": "Test"},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
|
|
def test_get_user(client, test_user):
|
|
response = client.get(f"/api/users/{test_user.id}")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json["email"] == test_user.email
|
|
|
|
def test_protected_route(client, auth_token):
|
|
response = client.get(
|
|
"/api/protected",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
```
|
|
|
|
## Test Factories
|
|
|
|
### Using factory_boy
|
|
|
|
```python
|
|
# tests/factories.py
|
|
import factory
|
|
from faker import Faker
|
|
from app.models import User, Project
|
|
|
|
fake = Faker()
|
|
|
|
|
|
class UserFactory(factory.Factory):
|
|
class Meta:
|
|
model = User
|
|
|
|
email = factory.LazyAttribute(lambda _: fake.email())
|
|
name = factory.LazyAttribute(lambda _: fake.name())
|
|
hashed_password = "hashed_password"
|
|
is_active = True
|
|
|
|
|
|
class ProjectFactory(factory.Factory):
|
|
class Meta:
|
|
model = Project
|
|
|
|
name = factory.LazyAttribute(lambda _: fake.company())
|
|
description = factory.LazyAttribute(lambda _: fake.sentence())
|
|
owner_id = None
|
|
|
|
|
|
# Async helper
|
|
async def create_user(session, **kwargs) -> User:
|
|
user = UserFactory(**kwargs)
|
|
session.add(user)
|
|
await session.commit()
|
|
await session.refresh(user)
|
|
return user
|
|
|
|
|
|
async def create_users(session, count: int = 5, **kwargs) -> list[User]:
|
|
users = [UserFactory(**kwargs) for _ in range(count)]
|
|
session.add_all(users)
|
|
await session.commit()
|
|
return users
|
|
```
|
|
|
|
### Using Factories in Tests
|
|
|
|
```python
|
|
from tests.factories import create_user, create_users
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_users(client, db_session, auth_headers):
|
|
# Create 20 users
|
|
await create_users(db_session, count=20)
|
|
|
|
response = await client.get("/api/users", headers=auth_headers)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["total"] >= 20
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filter_active_users(client, db_session, auth_headers):
|
|
await create_user(db_session, is_active=True)
|
|
await create_user(db_session, is_active=False)
|
|
|
|
response = await client.get(
|
|
"/api/users",
|
|
params={"active": True},
|
|
headers=auth_headers,
|
|
)
|
|
|
|
data = response.json()
|
|
assert all(u["is_active"] for u in data["items"])
|
|
```
|
|
|
|
## Authentication Testing
|
|
|
|
### Auth Fixtures
|
|
|
|
```python
|
|
from app.core.security import create_access_token, hash_password
|
|
|
|
@pytest.fixture
|
|
async def test_user(db_session):
|
|
"""Create a regular test user."""
|
|
user = User(
|
|
email="test@example.com",
|
|
hashed_password=hash_password("password123"),
|
|
is_active=True,
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
return user
|
|
|
|
@pytest.fixture
|
|
async def admin_user(db_session):
|
|
"""Create an admin user."""
|
|
user = User(
|
|
email="admin@example.com",
|
|
hashed_password=hash_password("admin123"),
|
|
is_active=True,
|
|
is_admin=True,
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
return user
|
|
|
|
@pytest.fixture
|
|
def auth_headers(test_user):
|
|
"""Headers with valid auth token."""
|
|
token = create_access_token({"sub": str(test_user.id)})
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
@pytest.fixture
|
|
def admin_headers(admin_user):
|
|
"""Headers with admin auth token."""
|
|
token = create_access_token({"sub": str(admin_user.id)})
|
|
return {"Authorization": f"Bearer {token}"}
|
|
```
|
|
|
|
### Auth Test Examples
|
|
|
|
```python
|
|
class TestAuth:
|
|
@pytest.mark.asyncio
|
|
async def test_login_success(self, client, test_user):
|
|
response = await client.post(
|
|
"/api/auth/login",
|
|
data={"username": test_user.email, "password": "password123"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "access_token" in response.json()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_wrong_password(self, client, test_user):
|
|
response = await client.post(
|
|
"/api/auth/login",
|
|
data={"username": test_user.email, "password": "wrong"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_protected_endpoint_no_auth(self, client):
|
|
response = await client.get("/api/protected")
|
|
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_endpoint_regular_user(self, client, auth_headers):
|
|
response = await client.delete(
|
|
"/api/admin/users/1",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
```
|
|
|
|
## Database Testing
|
|
|
|
### Transaction Rollback Pattern
|
|
|
|
```python
|
|
@pytest.fixture
|
|
async def db_session(engine):
|
|
"""Each test runs in a transaction that rolls back."""
|
|
async with engine.connect() as conn:
|
|
await conn.begin()
|
|
|
|
async with AsyncSession(bind=conn) as session:
|
|
yield session
|
|
|
|
await conn.rollback()
|
|
```
|
|
|
|
### Testing Repository Layer
|
|
|
|
```python
|
|
from app.repositories import UserRepository
|
|
|
|
class TestUserRepository:
|
|
@pytest.mark.asyncio
|
|
async def test_create(self, db_session):
|
|
repo = UserRepository(db_session)
|
|
|
|
user = await repo.create({
|
|
"email": "test@example.com",
|
|
"hashed_password": "hash",
|
|
})
|
|
|
|
assert user.id is not None
|
|
assert user.email == "test@example.com"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_by_email(self, db_session, test_user):
|
|
repo = UserRepository(db_session)
|
|
|
|
user = await repo.get_by_email(test_user.email)
|
|
|
|
assert user.id == test_user.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_soft_delete(self, db_session, test_user):
|
|
repo = UserRepository(db_session)
|
|
|
|
await repo.soft_delete(test_user.id)
|
|
|
|
user = await repo.get_by_id(test_user.id)
|
|
assert user.status == 0 # Soft deleted
|
|
```
|