postgres-testing

PostgreSQL Testing Standards

pgTAP (In-Database Unit Tests)

BEGIN;
SELECT plan(3);

-- Test table exists
SELECT has_table('public', 'orders', 'orders table should exist');

-- Test column exists
SELECT has_column('public', 'orders', 'customer_name', 'orders should have customer_name column');

-- Test function
SELECT results_eq(
    'SELECT total FROM get_order_by_id(1)',
    ARRAY[100.00::NUMERIC],
    'get_order_by_id should return correct total'
);

SELECT * FROM finish();
ROLLBACK;

pgTAP Patterns

• Wrap tests in BEGIN/ROLLBACK for isolation
• Use plan(N) to declare expected assertions
• Use has_table, has_column, has_index for schema tests
• Use results_eq for query result assertions
• Use throws_ok for error condition tests
• Run: pg_prove -d testdb tests/*.sql

pytest + psycopg

import pytest
import psycopg

@pytest.fixture
def db_conn():
    conn = psycopg.connect("postgresql://test:test@localhost:5432/testdb")
    yield conn
    conn.rollback()
    conn.close()

def test_get_order_by_id(db_conn):
    # Arrange
    with db_conn.cursor() as cur:
        cur.execute(
            "INSERT INTO orders (customer_name, total, created_by) VALUES (%s, %s, %s) RETURNING id",
            ("Test Customer", 100.00, "test"),
        )
        order_id = cur.fetchone()[0]

    # Act
    with db_conn.cursor() as cur:
        cur.execute("SELECT * FROM get_order_by_id(%s)", (order_id,))
        result = cur.fetchone()

    # Assert
    assert result is not None
    assert result[1] == "Test Customer"
    assert float(result[2]) == 100.00

EF Core Integration Tests

public class OrderRepositoryTests : IAsyncLifetime
{
    private readonly OrderDbContext _context;

    public OrderRepositoryTests()
    {
        var options = new DbContextOptionsBuilder<OrderDbContext>()
            .UseNpgsql("Host=localhost;Database=testdb;Username=test;Password=test")
            .Options;
        _context = new OrderDbContext(options);
    }

    public async Task InitializeAsync() => await _context.Database.EnsureCreatedAsync();
    public async Task DisposeAsync() => await _context.Database.EnsureDeletedAsync();

    [Fact]
    public async Task GetById_ReturnsOrder_WhenExists()
    {
        var order = new Order { CustomerName = "Test", Total = 100m };
        _context.Orders.Add(order);
        await _context.SaveChangesAsync();

        var result = await _context.Orders.FindAsync(order.Id);

        result.Should().NotBeNull();
        result!.CustomerName.Should().Be("Test");
    }
}

Test Database Lifecycle

• Use Docker PostgreSQL for test isolation
• CREATE DATABASE / DROP DATABASE per test suite
• Use ROLLBACK per test for speed
• Never test against shared development databases
• Use pg_restore for seeding test data from fixtures

Test Case Categories

Positive

pgTAP:

BEGIN;
SELECT plan(2);

SELECT ok(
    (SELECT COUNT(*) FROM orders WHERE customer_id = 'c1') = 1,
    'positive: insert order with valid data'
);

SELECT results_eq(
    'SELECT status FROM orders WHERE customer_id = $$c1$$',
    ARRAY['pending'],
    'positive: new order has pending status'
);

SELECT * FROM finish();
ROLLBACK;

pytest:

def test_positive_create_order(db_conn):
    cursor = db_conn.cursor()
    cursor.execute(
        "INSERT INTO orders (customer_id, status) VALUES (%s, %s) RETURNING id",
        ("c1", "pending")
    )
    order_id = cursor.fetchone()[0]
    assert order_id is not None

Negative

pgTAP:

SELECT throws_ok(
    $$INSERT INTO orders (customer_id, status) VALUES (NULL, 'pending')$$,
    23502,
    NULL,
    'negative: rejects null customer_id'
);

SELECT throws_ok(
    $$INSERT INTO orders (customer_id, status) VALUES ('c1', 'invalid_status')$$,
    23514,
    NULL,
    'negative: rejects invalid status value'
);

Edge

SELECT ok(
    (SELECT COUNT(*) FROM orders WHERE customer_id = 'nonexistent') = 0,
    'edge: empty result set returns zero rows'
);

SELECT lives_ok(
    $$INSERT INTO orders (customer_id, status) VALUES (repeat('c', 255), 'pending')$$,
    'edge: accepts max-length customer_id (255 chars)'
);

Security

SELECT lives_ok(
    $$INSERT INTO orders (customer_id, status) VALUES ('''; DROP TABLE orders; --', 'pending')$$,
    'security: injection payload stored as literal string'
);

SELECT ok(
    (SELECT COUNT(*) FROM information_schema.tables WHERE table_name = 'orders') = 1,
    'security: orders table still exists after injection attempt'
);

def test_security_sql_injection_prevented(db_conn):
    cursor = db_conn.cursor()
    malicious = "'; DROP TABLE orders; --"
    cursor.execute("SELECT * FROM orders WHERE customer_id = %s", (malicious,))
    results = cursor.fetchall()
    assert results == []

    cursor.execute("SELECT 1 FROM information_schema.tables WHERE table_name = 'orders'")
    assert cursor.fetchone() is not None

def test_security_connection_string_not_in_errors(db_conn):
    cursor = db_conn.cursor()
    try:
        cursor.execute("SELECT * FROM nonexistent_table")
    except Exception as e:
        error_msg = str(e)
        assert "password" not in error_msg.lower()
        assert "host=" not in error_msg.lower()