PostgreSQL Testing with Testcontainers and pg-mem: A Complete Guide

npm install pg-mem

import { newDb } from 'pg-mem';

const db = newDb();

// Get a client compatible with the `pg` library interface
const client = db.adapters.createPg();

async function setupSchema(client) {
  await client.query(`
    CREATE TABLE users (
      id SERIAL PRIMARY KEY,
      email TEXT UNIQUE NOT NULL,
      name TEXT NOT NULL,
      created_at TIMESTAMPTZ DEFAULT NOW()
    );
    
    CREATE TABLE orders (
      id SERIAL PRIMARY KEY,
      user_id INTEGER REFERENCES users(id),
      total DECIMAL(10,2) NOT NULL,
      status TEXT DEFAULT 'pending'
    );
  `);
}

import { newDb } from 'pg-mem';
import { UserRepository } from '../src/repositories/user';

describe('UserRepository', () => {
  let db, pool, repo;
  
  beforeEach(async () => {
    db = newDb();
    pool = db.adapters.createPgPool();
    await setupSchema(pool);
    repo = new UserRepository(pool);
  });
  
  afterEach(() => {
    pool.end();
  });
  
  test('creates a user', async () => {
    const user = await repo.create({
      email: 'alice@example.com',
      name: 'Alice'
    });
    
    expect(user.id).toBeDefined();
    expect(user.email).toBe('alice@example.com');
    expect(user.created_at).toBeDefined();
  });
  
  test('enforces unique email constraint', async () => {
    await repo.create({ email: 'dup@example.com', name: 'First' });
    
    await expect(
      repo.create({ email: 'dup@example.com', name: 'Second' })
    ).rejects.toThrow(/unique/i);
  });
  
  test('finds users by partial name match', async () => {
    await repo.create({ email: 'alice@example.com', name: 'Alice Smith' });
    await repo.create({ email: 'bob@example.com', name: 'Bob Jones' });
    await repo.create({ email: 'alex@example.com', name: 'Alex Smith' });
    
    const smiths = await repo.findByName('Smith');
    expect(smiths).toHaveLength(2);
    expect(smiths.map(u => u.email)).toContain('alice@example.com');
  });
  
  test('handles complex aggregation query', async () => {
    const user = await repo.create({ email: 'rich@example.com', name: 'Rich' });
    
    await pool.query(`
      INSERT INTO orders (user_id, total, status) VALUES
        ($1, 99.99, 'completed'),
        ($1, 49.50, 'completed'),
        ($1, 25.00, 'pending')
    `, [user.id]);
    
    const stats = await repo.getOrderStats(user.id);
    expect(stats.total_completed).toBe('149.49');
    expect(stats.order_count).toBe(2);
  });
});

pip install testcontainers[postgres] psycopg2-binary pytest

import pytest
import psycopg2
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope="session")
def postgres_container():
    with PostgresContainer("postgres:16-alpine") as container:
        yield container

@pytest.fixture(scope="session")  
def db_connection(postgres_container):
    conn = psycopg2.connect(postgres_container.get_connection_url())
    yield conn
    conn.close()

@pytest.fixture(scope="function")
def db(db_connection):
    """Wrap each test in a transaction that gets rolled back"""
    cursor = db_connection.cursor()
    db_connection.autocommit = False
    
    yield cursor
    
    db_connection.rollback()  # Rollback after each test
    cursor.close()

def test_user_creation(db):
    db.execute("""
        INSERT INTO users (email, name) 
        VALUES ('test@example.com', 'Test User')
        RETURNING id, email, name
    """)
    user = db.fetchone()
    
    assert user['email'] == 'test@example.com'
    # Transaction is rolled back after this test
    # Next test starts with a clean slate

def test_foreign_key_constraint(db):
    # This test doesn't see the user from the previous test
    with pytest.raises(psycopg2.IntegrityError):
        db.execute("""
            INSERT INTO orders (user_id, total) 
            VALUES (99999, 100.00)
        """)

