Testy integracyjne API w FastAPI i Django – praktyczny przewodnik

Czym są testy integracyjne?

Testy integracyjne (integration tests) sprawdzają współpracę wielu komponentów systemu. W kontekście aplikacji webowych oznaczają testowanie:

• Całych endpointów API – od żądania HTTP do odpowiedzi
• Integracji z bazą danych – zapisywanie i odczytywanie danych
• Autoryzacji i autentykacji – logowanie, tokeny, uprawnienia
• Middleware'u – przetwarzanie żądań i odpowiedzi
• Zewnętrznych serwisów – API, systemy płatności, emaile

Różnica między testami jednostkowymi a integracyjnymi:

• Testy jednostkowe – testują jedną funkcję/metodę w izolacji (często z mockami)
• Testy integracyjne – testują współpracę wielu komponentów razem

Zanim zagłębisz się w testy integracyjne, upewnij się, że znasz podstawy testowania. Sprawdź wprowadzenie do unittest lub testowanie z pytest.

Testy integracyjne w FastAPI

FastAPI jest zbudowany na Starlette i ma doskonałe wsparcie dla testowania. Framework udostępnia TestClient, który pozwala testować aplikację bez uruchamiania serwera.

Podstawowa struktura aplikacji FastAPI

Zacznijmy od prostej aplikacji FastAPI:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# main.py
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from typing import List

app = FastAPI()
security = HTTPBearer()

class Item(BaseModel):
id: int
name: str
price: float

# Przykładowa "baza danych" w pamięci
items_db: List[Item] = [
Item(id=1, name="Laptop", price=2999.99),
Item(id=2, name="Mysz", price=89.99),
]

@app.get("/items", response_model=List[Item])
async def get_items():
    return items_db

@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: int):
    item = next((item for item in items_db if item.id == item_id), None)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.post("/items", response_model=Item)
async def create_item(item: Item):
    if any(i.id == item.id for i in items_db):
        raise HTTPException(status_code=400, detail="Item already exists")
    items_db.append(item)
    return item

Podstawowe testy z TestClient

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
# test_api_basic.py - podstawowe testy API
from fastapi.testclient import TestClient
from main import app
import pytest

client = TestClient(app)

def test_get_items():
    response = client.get("/items")
    assert response.status_code == 200
    data = response.json()
    assert isinstance(data, list)
    assert len(data) == 2

def test_get_item_success():
    response = client.get("/items/1")
    assert response.status_code == 200
    data = response.json()
    assert data["id"] == 1
    assert data["name"] == "Laptop"

def test_get_item_not_found():
    response = client.get("/items/999")
    assert response.status_code == 404
    assert "not found" in response.json()["detail"].lower()

def test_create_item():
    new_item = {
        "id": 3,
        "name": "Klawiatura",
        "price": 299.99
    }
    response = client.post("/items", json=new_item)
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Klawiatura"

    # Weryfikacja, że item został dodany
    response = client.get("/items")
    assert len(response.json()) == 3

Uruchomienie testów:

Bash
Kopiuj
1
pytest test_api.py -v

Testowanie POST, PUT, DELETE

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
def test_create_item_duplicate_id():
    """Test, że nie można utworzyć itemu z istniejącym ID"""
    new_item = {"id": 1, "name": "Duplikat", "price": 100.0}
    response = client.post("/items", json=new_item)
    assert response.status_code == 400

def test_delete_item():
    """Usuwanie itemu (zakładając, że mamy endpoint DELETE)"""
    # Najpierw tworzymy item
    new_item = {"id": 100, "name": "Test", "price": 50.0}
    client.post("/items", json=new_item)

    # Potem usuwamy
    response = client.delete("/items/100")
    assert response.status_code == 200

    # Weryfikacja, że został usunięty
    response = client.get("/items/100")
    assert response.status_code == 404

Izolacja testów z fixture'ami pytest

Problem z powyższymi testami: modyfikują one wspólną bazę danych items_db, więc testy mogą wpływać na siebie. Użyjmy fixture'ów do izolacji:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# test_api_fixtures.py - testy z fixture'ami
from fastapi.testclient import TestClient
from main import app, items_db
import pytest

