Охватываемые темы: Installation, TestClient (synchronous), Test database, Tests with fixtures.
Installation
pip install pytest httpx pytest-asyncio
TestClient (synchronous)
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_root():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello, World!"}
def test_create_task():
response = client.post("/tasks/", json={"title": "Test Task"})
assert response.status_code == 201
assert response.json()["title"] == "Test Task"
Test database
# conftest.py
import pytest
from sqlmodel import SQLModel, create_engine, Session
from fastapi.testclient import TestClient
TEST_DATABASE_URL = "sqlite:///./test.db"
test_engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False})
@pytest.fixture(autouse=True)
def setup_db():
SQLModel.metadata.create_all(test_engine)
yield
SQLModel.metadata.drop_all(test_engine)
@pytest.fixture
def session():
with Session(test_engine) as session:
yield session
@pytest.fixture
def client(session):
def override_get_session():
yield session
app.dependency_overrides[get_session] = override_get_session
yield TestClient(app)
app.dependency_overrides.clear()
Tests with fixtures
def test_get_task(client, session):
# Create data directly in the database
task = Task(title="Test Task", status="todo")
session.add(task)
session.commit()
response = client.get(f"/tasks/{task.id}")
assert response.status_code == 200
assert response.json()["title"] == "Test Task"
def test_get_task_not_found(client):
response = client.get("/tasks/9999")
assert response.status_code == 404
Authenticated requests
def test_protected_endpoint(client):
# Without token
response = client.get("/tasks/")
assert response.status_code == 401
# With token
token = create_access_token({"sub": "1"})
response = client.get("/tasks/", headers={"Authorization": f"Bearer {token}"})
assert response.status_code == 200
Async tests
import pytest
import httpx
@pytest.mark.asyncio
async def test_async():
async with httpx.AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/tasks/")
assert response.status_code == 200
💬 Comments (0)
No comments yet
Be the first to share your opinion about this article!