const { PostgreSqlContainer } = require('@testcontainers/postgresql');
const { Pool } = require('pg');

describe('Order Service Integration', () => {
  let container, pool;
  
  beforeAll(async () => {
    container = await new PostgreSqlContainer('postgres:16-alpine')
      .withDatabase('testdb')
      .withUsername('testuser')
      .withPassword('testpass')
      .start();
    
    pool = new Pool({
      connectionString: container.getConnectionUri()
    });
    
    // Run migrations
    await runMigrations(pool);
  }, 60000); // 60s timeout for container startup
  
  afterAll(async () => {
    await pool.end();
    await container.stop();
  });
  
  beforeEach(async () => {
    await pool.query('BEGIN');
  });
  
  afterEach(async () => {
    await pool.query('ROLLBACK');
  });
  
  test('calculates order total with discounts', async () => {
    // Insert test data
    const { rows: [user] } = await pool.query(
      'INSERT INTO users (email, name) VALUES ($1, $2) RETURNING *',
      ['buyer@test.com', 'Test Buyer']
    );
    
    await pool.query(
      'INSERT INTO orders (user_id, total, discount_pct) VALUES ($1, $2, $3)',
      [user.id, 100.00, 10]
    );
    
    const { rows: [order] } = await pool.query(
      'SELECT calculate_final_total(id) as final FROM orders WHERE user_id = $1',
      [user.id]
    );
    
    expect(parseFloat(order.final)).toBe(90.00);
  });
});

@pytest.fixture(scope="session")
def postgres_with_extensions():
    with PostgresContainer("postgres:16-alpine") as container:
        conn = psycopg2.connect(container.get_connection_url())
        cur = conn.cursor()
        cur.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm")
        cur.execute("CREATE EXTENSION IF NOT EXISTS uuid-ossp")
        conn.commit()
        yield conn
        conn.close()

def test_trigram_similarity_search(db_with_extensions):
    db_with_extensions.execute("""
        CREATE INDEX IF NOT EXISTS products_name_trgm 
        ON products USING gin (name gin_trgm_ops)
    """)
    
    db_with_extensions.execute("""
        SELECT name, similarity(name, 'PostgresQL') as sim
        FROM products
        WHERE name % 'PostgresQL'
        ORDER BY sim DESC
        LIMIT 5
    """)
    
    results = db_with_extensions.fetchall()
    assert len(results) > 0
    assert results[0]['sim'] > 0.3

import { PrismaClient } from '@prisma/client';
import { PostgreSqlContainer } from '@testcontainers/postgresql';
import { execSync } from 'child_process';

let prisma: PrismaClient;
let container: StartedPostgreSqlContainer;

beforeAll(async () => {
  container = await new PostgreSqlContainer('postgres:16-alpine').start();
  
  process.env.DATABASE_URL = container.getConnectionUri();
  
  // Run Prisma migrations
  execSync('npx prisma migrate deploy', {
    env: { ...process.env, DATABASE_URL: container.getConnectionUri() }
  });
  
  prisma = new PrismaClient({
    datasources: { db: { url: container.getConnectionUri() } }
  });
});

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

test('creates user with correct defaults', async () => {
  const user = await prisma.user.create({
    data: { email: 'prisma@test.com', name: 'Prisma User' }
  });
  
  expect(user.id).toMatch(/^[0-9a-f-]{36}$/); // UUID
  expect(user.createdAt).toBeInstanceOf(Date);
  expect(user.role).toBe('USER'); // Default value
});

name: Database Tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-node@v4
        with:
          node-version: '20'
      
      - name: Install dependencies
        run: npm ci
      
      # Testcontainers handles Docker automatically in GitHub Actions
      - name: Run integration tests
        run: npm test
        env:
          DOCKER_HOST: unix:///var/run/docker.sock

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

steps:
  - name: Run tests
    run: pytest tests/
    env:
      DATABASE_URL: postgresql://postgres:testpass@localhost:5432/testdb