@pytest.fixture
def client():
    return TestClient(app)

@pytest.fixture
def clean_db():
    """Czyści bazę danych przed i po teście"""
    original_items = items_db.copy()
    yield
    items_db.clear()
    items_db.extend(original_items)

def test_get_items(client, clean_db):
    response = client.get("/items")
    assert response.status_code == 200

def test_create_item(client, clean_db):
    new_item = {"id": 999, "name": "Test", "price": 100.0}
    response = client.post("/items", json=new_item)
    assert response.status_code == 200

Testowanie z bazą danych (SQLAlchemy)

W prawdziwych aplikacjach używasz bazy danych. Oto przykład z SQLAlchemy:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
# main_db.py
from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class ItemDB(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Float)

Base.metadata.create_all(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app = FastAPI()

@app.get("/items/{item_id}")
async def get_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(ItemDB).filter(ItemDB.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404)
    return {"id": item.id, "name": item.name, "price": item.price}

Testy z izolowaną bazą danych:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
# test_db.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from main_db import app, Base, get_db

# Testowa baza danych w pamięci
SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

@pytest.fixture(scope="function")
def db_session():
    Base.metadata.create_all(bind=engine)
    yield
    Base.metadata.drop_all(bind=engine)

def test_get_item(db_session):
    client = TestClient(app)
    # Najpierw musimy utworzyć item w bazie
    # (zakładając, że mamy endpoint POST)
    response = client.post("/items", json={"id": 1, "name": "Test", "price": 100.0})

    # Teraz możemy go pobrać
    response = client.get("/items/1")
    assert response.status_code == 200
    assert response.json()["name"] == "Test"

Testowanie autoryzacji w FastAPI

Wiele endpointów wymaga autoryzacji. Oto przykład:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# main_auth.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt

app = FastAPI()
security = HTTPBearer()

SECRET_KEY = "your-secret-key"

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    try:
        payload = jwt.decode(credentials.credentials, SECRET_KEY)
        return payload
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials"
        )

@app.get("/protected")
async def protected_route(token_data: dict = Depends(verify_token)):
    return {"message": "Access granted", "user": token_data.get("sub")}

Testy autoryzacji:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# test_auth.py
from fastapi.testclient import TestClient
from jose import jwt
from main_auth import app, SECRET_KEY
import pytest

client = TestClient(app)

def test_protected_route_no_token():
    """Test bez tokena"""
    response = client.get("/protected")
    assert response.status_code == 403

def test_protected_route_invalid_token():
    """Test z nieprawidłowym tokenem"""
    response = client.get("/protected", headers={"Authorization": "Bearer invalid_token"})
    assert response.status_code == 401

