KAHIBARO
Discord Login Register

19.8. Testing Databases

Understanding Database Testing

When your backend code talks to a database, many bugs appear at that boundary. Queries might be wrong, constraints might fail, or performance might be terrible. Database testing focuses on checking that:

You will not test the database engine itself, but how your code uses it.

There are two broad strategies:

Both are useful. This chapter focuses on concrete patterns for testing with a database, and how to keep such tests reliable and fast.

Important rule: Business logic that depends heavily on the database must be covered by integration tests that talk to a real database, not only by mocked tests.

Types of Database Tests

Unit tests with mocked database access

Sometimes you want to test pure business logic without touching a real database. For example, you may have a function that:

You can test this by mocking the repository, so you control what it returns.

Example in Python with a fake repository:

python
class FakeUserRepo:
    def __init__(self, users):
        self._users = users
    def get_by_email(self, email):
        return next((u for u in self._users if u["email"] == email), None)
def can_user_place_order(user_repo, email):
    user = user_repo.get_by_email(email)
    if user is None:
        return False
    return user["is_active"] and not user["is_banned"]
def test_can_user_place_order_active_user():
    repo = FakeUserRepo([
        {"email": "alice@example.com", "is_active": True, "is_banned": False}
    ])
    assert can_user_place_order(repo, "alice@example.com") is True

Here, the repository is an in‑memory object. No database is involved. This is fast and reliable, but it does not test any SQL or schema.

Use this style to test business rules, not database integration.

Integration tests against a real database

Integration tests check that:

For this you need a real database instance. In this course you often use PostgreSQL.

Example structure of a test:

  1. Start a test database (or use a dedicated test database).
  2. Apply the database schema (migrations).
  3. Begin a transaction.
  4. Insert some test data.
  5. Run your application code that talks to the DB.
  6. Assert results.
  7. Roll back or drop the database.

We will look at how to organize this in practice.

Choosing a Database Strategy for Tests

There are three common setups.

StrategyDescriptionProsCons
Shared test DBOne database for all testsSimple to startTests can interfere with each other
Transaction per testOne DB, but each test runs in a transactionFast, isolated stateMore setup complexity
Ephemeral DB per test suiteNew database per run or per test classVery isolated, realisticSlower, requires automation (e.g. Docker)

For most backend projects:

Using a Separate Test Database

Never run tests against your production database.

You should have at least:

In configuration, choose the database based on an environment variable, such as APP_ENV=test.

Example connection URLs:

In Python (pseudocode):

python
import os
ENV = os.getenv("APP_ENV", "development")
if ENV == "test":
    DATABASE_URL = "postgresql://test_user:test_pass@localhost/myapp_test"
else:
    DATABASE_URL = "postgresql://dev_user:dev_pass@localhost/myapp_dev"

Important rule: Use a separate database or schema for tests, and never point tests to production or development data.

Preparing the Schema for Tests

Applying migrations

If you use migrations (for example Alembic with SQLAlchemy), your test database must have the latest schema.

Common workflow:

  1. Drop and recreate the test database (optional but common in CI).
  2. Run all migrations to bring the schema up to date.
  3. Run tests.

Example command sequence:

bash
# Pseudocode, depends on your tooling
dropdb myapp_test || true
createdb myapp_test
alembic upgrade head
pytest

This way, your test database structure matches production as closely as possible.

Resetting data between tests

You need each test to see a clean state. There are two main techniques:

  1. Truncate all tables between tests.
  2. Wrap each test in a transaction and roll back after the test.

The transaction pattern is usually faster and safer.

Using Transactions in Tests

The basic idea:

This works because a rollback undoes every insert, update, and delete made during the test.

Example with raw SQL (conceptual)

Imagine a test framework that lets you run setup and teardown:

python
def setup_function():
    global conn
    conn = psycopg2.connect(TEST_DATABASE_URL)
    conn.autocommit = False  # explicit transaction
    global cur
    cur = conn.cursor()
def teardown_function():
    conn.rollback()
    cur.close()
    conn.close()
def test_insert_user():
    cur.execute(
        "INSERT INTO users (email, is_active) VALUES (%s, %s)",
        ("alice@example.com", True),
    )
    cur.execute("SELECT email, is_active FROM users WHERE email = %s", ("alice@example.com",))
    row = cur.fetchone()
    assert row == ("alice@example.com", True)

After teardown_function runs, the rollback removes the inserted user.

Example pattern with SQLAlchemy and pytest

In SQLAlchemy, you typically use sessions and engines. A common test setup:

python
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from myapp.database import Base  # SQLAlchemy models Base
from myapp.database import get_db  # dependency used in your app
TEST_DATABASE_URL = "postgresql://test_user:test_pass@localhost/myapp_test"
engine = create_engine(TEST_DATABASE_URL)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture(scope="session")
def setup_database():
    # Ensure schema is created; or run migrations separately
    Base.metadata.create_all(bind=engine)
    yield
    Base.metadata.drop_all(bind=engine)
@pytest.fixture
def db_session(setup_database):
    connection = engine.connect()
    transaction = connection.begin()
    session = TestingSessionLocal(bind=connection)
    try:
        yield session
    finally:
        session.close()
        transaction.rollback()
        connection.close()

Now any test that receives db_session can interact with the database, and all changes will be rolled back after the test.

Example test:

python
from myapp.models import User
def test_create_user(db_session):
    user = User(email="alice@example.com", is_active=True)
    db_session.add(user)
    db_session.commit()
    db_session.refresh(user)
    assert user.id is not None
    user_from_db = db_session.query(User).filter_by(email="alice@example.com").first()
    assert user_from_db is not None

After the test, the rollback in the fixture cleans up all data.

Seeding Test Data

Some tests need specific data already present. For example, you might need:

Instead of repeating inserts in every test, you can create helper functions or fixtures that seed data.

Seed functions

python
def create_user(db, email, is_admin=False):
    user = User(email=email, is_admin=is_admin)
    db.add(user)
    db.commit()
    db.refresh(user)
    return user
def create_product(db, name, price):
    product = Product(name=name, price=price)
    db.add(product)
    db.commit()
    db.refresh(product)
    return product

Then in your tests:

python
def test_admin_can_see_all_orders(db_session):
    admin = create_user(db_session, "admin@example.com", is_admin=True)
    user = create_user(db_session, "bob@example.com")
    product = create_product(db_session, "Book", 10.0)
    # ... create orders owned by user and others ...
    orders = order_service.list_orders_for_admin(db_session, admin_id=admin.id)
    assert len(orders) >= 1

Because each test runs inside a transaction, the seed data is removed after the test, but each test starts with a clean state.

Testing Queries and ORM Logic

When using an ORM, you may write complex query logic that combines filters, joins, and aggregations. These are good candidates for database tests.

Example: filtering active users

Repository function:

python
def list_active_users(db):
    return db.query(User).filter(User.is_active.is_(True)).all()

Test:

python
def test_list_active_users_returns_only_active(db_session):
    inactive = User(email="inactive@example.com", is_active=False)
    active1 = User(email="active1@example.com", is_active=True)
    active2 = User(email="active2@example.com", is_active=True)
    db_session.add_all([inactive, active1, active2])
    db_session.commit()
    active_users = list_active_users(db_session)
    emails = {u.email for u in active_users}
    assert emails == {"active1@example.com", "active2@example.com"}

You test the actual SQL generated by the ORM and ensure it behaves correctly with the database.

Example: pagination

Imagine a function:

python
def list_users_paginated(db, offset: int, limit: int):
    return (
        db.query(User)
        .order_by(User.id)
        .offset(offset)
        .limit(limit)
        .all()
    )

Test both the number of results and ordering:

python
def test_list_users_paginated_returns_correct_slice(db_session):
    for i in range(10):
        db_session.add(User(email=f"user{i}@example.com", is_active=True))
    db_session.commit()
    page = list_users_paginated(db_session, offset=3, limit=4)
    assert len(page) == 4
    assert page[0].email == "user3@example.com"
    assert page[-1].email == "user6@example.com"

Testing Constraints and Transactions

Databases protect your data with constraints and transaction semantics. You should test that your code handles these properly.

Testing unique constraints

Suppose your users.email must be unique.

In SQLAlchemy:

python
from sqlalchemy.exc import IntegrityError
def test_cannot_create_two_users_with_same_email(db_session):
    user1 = User(email="alice@example.com", is_active=True)
    db_session.add(user1)
    db_session.commit()
    user2 = User(email="alice@example.com", is_active=True)
    db_session.add(user2)
    try:
        db_session.commit()
        assert False, "Expected IntegrityError"
    except IntegrityError:
        db_session.rollback()

You verify that the database rejects invalid state and that your code correctly handles IntegrityError.

Testing foreign keys

If orders.user_id references users.id, then you cannot insert an order for a missing user.

python
from sqlalchemy.exc import IntegrityError
def test_order_must_reference_existing_user(db_session):
    order = Order(user_id=9999, total=100.0)
    db_session.add(order)
    try:
        db_session.commit()
        assert False, "Expected IntegrityError due to foreign key"
    except IntegrityError:
        db_session.rollback()

Testing transactions that span multiple statements

