Stored Procedures and Triggers

Business logic that lives in the database — stored procedures, functions, and triggers — must be tested like any other code. These are particularly common in financial systems, inventory management, and anywhere that data integrity is critical enough to enforce at the database level.

Why Business Logic in the Database?

The trade-off: database logic is harder to version control, harder to test, and harder to debug than application code. This is exactly why testing it is important.

Testing Stored Procedures

Basic Function Test

def test_calculate_order_total(db):
    cursor = db.cursor()

    # Set up test data
    cursor.execute("INSERT INTO orders (id, status) VALUES ('test-order', 'pending')")
    cursor.execute("""
        INSERT INTO line_items (order_id, quantity, unit_price) VALUES
        ('test-order', 2, 10.00),
        ('test-order', 1, 25.50)
    """)

    # Call the stored procedure
    cursor.execute("SELECT calculate_order_total('test-order')")
    total = cursor.fetchone()[0]

    # Verify calculation
    expected = (2 * 10.00) + (1 * 25.50)  # 45.50
    assert total == expected, f"Expected {expected}, got {total}"

Testing Edge Cases

def test_calculate_order_total_empty_order(db):
    """Order with no line items should return 0."""
    cursor = db.cursor()
    cursor.execute("INSERT INTO orders (id, status) VALUES ('empty-order', 'pending')")
    cursor.execute("SELECT calculate_order_total('empty-order')")
    assert cursor.fetchone()[0] == 0

def test_calculate_order_total_nonexistent(db):
    """Non-existent order should raise an error or return NULL."""
    cursor = db.cursor()
    cursor.execute("SELECT calculate_order_total('nonexistent')")
    result = cursor.fetchone()[0]
    assert result is None or result == 0

def test_calculate_order_total_large_quantities(db):
    """Test with large quantities to check for overflow."""
    cursor = db.cursor()
    cursor.execute("INSERT INTO orders (id, status) VALUES ('big-order', 'pending')")
    cursor.execute("""
        INSERT INTO line_items (order_id, quantity, unit_price) VALUES
        ('big-order', 999999, 9999.99)
    """)
    cursor.execute("SELECT calculate_order_total('big-order')")
    total = cursor.fetchone()[0]
    expected = 999999 * 9999.99
    assert abs(total - expected) < 0.01  # Allow small floating-point difference

def test_calculate_order_total_precision(db):
    """Test decimal precision (common bug with money calculations)."""
    cursor = db.cursor()
    cursor.execute("INSERT INTO orders (id, status) VALUES ('precise-order', 'pending')")
    cursor.execute("""
        INSERT INTO line_items (order_id, quantity, unit_price) VALUES
        ('precise-order', 3, 0.10)
    """)
    cursor.execute("SELECT calculate_order_total('precise-order')")
    total = cursor.fetchone()[0]
    # 3 * 0.10 should be exactly 0.30, not 0.30000000000000004
    assert total == 0.30, f"Precision error: expected 0.30, got {total}"

Testing Triggers

Triggers run automatically when data is inserted, updated, or deleted. They are invisible to the application — which makes them especially important to test.

updated_at Trigger

def test_updated_at_trigger(db):
    """Verify that updated_at changes when a row is modified."""
    cursor = db.cursor()

    # Create a user
    cursor.execute("""
        INSERT INTO users (id, name, email)
        VALUES ('trigger-test', 'Original Name', 'trigger@test.com')
        RETURNING created_at, updated_at
    """)
    created_at, updated_at = cursor.fetchone()
    assert created_at == updated_at  # On creation, both should be the same

    # Wait briefly and update
    import time
    time.sleep(1)
    cursor.execute("""
        UPDATE users SET name = 'Updated Name'
        WHERE id = 'trigger-test'
    """)

    # Verify updated_at changed
    cursor.execute("SELECT updated_at FROM users WHERE id = 'trigger-test'")
    new_updated_at = cursor.fetchone()[0]
    assert new_updated_at > updated_at, "updated_at should be newer after UPDATE"

def test_updated_at_unchanged_on_same_value(db):
    """Some implementations only update timestamp if data actually changed."""
    cursor = db.cursor()
    cursor.execute("""
        INSERT INTO users (id, name, email)
        VALUES ('same-val-test', 'Same Name', 'sameval@test.com')
    """)

    import time
    time.sleep(1)

    # Update with the same value
    cursor.execute("""
        UPDATE users SET name = 'Same Name'
        WHERE id = 'same-val-test'
    """)

    cursor.execute("""
        SELECT created_at, updated_at FROM users WHERE id = 'same-val-test'
    """)
    created_at, updated_at = cursor.fetchone()
    # Behavior depends on trigger implementation:
    # Some triggers update timestamp regardless, others only on actual change