def test_protected_route_valid_token():
    """Test z prawidłowym tokenem"""
    token = jwt.encode({"sub": "test_user"}, SECRET_KEY, algorithm="HS256")
    response = client.get("/protected", headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 200
    assert response.json()["user"] == "test_user"

Testy integracyjne w Django REST Framework

Django REST Framework (DRF) ma własne narzędzia do testowania. Używa się APIClient zamiast standardowego Client Django.

Podstawowa struktura projektu Django

Zacznijmy od prostego modelu i ViewSet:

Python
Kopiuj
1
2
3
4
5
6
7
8
# models.py
from django.db import models

class Book(models.Model):
title = models.CharField(max_length=200)
author = models.CharField(max_length=100)
price = models.DecimalField(max_digits=10, decimal_places=2)
created_at = models.DateTimeField(auto_now_add=True)

Python
Kopiuj
1
2
3
4
5
6
7
8
# serializers.py
from rest_framework import serializers
from .models import Book

class BookSerializer(serializers.ModelSerializer):
    class Meta:
        model = Book
        fields = ['id', 'title', 'author', 'price', 'created_at']

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# views.py
from rest_framework import viewsets
from rest_framework.decorators import action
from rest_framework.response import Response
from .models import Book
from .serializers import BookSerializer

class BookViewSet(viewsets.ModelViewSet):
    queryset = Book.objects.all()
    serializer_class = BookSerializer

    @action(detail=True, methods=['get'])
    def details(self, request, pk=None):
        book = self.get_object()
        return Response({
            'title': book.title,
            'author': book.author,
            'price': float(book.price)
        })

Podstawowe testy z APIClient

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
# tests.py
from django.test import TestCase
from rest_framework.test import APIClient
from rest_framework import status
from .models import Book

class BookAPITestCase(TestCase):
    def setUp(self):
        """Przygotowanie danych przed każdym testem"""
        self.client = APIClient()
        self.book = Book.objects.create(
            title="Python dla początkujących",
            author="Jan Kowalski",
            price=59.99
        )

    def test_list_books(self):
        """Test listowania książek"""
        response = self.client.get('/api/books/')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(len(response.data), 1)
        self.assertEqual(response.data[0]['title'], "Python dla początkujących")

    def test_get_book_detail(self):
        """Test pobierania szczegółów książki"""
        response = self.client.get(f'/api/books/{self.book.id}/')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(response.data['title'], self.book.title)

    def test_create_book(self):
        """Test tworzenia nowej książki"""
        data = {
            'title': 'Django REST Framework',
            'author': 'Anna Nowak',
            'price': '79.99'
        }
        response = self.client.post('/api/books/', data, format='json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        self.assertEqual(Book.objects.count(), 2)
        self.assertEqual(Book.objects.get(id=response.data['id']).title, 'Django REST Framework')

    def test_update_book(self):
        """Test aktualizacji książki"""
        data = {'title': 'Zaktualizowany tytuł', 'author': self.book.author, 'price': str(self.book.price)}
        response = self.client.put(f'/api/books/{self.book.id}/', data, format='json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.book.refresh_from_db()
        self.assertEqual(self.book.title, 'Zaktualizowany tytuł')

    def test_delete_book(self):
        """Test usuwania książki"""
        response = self.client.delete(f'/api/books/{self.book.id}/')
        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
        self.assertEqual(Book.objects.count(), 0)

Testy z pytest i pytest-django

Django można również testować z pytest używając pytest-django:

Bash
Kopiuj
1
pip install pytest-django

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# conftest.py
import pytest

@pytest.fixture
def api_client():
    from rest_framework.test import APIClient
    return APIClient()

@pytest.fixture
def book(db):
    from books.models import Book
    return Book.objects.create(
        title="Test Book",
        author="Test Author",
        price=50.00
    )

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# test_books.py
import pytest
from rest_framework import status

@pytest.mark.django_db
def test_list_books(api_client, book):
    response = api_client.get('/api/books/')
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data) == 1

@pytest.mark.django_db
def test_create_book(api_client):
    data = {
        'title': 'New Book',
        'author': 'New Author',
        'price': '99.99'
    }
    response = api_client.post('/api/books/', data, format='json')
    assert response.status_code == status.HTTP_201_CREATED

Testowanie autoryzacji w Django

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
# tests_auth.py
from django.contrib.auth.models import User
from rest_framework.test import APIClient
from rest_framework import status
from django.test import TestCase

class BookAuthTestCase(TestCase):
    def setUp(self):
        self.client = APIClient()
        self.user = User.objects.create_user(
            username='testuser',
            password='testpass123'
        )
        self.book = Book.objects.create(
            title="Protected Book",
            author="Author",
            price=50.00
        )

    def test_unauthenticated_access(self):
        """Test dostępu bez autoryzacji"""
        response = self.client.get('/api/books/')
        # Jeśli endpoint wymaga autoryzacji:
        # self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
        # W przeciwnym razie:
        self.assertEqual(response.status_code, status.HTTP_200_OK)

    def test_authenticated_access(self):
        """Test dostępu z autoryzacją"""
        self.client.force_authenticate(user=self.user)
        response = self.client.get('/api/books/')
        self.assertEqual(response.status_code, status.HTTP_200_OK)

    def test_token_authentication(self):
        """Test z tokenem JWT (jeśli używasz djangorestframework-simplejwt)"""
        from rest_framework_simplejwt.tokens import RefreshToken
        token = RefreshToken.for_user(self.user)
        self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.access_token}')
        response = self.client.get('/api/books/')
        self.assertEqual(response.status_code, status.HTTP_200_OK)

Testowanie niestandardowych akcji w ViewSet

Python
Kopiuj
1
2
3
4
5
6
def test_custom_action(self):
    """Test niestandardowej akcji 'details'"""
    response = self.client.get(f'/api/books/{self.book.id}/details/')
    self.assertEqual(response.status_code, status.HTTP_200_OK)
    self.assertIn('title', response.data)
    self.assertIn('author', response.data)

Wzorce i najlepsze praktyki

1. Izolacja testów

Każdy test powinien być niezależny. Używaj setUp/tearDown lub fixture'ów do czyszczenia stanu:

Python
Kopiuj
1
2
3
4
5
@pytest.fixture(autouse=True)
def clean_database(db):
    """Automatycznie czyści bazę przed każdym testem"""
    yield
    Book.objects.all().delete()

2. Używaj factory pattern dla danych testowych

Zamiast tworzenia obiektów ręcznie w każdym teście, użyj factory (np. factory_boy):

Bash
Kopiuj
1
pip install factory_boy

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
# factories.py
import factory
from .models import Book

class BookFactory(factory.django.DjangoModelFactory):
    class Meta:
        model = Book

    title = factory.Sequence(lambda n: f"Book {n}")
    author = "Test Author"
    price = 50.00

Python
Kopiuj
1
2
3
4
5
# tests.py
def test_multiple_books():
    books = BookFactory.create_batch(5)
    response = client.get('/api/books/')
    assert len(response.data) == 5

3. Testuj edge cases

Nie testuj tylko "szczęśliwej ścieżki". Sprawdzaj także:

• Błędne dane wejściowe
• Brakujące pola
• Nieprawidłowe formaty
• Graniczne wartości
• Duplikaty
• Brak uprawnień

Python
Kopiuj
1
2
3
4
5
6
def test_create_book_invalid_data():
    """Test z nieprawidłowymi danymi"""
    data = {'title': ''}  # Brak wymaganych pól
    response = client.post('/api/books/', data, format='json')
    assert response.status_code == 400
    assert 'author' in response.data  # Błąd walidacji

4. Sprawdzaj status codes

Używaj konkretnych kodów statusu HTTP, nie tylko 200:

Python
Kopiuj
1
2
3
4
5
assert response.status_code == status.HTTP_201_CREATED  # Tworzenie
assert response.status_code == status.HTTP_204_NO_CONTENT  # Usuwanie
assert response.status_code == status.HTTP_400_BAD_REQUEST  # Błąd walidacji
assert response.status_code == status.HTTP_404_NOT_FOUND  # Nie znaleziono
assert response.status_code == status.HTTP_401_UNAUTHORIZED  # Brak autoryzacji

5. Sprawdzaj strukturę odpowiedzi

Nie tylko status, ale też dane:

Python
Kopiuj
1
2
3
4
5
6
7
8
def test_response_structure():
    response = client.get('/api/books/1/')
    assert response.status_code == 200
    data = response.json()
    assert 'id' in data
    assert 'title' in data
    assert 'author' in data
    assert isinstance(data['price'], (int, float))

Podsumowanie

Testy integracyjne są kluczowe dla zapewnienia, że wszystkie komponenty aplikacji działają razem poprawnie. Kluczowe punkty:

• ✅ Używaj TestClient w FastAPI i APIClient w Django
• ✅ Izoluj testy używając fixture'ów lub setUp/tearDown
• ✅ Testuj autoryzację i uprawnienia
• ✅ Testuj zarówno sukces, jak i błędy
• ✅ Sprawdzaj kod statusu HTTP i strukturę odpowiedzi
• ✅ Używaj factory pattern dla danych testowych

Co dalej?

Teraz, gdy znasz podstawy testów integracyjnych, możesz pogłębić wiedzę:

• Mockowanie i fixture'y w testach Python – naucz się izolować testy od zewnętrznych zależności
• Testowanie z pytest – wykorzystaj pełny potencjał pytest w testach integracyjnych
• Automatyzacja testów w GitHub Actions – uruchamiaj testy automatycznie przy każdym commicie

Pamiętaj: testy integracyjne powinny być szybkie, ale jednocześnie weryfikować rzeczywistą współpracę komponentów. Znajdź równowagę między szybkością a szczegółowością!