Database Testing

ORM Testing Patterns: How to Test SQLAlchemy, Prisma, and TypeORM Correctly

HelpMeTest

HelpMeTest

— 7 min read
ORM Testing Patterns: How to Test SQLAlchemy, Prisma, and TypeORM Correctly

ORMs hide SQL behind object abstractions, which makes application code cleaner but creates a testing trap: it's tempting to mock the ORM and test nothing real. Mock-based ORM tests can show 100% coverage while completely missing query logic bugs, N+1 problems, and constraint violations that only appear with a real database.

This guide covers ORM testing patterns that actually catch bugs, with examples for SQLAlchemy (Python), Prisma (TypeScript/Node.js), and TypeORM (TypeScript).

The Core Problem with Mocking ORMs

Consider a typical repository method:

class OrderRepository:
    def get_orders_with_items(self, customer_id: int):
        return (
            db.session.query(Order)
            .options(joinedload(Order.items))
            .filter(Order.customer_id == customer_id)
            .filter(Order.status != 'cancelled')
            .order_by(Order.created_at.desc())
            .all()
        )

A mock-based test:

def test_get_orders_with_items_mock():
    mock_session = MagicMock()
    mock_session.query.return_value.options.return_value.filter.return_value\
        .filter.return_value.order_by.return_value.all.return_value = [
            Order(id=1, customer_id=42, status='completed')
        ]
    
    repo = OrderRepository(mock_session)
    result = repo.get_orders_with_items(42)
    
    assert len(result) == 1  # ✓ passes

This test passes but doesn't verify:

  • Whether the query actually filters cancelled orders
  • Whether the joinedload prevents N+1 queries
  • Whether the ORDER BY is applied correctly
  • Whether the SQL generated is valid

The mock is just testing that you called the mock's methods in the right order.

Pattern 1: Real Database, Transaction Rollback

The most reliable ORM testing pattern uses a real database with transaction rollback to isolate tests:

SQLAlchemy

# conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from your_app.models import Base

@pytest.fixture(scope="session")
def engine():
    return create_engine("postgresql://test:test@localhost/testdb")

@pytest.fixture(scope="session")
def tables(engine):
    Base.metadata.create_all(engine)
    yield
    Base.metadata.drop_all(engine)

@pytest.fixture
def db_session(engine, tables):
    """Each test gets a transaction that rolls back at the end."""
    connection = engine.connect()
    transaction = connection.begin()
    session = sessionmaker(bind=connection)()
    
    # Make nested transactions work (for code that uses session.begin())
    session.begin_nested()
    
    yield session
    
    session.close()
    transaction.rollback()
    connection.close()
# tests/test_order_repository.py
from your_app.repositories import OrderRepository
from your_app.models import Customer, Order, OrderItem

def test_get_orders_excludes_cancelled(db_session):
    """Cancelled orders are not returned."""
    customer = Customer(id=1, name="Test Customer")
    db_session.add(customer)
    
    db_session.add_all([
        Order(id=1, customer_id=1, status="completed"),
        Order(id=2, customer_id=1, status="cancelled"),
        Order(id=3, customer_id=1, status="pending"),
    ])
    db_session.flush()
    
    repo = OrderRepository(db_session)
    results = repo.get_orders_with_items(customer_id=1)
    
    result_ids = {o.id for o in results}
    assert result_ids == {1, 3}, "Cancelled order should be excluded"
    assert 2 not in result_ids

def test_get_orders_sorted_by_created_at_desc(db_session):
    """Orders are returned newest first."""
    customer = Customer(id=1, name="Test Customer")
    db_session.add(customer)
    
    from datetime import datetime, timedelta
    base = datetime(2024, 1, 1)
    db_session.add_all([
        Order(id=1, customer_id=1, status="completed", created_at=base),
        Order(id=2, customer_id=1, status="completed", created_at=base + timedelta(days=1)),
        Order(id=3, customer_id=1, status="completed", created_at=base + timedelta(days=2)),
    ])
    db_session.flush()
    
    repo = OrderRepository(db_session)
    results = repo.get_orders_with_items(customer_id=1)
    
    assert [o.id for o in results] == [3, 2, 1], "Orders should be newest first"

def test_get_orders_for_other_customer_returns_empty(db_session):
    """Querying for customer with no orders returns empty list."""
    repo = OrderRepository(db_session)
    results = repo.get_orders_with_items(customer_id=99999)
    assert results == []

Prisma (TypeScript/Node.js)

// tests/setup.ts
import { PrismaClient } from '@prisma/client';
import { execSync } from 'child_process';

const prisma = new PrismaClient({
  datasources: {
    db: { url: process.env.TEST_DATABASE_URL },
  },
});

beforeAll(async () => {
  // Apply migrations to test database
  execSync('npx prisma migrate deploy', {
    env: { ...process.env, DATABASE_URL: process.env.TEST_DATABASE_URL },
  });
});

