---
name: testing-backend
description: "Backend testing: pytest patterns (Python), Jest patterns (Node.js), integration testing, database fixtures/factories, API contract testing. Use when writing or reviewing backend tests: unit, integration, database and API level."
tags: [testing, pytest, jest, integration, database, api-testing, backend]
version: "2025.1"
---

# Backend Testing

## Testing Pyramid for Backend

```
         [Contract Tests]
           (API schema compliance)
      [Integration Tests]
        (DB, services, HTTP endpoints)
   [Unit Tests]
     (business logic, pure functions)
```

## pytest Patterns (Python)

### Project Structure

```
tests/
  conftest.py           # Shared fixtures
  unit/
    test_services.py
    test_models.py
  integration/
    conftest.py         # DB fixtures
    test_api.py
    test_repositories.py
```

### Fixtures and Factories

```python
# tests/conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from app.main import app
from app.database import Base, get_db

TEST_DB_URL = "postgresql+asyncpg://test:test@localhost:5432/test_db"

@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"

@pytest.fixture(scope="session")
async def engine():
    engine = create_async_engine(TEST_DB_URL)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()

@pytest.fixture
async def db_session(engine):
    async with AsyncSession(engine) as session:
        async with session.begin():
            yield session
            await session.rollback()  # Rollback after each test

@pytest.fixture
async def client(db_session):
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        yield client
    app.dependency_overrides.clear()
```

```python
# tests/factories.py
from dataclasses import dataclass, field
import uuid

@dataclass
class UserFactory:
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    name: str = "Test User"
    email: str = field(default_factory=lambda: f"user-{uuid.uuid4().hex[:8]}@test.com")
    role: str = "user"

    def build(self) -> dict:
        return {"id": self.id, "name": self.name, "email": self.email, "role": self.role}

    async def create(self, db: AsyncSession) -> User:
        user = User(**self.build())
        db.add(user)
        await db.flush()
        return user
```

### Unit Tests

```python
# tests/unit/test_services.py
import pytest
from unittest.mock import AsyncMock, MagicMock
from app.services.user_service import UserService
from app.exceptions import UserNotFoundError, DuplicateEmailError

class TestUserService:
    @pytest.fixture
    def mock_repo(self):
        return AsyncMock()

    @pytest.fixture
    def service(self, mock_repo):
        return UserService(repository=mock_repo)

    async def test_get_user_returns_user(self, service, mock_repo):
        expected = {"id": "1", "name": "Alice", "email": "alice@test.com"}
        mock_repo.find_by_id.return_value = expected

        result = await service.get_user("1")

        assert result == expected
        mock_repo.find_by_id.assert_called_once_with("1")

    async def test_get_user_raises_not_found(self, service, mock_repo):
        mock_repo.find_by_id.return_value = None

        with pytest.raises(UserNotFoundError, match="User 999 not found"):
            await service.get_user("999")

    async def test_create_user_validates_unique_email(self, service, mock_repo):
        mock_repo.find_by_email.return_value = {"id": "existing"}

        with pytest.raises(DuplicateEmailError):
            await service.create_user(name="Bob", email="existing@test.com")

    @pytest.mark.parametrize("role,expected_permissions", [
        ("admin", ["read", "write", "delete"]),
        ("user", ["read", "write"]),
        ("viewer", ["read"]),
    ])
    async def test_get_permissions_by_role(self, service, role, expected_permissions):
        result = service.get_permissions(role)
        assert result == expected_permissions
```

### Integration Tests

```python
# tests/integration/test_api.py
import pytest
from httpx import AsyncClient

class TestUserAPI:
    async def test_create_user(self, client: AsyncClient):
        response = await client.post("/api/v1/users", json={
            "name": "Alice",
            "email": "alice@example.com",
            "role": "user",
        })

        assert response.status_code == 201
        data = response.json()["data"]
        assert data["name"] == "Alice"
        assert data["email"] == "alice@example.com"
        assert "id" in data

    async def test_create_user_duplicate_email(self, client: AsyncClient, db_session):
        # Arrange: create existing user
        from tests.factories import UserFactory
        await UserFactory(email="taken@example.com").create(db_session)

        # Act
        response = await client.post("/api/v1/users", json={
            "name": "Bob",
            "email": "taken@example.com",
        })

        # Assert
        assert response.status_code == 409
        assert response.json()["error"]["code"] == "DUPLICATE_EMAIL"

    async def test_list_users_pagination(self, client: AsyncClient, db_session):
        from tests.factories import UserFactory
        for i in range(25):
            await UserFactory(name=f"User {i}").create(db_session)

        response = await client.get("/api/v1/users?per_page=10&page=1")

        assert response.status_code == 200
        body = response.json()
        assert len(body["data"]) == 10
        assert body["meta"]["total"] == 25
        assert body["meta"]["has_next"] is True

    async def test_get_user_not_found(self, client: AsyncClient):
        response = await client.get("/api/v1/users/nonexistent-id")
        assert response.status_code == 404
```

## Jest Patterns (Node.js)