Some operations write to multiple tables inside one transaction. For example, in an e‑commerce system:

  1. Create an order record.
  2. Insert order items.
  3. Decrease inventory.

If any step fails, the whole operation should roll back.

Pseudocode:

python
def create_order(db, user_id, items):
    # items is list of (product_id, quantity)
    order = Order(user_id=user_id, status="pending")
    db.add(order)
    db.flush()  # get order.id without commit
    total = 0
    for product_id, quantity in items:
        product = db.query(Product).get(product_id)
        if product.stock < quantity:
            raise ValueError("Not enough stock")
        product.stock -= quantity
        line_total = product.price * quantity
        total += line_total
        db.add(OrderItem(order_id=order.id, product_id=product_id,
                         quantity=quantity, price=product.price))
    order.total = total
    db.commit()
    db.refresh(order)
    return order

Test:

python
def test_create_order_rolls_back_on_insufficient_stock(db_session):
    user = User(email="bob@example.com", is_active=True)
    db_session.add(user)
    product = Product(name="Book", price=10.0, stock=1)
    db_session.add(product)
    db_session.commit()
    try:
        create_order(
            db_session,
            user_id=user.id,
            items=[(product.id, 2)],  # request quantity > stock
        )
        assert False, "Expected ValueError due to insufficient stock"
    except ValueError:
        db_session.rollback()
    # Ensure stock is unchanged
    p = db_session.query(Product).get(product.id)
    assert p.stock == 1
    # Ensure no order was created
    orders_count = db_session.query(Order).count()
    assert orders_count == 0

You verify both the application logic and transaction behavior.

Important rule: Any multi‑step operation that must be all‑or‑nothing should be tested to confirm that partial writes are rolled back when an error occurs.

Using Docker for Test Databases

In continuous integration or on developer machines, you often run the test database in Docker.

Example docker-compose.yml for tests:

yaml
version: "3.9"
services:
  db_test:
    image: postgres:16
    environment:
      POSTGRES_USER: test_user
      POSTGRES_PASSWORD: test_pass
      POSTGRES_DB: myapp_test
    ports:
      - "5433:5432"  # test DB on different port

Then set your TEST_DATABASE_URL to:

text
postgresql://test_user:test_pass@localhost:5433/myapp_test

In CI, you can:

  1. Start the Docker service.
  2. Wait for PostgreSQL to be ready.
  3. Run migrations.
  4. Run tests.

You now get a clean database per test run that behaves like production.

Testing Read‑Only Queries and Reports

Reporting queries, aggregations, and analytics often use more complex SQL. They can break silently if the schema or assumptions change.

Example: total revenue per day.

Repository:

python
from sqlalchemy import func
def get_daily_revenue(db):
    return (
        db.query(
            func.date(Order.created_at).label("day"),
            func.sum(Order.total).label("revenue"),
        )
        .group_by(func.date(Order.created_at))
        .order_by(func.date(Order.created_at))
        .all()
    )

Test:

python
from datetime import datetime
def test_get_daily_revenue_groups_and_sums_by_day(db_session):
    user = User(email="report@example.com", is_active=True)
    db_session.add(user)
    db_session.commit()
    orders = [
        Order(user_id=user.id, total=10.0,
              created_at=datetime(2024, 1, 1, 10, 0, 0)),
        Order(user_id=user.id, total=20.0,
              created_at=datetime(2024, 1, 1, 12, 0, 0)),
        Order(user_id=user.id, total=5.0,
              created_at=datetime(2024, 1, 2, 9, 0, 0)),
    ]
    db_session.add_all(orders)
    db_session.commit()
    rows = get_daily_revenue(db_session)
    result = {(row.day, float(row.revenue)) for row in rows}
    assert result == {
        (datetime(2024, 1, 1).date(), 30.0),
        (datetime(2024, 1, 2).date(), 5.0),
    }

Such tests give you confidence that your reports remain accurate.

Common Pitfalls in Database Testing

Tests that depend on run order

If tests share a database and do not clean state properly, running them in a different order can cause failures. For example:

If B runs after A, it fails.

Avoid this by:

Time‑dependent tests

Queries that depend on the current time can be tricky. For example, selecting rows created in the last 24 hours.

Solutions:

Slow tests

Database operations are slower than in‑memory ones. To keep your test suite fast:

Mixing unit and integration tests

Do not let unit tests accidentally connect to the database. Keep a clear distinction:

Summary

In database testing, you aim to verify how your backend interacts with the database, including queries, constraints, and transactions. The key practices are:

This way your backend tests cover not only your code, but also how it behaves with the real database that your users rely on.

Views: 6

Comments

Please login to add a comment.

Don't have an account? Register now!