afterEach(async () => {
  // Clean up in reverse FK order
  await prisma.orderItem.deleteMany();
  await prisma.order.deleteMany();
  await prisma.customer.deleteMany();
});

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

export { prisma };
// tests/order-repository.test.ts
import { prisma } from './setup';
import { OrderRepository } from '../src/repositories/order-repository';

describe('OrderRepository', () => {
  let repo: OrderRepository;

  beforeEach(() => {
    repo = new OrderRepository(prisma);
  });

  it('excludes cancelled orders', async () => {
    const customer = await prisma.customer.create({
      data: { id: 1, name: 'Test Customer', email: 'test@example.com' },
    });

    await prisma.order.createMany({
      data: [
        { id: 1, customerId: customer.id, status: 'completed', total: 100 },
        { id: 2, customerId: customer.id, status: 'cancelled', total: 50 },
        { id: 3, customerId: customer.id, status: 'pending', total: 75 },
      ],
    });

    const orders = await repo.getOrdersWithItems(customer.id);
    const ids = orders.map(o => o.id);

    expect(ids).toContain(1);
    expect(ids).toContain(3);
    expect(ids).not.toContain(2);
  });

  it('loads items with orders (no N+1)', async () => {
    const customer = await prisma.customer.create({
      data: { id: 1, name: 'Test', email: 'test@example.com' },
    });

    const order = await prisma.order.create({
      data: {
        customerId: customer.id,
        status: 'completed',
        total: 200,
        items: {
          create: [
            { productId: 'p1', quantity: 2, price: 50 },
            { productId: 'p2', quantity: 1, price: 100 },
          ],
        },
      },
    });

    const orders = await repo.getOrdersWithItems(customer.id);

    expect(orders[0].items).toHaveLength(2);
    expect(orders[0].items.map(i => i.productId)).toEqual(
      expect.arrayContaining(['p1', 'p2'])
    );
  });
});

Pattern 2: Testing N+1 Query Detection

N+1 queries are the most common ORM performance bug. Test for them explicitly:

def test_no_n_plus_1_when_loading_orders_with_items(db_session):
    """Loading 10 orders with items should issue exactly 2 queries (1 + 1 join)."""
    customer = Customer(id=1, name="Test")
    db_session.add(customer)
    
    for i in range(10):
        order = Order(id=i+1, customer_id=1, status="completed")
        order.items = [
            OrderItem(product_id=f"p{i}", quantity=1, price=10.0)
        ]
        db_session.add(order)
    db_session.flush()
    
    query_count = 0
    
    def count_queries(conn, cursor, statement, parameters, context, executemany):
        nonlocal query_count
        query_count += 1
    
    from sqlalchemy import event
    event.listen(db_session.bind, "before_cursor_execute", count_queries)
    
    repo = OrderRepository(db_session)
    orders = repo.get_orders_with_items(customer_id=1)
    
    # Touch all relationships to trigger any lazy loads
    for order in orders:
        _ = order.items
    
    event.remove(db_session.bind, "before_cursor_execute", count_queries)
    
    assert query_count <= 2, \
        f"N+1 detected: {query_count} queries for 10 orders (expected ≤2)"
    assert len(orders) == 10

Pattern 3: Testing ORM Transactions

Test that your repository correctly uses transactions for operations that must be atomic:

def test_create_order_with_items_is_atomic(db_session):
    """If item creation fails, the order is also rolled back."""
    customer = Customer(id=1, name="Test")
    db_session.add(customer)
    db_session.flush()
    
    class FakeOrderItem:
        """Simulates an item that will cause a DB error."""
        def __init__(self):
            self.price = "not-a-number"  # Invalid type
    
    repo = OrderRepository(db_session)
    
    with pytest.raises(Exception):
        repo.create_order_with_items(
            customer_id=1,
            items=[FakeOrderItem()]
        )
    
    # Order should not exist (transaction rolled back)
    order_count = db_session.query(Order).filter_by(customer_id=1).count()
    assert order_count == 0, "Failed order creation should not leave orphaned order"

Pattern 4: Testing Query Builders

Some ORMs let you build dynamic queries. Test the query builder logic separately:

class OrderQueryBuilder:
    def __init__(self, session):
        self.session = session
        self._query = session.query(Order)
    
    def for_customer(self, customer_id: int):
        self._query = self._query.filter(Order.customer_id == customer_id)
        return self
    
    def with_status(self, *statuses: str):
        self._query = self._query.filter(Order.status.in_(statuses))
        return self
    
    def created_after(self, date):
        self._query = self._query.filter(Order.created_at >= date)
        return self
    
    def build(self):
        return self._query.all()