```typescript
// tests/setup.ts
import { PrismaClient } from '@prisma/client';

const prisma = new PrismaClient();

beforeEach(async () => {
  // Clean tables in correct order (respect foreign keys)
  await prisma.$transaction([
    prisma.postTag.deleteMany(),
    prisma.post.deleteMany(),
    prisma.user.deleteMany(),
  ]);
});

afterAll(async () => {
  await prisma.$disconnect();
});

export { prisma };
```

```typescript
// tests/unit/user-service.test.ts
import { UserService } from '@/services/user-service';
import { UserRepository } from '@/repositories/user-repository';

jest.mock('@/repositories/user-repository');

describe('UserService', () => {
  let service: UserService;
  let mockRepo: jest.Mocked<UserRepository>;

  beforeEach(() => {
    mockRepo = new UserRepository() as jest.Mocked<UserRepository>;
    service = new UserService(mockRepo);
  });

  describe('getUserById', () => {
    it('returns user when found', async () => {
      const expected = { id: '1', name: 'Alice', email: 'alice@test.com' };
      mockRepo.findById.mockResolvedValue(expected);

      const result = await service.getUserById('1');

      expect(result).toEqual(expected);
      expect(mockRepo.findById).toHaveBeenCalledWith('1');
    });

    it('throws NotFoundError when user does not exist', async () => {
      mockRepo.findById.mockResolvedValue(null);

      await expect(service.getUserById('999')).rejects.toThrow('User not found');
    });
  });
});
```

```typescript
// tests/integration/user-api.test.ts
import request from 'supertest';
import { app } from '@/app';
import { prisma } from '../setup';

describe('POST /api/v1/users', () => {
  it('creates a user and returns 201', async () => {
    const response = await request(app)
      .post('/api/v1/users')
      .send({ name: 'Alice', email: 'alice@example.com' })
      .expect(201);

    expect(response.body.data).toMatchObject({
      name: 'Alice',
      email: 'alice@example.com',
    });
    expect(response.body.data.id).toBeDefined();

    // Verify in database
    const user = await prisma.user.findUnique({ where: { email: 'alice@example.com' } });
    expect(user).not.toBeNull();
  });

  it('returns 422 for invalid email', async () => {
    const response = await request(app)
      .post('/api/v1/users')
      .send({ name: 'Bob', email: 'invalid' })
      .expect(422);

    expect(response.body.error.details[0].field).toBe('email');
  });
});
```

## Database Testing

```python
# Isolated transactions  -  each test rolls back
@pytest.fixture
async def db_session(engine):
    connection = await engine.connect()
    transaction = await connection.begin()
    session = AsyncSession(bind=connection)

    yield session

    await session.close()
    await transaction.rollback()
    await connection.close()

# Test database state
async def test_user_creation(db_session):
    user = User(name="Alice", email="alice@test.com")
    db_session.add(user)
    await db_session.flush()

    result = await db_session.get(User, user.id)
    assert result.name == "Alice"
    assert result.created_at is not None
```

## API Contract Testing

```typescript
// Using Zod for response contract validation
import { z } from 'zod';

const UserResponseSchema = z.object({
  data: z.object({
    id: z.string().uuid(),
    name: z.string(),
    email: z.string().email(),
    role: z.enum(['admin', 'user', 'viewer']),
    created_at: z.string().datetime(),
  }),
});

describe('User API Contract', () => {
  it('GET /users/:id matches contract', async () => {
    const user = await createTestUser();
    const response = await request(app).get(`/api/v1/users/${user.id}`);

    const result = UserResponseSchema.safeParse(response.body);
    expect(result.success).toBe(true);
  });

  it('GET /users matches list contract', async () => {
    const response = await request(app).get('/api/v1/users');

    const result = UserListResponseSchema.safeParse(response.body);
    expect(result.success).toBe(true);
  });
});
```

## Do's

- Write unit tests for business logic and service layer
- Write integration tests for API endpoints with real database
- Use factories/builders to create test data consistently
- Use transaction rollback to isolate database tests
- Test error paths: validation errors, not found, unauthorized, conflicts
- Use parameterized tests for similar test cases with different inputs
- Mock external services (HTTP, email, queues) but not your own database
- Run integration tests in CI with a real database (Docker/Testcontainers)

## Don'ts

- Do not share state between tests; each test must be independent
- Do not test implementation details (private methods, internal queries)
- Do not use `sleep()` in tests; use async assertions and waiters
- Do not test framework behavior (ORM queries, HTTP routing)
- Do not mock what you own (repositories in integration tests)
- Do not write tests that pass when the code is broken (false positives)
- Do not ignore flaky tests; fix the underlying timing/state issue
- Do not test without a real database in integration tests

## Troubleshooting

| Problem | Cause | Solution |
|---------|-------|----------|
| Tests pass alone, fail together | Shared database state | Use transaction rollback per test |
| Slow test suite | Real I/O in unit tests | Mock external calls in unit tests; reserve DB for integration |
| Flaky async tests | Race conditions or timeouts | Use proper async assertions, increase timeout |
| Factory creates duplicate data | Unique constraints violated | Use random values in factories (UUID, random email) |
| Mock not applied | Wrong import path or scope | Verify mock target matches actual import path |
| DB schema out of sync | Migrations not run in test DB | Run migrations before test suite in CI |