Audit Trail Trigger

def test_audit_trail_on_update(db):
    """Verify that changes to users are logged in audit table."""
    cursor = db.cursor()

    # Create user
    cursor.execute("""
        INSERT INTO users (id, name, email, role)
        VALUES ('audit-test', 'Audit User', 'audit@test.com', 'viewer')
    """)

    # Update role
    cursor.execute("""
        UPDATE users SET role = 'admin' WHERE id = 'audit-test'
    """)

    # Check audit table
    cursor.execute("""
        SELECT old_value, new_value, changed_field, changed_at
        FROM audit_log
        WHERE table_name = 'users' AND record_id = 'audit-test'
        ORDER BY changed_at DESC LIMIT 1
    """)
    row = cursor.fetchone()
    assert row is not None, "Audit log entry should exist"
    assert row[0] == "viewer"   # old_value
    assert row[1] == "admin"    # new_value
    assert row[2] == "role"     # changed_field

def test_audit_trail_on_delete(db):
    """Verify that deletions are logged."""
    cursor = db.cursor()
    cursor.execute("""
        INSERT INTO users (id, name, email)
        VALUES ('delete-audit', 'Delete Me', 'delete@test.com')
    """)
    cursor.execute("DELETE FROM users WHERE id = 'delete-audit'")

    cursor.execute("""
        SELECT action FROM audit_log
        WHERE table_name = 'users' AND record_id = 'delete-audit'
    """)
    row = cursor.fetchone()
    assert row is not None
    assert row[0] == "DELETE"

Testing Computed/Derived Values

Some systems compute values in the database (materialized views, generated columns):

def test_user_order_count_computed(db):
    """Verify computed order_count matches actual orders."""
    cursor = db.cursor()

    # Create user with orders
    cursor.execute("INSERT INTO users (id, name, email) VALUES ('count-test', 'Count User', 'count@test.com')")
    cursor.execute("""
        INSERT INTO orders (user_id, status, total) VALUES
        ('count-test', 'completed', 50.00),
        ('count-test', 'completed', 75.00),
        ('count-test', 'cancelled', 25.00)
    """)

    # If there is a materialized view or computed column:
    cursor.execute("SELECT order_count FROM user_stats WHERE user_id = 'count-test'")
    computed_count = cursor.fetchone()[0]

    # Verify against actual count
    cursor.execute("SELECT COUNT(*) FROM orders WHERE user_id = 'count-test'")
    actual_count = cursor.fetchone()[0]

    assert computed_count == actual_count

Testing Procedure Error Handling

def test_transfer_funds_insufficient_balance(db):
    """Transfer should fail if source account has insufficient funds."""
    cursor = db.cursor()

    # Set up accounts
    cursor.execute("""
        INSERT INTO accounts (id, balance) VALUES
        ('acct-a', 100.00),
        ('acct-b', 50.00)
    """)

    # Attempt transfer exceeding balance
    with pytest.raises(Exception):  # Specific exception depends on implementation
        cursor.execute("SELECT transfer_funds('acct-a', 'acct-b', 200.00)")

    # Verify no money was moved (transaction should be rolled back internally)
    cursor.execute("SELECT balance FROM accounts WHERE id = 'acct-a'")
    assert cursor.fetchone()[0] == 100.00  # Unchanged
    cursor.execute("SELECT balance FROM accounts WHERE id = 'acct-b'")
    assert cursor.fetchone()[0] == 50.00   # Unchanged

Practical Exercise

1. Write tests for a calculate_order_total function: normal case, empty order, large numbers, precision
2. Write tests for an updated_at trigger: verify it changes on update
3. Write tests for an audit trail trigger: verify inserts, updates, and deletes are logged
4. Write tests for a fund transfer procedure: success case, insufficient balance, same account
5. Test a trigger that prevents deletion of admin users

Key Takeaways

• Database business logic (procedures, triggers) must be tested like application code
• Test edge cases: empty inputs, large numbers, precision, concurrent access
• Triggers are invisible to the application — testing is the only way to verify they work
• Test both success and failure paths of stored procedures
• Audit trail triggers require specific verification: old value, new value, changed field, timestamp