def test_query_builder_chaining(db_session):
    """Query builder correctly combines multiple filters."""
    # Setup
    db_session.add_all([
        Order(id=1, customer_id=1, status="completed", created_at=datetime(2024, 1, 1)),
        Order(id=2, customer_id=1, status="pending",   created_at=datetime(2024, 2, 1)),
        Order(id=3, customer_id=1, status="completed", created_at=datetime(2024, 3, 1)),
        Order(id=4, customer_id=2, status="completed", created_at=datetime(2024, 3, 1)),
    ])
    db_session.flush()
    
    # Query: customer 1, completed, after Feb 1
    results = (
        OrderQueryBuilder(db_session)
        .for_customer(1)
        .with_status("completed")
        .created_after(datetime(2024, 2, 1))
        .build()
    )
    
    assert len(results) == 1
    assert results[0].id == 3

Pattern 5: TypeORM Repository Testing

// tests/user-repository.test.ts
import { DataSource } from 'typeorm';
import { UserRepository } from '../src/repositories/user-repository';
import { User } from '../src/entities/user.entity';

describe('UserRepository', () => {
  let dataSource: DataSource;
  let userRepo: UserRepository;

  beforeAll(async () => {
    dataSource = new DataSource({
      type: 'postgres',
      url: process.env.TEST_DATABASE_URL,
      entities: [User],
      synchronize: true, // Use migrations in production, sync for tests
      logging: false,
    });
    await dataSource.initialize();
  });

  afterAll(async () => {
    await dataSource.destroy();
  });

  beforeEach(async () => {
    await dataSource.getRepository(User).clear();
  });

  it('findActiveByEmail returns only active users', async () => {
    userRepo = new UserRepository(dataSource);

    await dataSource.getRepository(User).save([
      { email: 'active@test.com', name: 'Active User', isActive: true },
      { email: 'inactive@test.com', name: 'Inactive User', isActive: false },
    ]);

    const result = await userRepo.findActiveByEmail('active@test.com');

    expect(result).not.toBeNull();
    expect(result!.email).toBe('active@test.com');
  });

  it('findActiveByEmail returns null for inactive user', async () => {
    userRepo = new UserRepository(dataSource);

    await dataSource.getRepository(User).save({
      email: 'inactive@test.com',
      name: 'Inactive User',
      isActive: false,
    });

    const result = await userRepo.findActiveByEmail('inactive@test.com');
    expect(result).toBeNull();
  });
});

When Mocking IS Appropriate

Mocking has its place in ORM testing:

Use real database for:

  • Repository/DAO layer tests (these should always hit a real DB)
  • Integration tests that test query logic
  • Tests that verify constraint behavior

Mocks are acceptable for:

  • Application service layer tests where you're testing business logic, not query logic
  • Tests that would require complex data setup unrelated to what you're testing
  • Unit tests of query builders (test the builder output, not the DB execution)
# OK: Testing business logic in a service, not query logic
def test_checkout_service_sends_confirmation_email(mock_order_repo, mock_email_service):
    """Checkout service sends email after successful order creation."""
    mock_order_repo.create.return_value = Order(id=1, total=100.00)
    
    service = CheckoutService(mock_order_repo, mock_email_service)
    service.checkout(customer_id=1, cart=[{"product_id": "p1", "qty": 1}])
    
    mock_email_service.send_confirmation.assert_called_once_with(
        order_id=1,
        customer_id=1
    )

CI/CD Setup

# .github/workflows/orm-tests.yml
name: ORM Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:16-alpine
        env:
          POSTGRES_DB: testdb
          POSTGRES_USER: test
          POSTGRES_PASSWORD: test
        ports:
          - 5432:5432
        options: --health-cmd pg_isready --health-interval 10s --health-retries 5

    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - run: pip install pytest sqlalchemy psycopg2-binary alembic

      - name: Run migrations
        run: alembic upgrade head
        env:
          DATABASE_URL: postgresql://test:test@localhost/testdb

      - name: Run ORM tests
        run: pytest tests/ -v --tb=short
        env:
          DATABASE_URL: postgresql://test:test@localhost/testdb

Common ORM Testing Mistakes

Mocking the session/client: Testing db.session.query() calls on a mock tests nothing about your query logic.

Not testing N+1: Eager loading relationships is easy to forget. Add query count assertions for any query that loads related models.

Assuming test isolation: Tests that don't clean up after themselves create ordering dependencies. Use transactions with rollback or explicit cleanup.

Testing ORM internals: Don't test that SQLAlchemy generates the SQL you expect. Test the behavior — the rows returned, constraints enforced, counts correct.

No FK constraint testing: ORMs make it easy to skip FK validation. Test that your constraints actually reject bad data.

ORM testing is only useful if it uses a real database. The transaction rollback pattern makes real-database tests as fast as mock tests for most test suites — a full suite of 200 repository tests typically runs in under 30 seconds against a local Postgres instance. The payoff is tests that actually catch ORM bugs, query logic errors, and N+1 problems before they reach production.

Read more

Start now free