Testing FastAPI with pytest

pip install pytest httpx pytest-asyncio

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"

# 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()

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

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

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