Database Testing: Complete Guide to Data Integrity and Validation

-- Verify table exists with expected columns
SELECT column_name, data_type, is_nullable, column_default
FROM information_schema.columns
WHERE table_schema = 'public'
  AND table_name = 'orders'
ORDER BY ordinal_position;

-- List all constraints on a table
SELECT constraint_name, constraint_type
FROM information_schema.table_constraints
WHERE table_name = 'orders';
 
-- Verify foreign key relationships
SELECT
    tc.constraint_name,
    kcu.column_name,
    ccu.table_name AS referenced_table,
    ccu.column_name AS referenced_column
FROM information_schema.table_constraints tc
JOIN information_schema.key_column_usage kcu
    ON tc.constraint_name = kcu.constraint_name
JOIN information_schema.constraint_column_usage ccu
    ON tc.constraint_name = ccu.constraint_name
WHERE tc.constraint_type = 'FOREIGN KEY'
    AND tc.table_name = 'orders';

-- List all indexes on a table (PostgreSQL)
SELECT indexname, indexdef
FROM pg_indexes
WHERE tablename = 'orders';

-- Check view definition
SELECT view_definition
FROM information_schema.views
WHERE table_name = 'active_orders_summary';
 
-- Verify view returns expected columns
SELECT column_name, data_type
FROM information_schema.columns
WHERE table_name = 'active_orders_summary';

-- Valid insert should succeed
INSERT INTO customers (name, email, phone)
VALUES ('Jane Smith', 'jane@example.com', '+1-555-0123');
 
-- Verify the insert
SELECT * FROM customers WHERE email = 'jane@example.com';
 
-- Duplicate email should fail (assuming UNIQUE constraint)
INSERT INTO customers (name, email, phone)
VALUES ('Another Jane', 'jane@example.com', '+1-555-0456');
-- Expected: ERROR unique constraint violation
 
-- NULL in required field should fail
INSERT INTO customers (name, email, phone)
VALUES (NULL, 'test@example.com', '+1-555-0789');
-- Expected: ERROR not-null constraint violation

-- Test basic retrieval
SELECT * FROM orders WHERE customer_id = 42;
 
-- Test join correctness
SELECT o.id, c.name, o.total_amount
FROM orders o
JOIN customers c ON o.customer_id = c.id
WHERE o.status = 'completed';
 
-- Test aggregation accuracy
SELECT customer_id, COUNT(*) as order_count, SUM(total_amount) as total_spent
FROM orders
GROUP BY customer_id
HAVING SUM(total_amount) > 1000;
 
-- Test with NULL handling
SELECT * FROM customers WHERE phone IS NULL;
SELECT * FROM orders WHERE discount_code IS NOT NULL;

-- Update a single record
UPDATE orders SET status = 'shipped' WHERE id = 1001;
SELECT status FROM orders WHERE id = 1001;
-- Expected: 'shipped'
 
-- Verify other records are unaffected
SELECT COUNT(*) FROM orders WHERE status = 'shipped' AND id != 1001;
 
-- Test constraint enforcement on update
UPDATE orders SET total_amount = -50 WHERE id = 1001;
-- Expected: ERROR check constraint violation (if CHECK total_amount >= 0 exists)

-- Delete a record with no dependencies
DELETE FROM customers WHERE id = 999;
SELECT COUNT(*) FROM customers WHERE id = 999;
-- Expected: 0
 
-- Attempt to delete a record with foreign key references
DELETE FROM customers WHERE id = 42;
-- Expected behavior depends on foreign key configuration:
-- RESTRICT: ERROR, deletion blocked
-- CASCADE: Customer and all their orders deleted
-- SET NULL: Customer deleted, orders.customer_id set to NULL

-- Bulk insert
INSERT INTO orders (customer_id, total_amount, status)
SELECT id, 0.00, 'draft'
FROM customers
WHERE created_at > '2026-01-01';
 
-- Verify count matches
SELECT COUNT(*) FROM orders WHERE status = 'draft';

-- Insert order with non-existent customer should fail
INSERT INTO orders (customer_id, total_amount, status)
VALUES (999999, 100.00, 'pending');
-- Expected: ERROR foreign key constraint violation
 
-- Verify cascading behavior
-- If ON DELETE CASCADE is configured:
DELETE FROM customers WHERE id = 42;
SELECT COUNT(*) FROM orders WHERE customer_id = 42;
-- Expected: 0 (orders were cascaded)

-- Negative amount should fail
INSERT INTO orders (customer_id, total_amount, status)
VALUES (1, -50.00, 'pending');
-- Expected: ERROR check constraint violation
 
-- Invalid status should fail
INSERT INTO orders (customer_id, total_amount, status)
VALUES (1, 100.00, 'invalid_status');
-- Expected: ERROR check constraint violation (if CHECK constraint exists)

-- Duplicate email should fail
INSERT INTO customers (name, email) VALUES ('Test User', 'existing@example.com');
-- Expected: ERROR unique constraint violation
 
-- NULL handling in unique columns
-- Most databases allow multiple NULLs in a UNIQUE column
INSERT INTO customers (name, email) VALUES ('User A', NULL);
INSERT INTO customers (name, email) VALUES ('User B', NULL);
-- Expected: Both succeed (NULLs are not considered equal)

-- Verify order totals match line items
SELECT o.id, o.total_amount, SUM(li.quantity * li.unit_price) as calculated_total
FROM orders o
JOIN line_items li ON o.id = li.order_id
GROUP BY o.id, o.total_amount
HAVING o.total_amount != SUM(li.quantity * li.unit_price);
-- Expected: Empty result set (no mismatches)

-- Example stored procedure
CREATE OR REPLACE PROCEDURE process_refund(
    p_order_id INTEGER,
    p_refund_amount DECIMAL(10,2),
    OUT p_status VARCHAR(20)
)
LANGUAGE plpgsql AS $$
BEGIN
    -- Business logic here
    IF p_refund_amount <= 0 THEN
        p_status := 'INVALID_AMOUNT';
        RETURN;
    END IF;
 
    UPDATE orders
    SET total_amount = total_amount - p_refund_amount,
        status = 'refunded'
    WHERE id = p_order_id AND total_amount >= p_refund_amount;
 
    IF FOUND THEN
        p_status := 'SUCCESS';
    ELSE
        p_status := 'FAILED';
    END IF;
END;
$$;
 
-- Test: Valid refund
CALL process_refund(1001, 25.00, NULL);
-- Expected: p_status = 'SUCCESS'
 
-- Test: Refund exceeds order total
CALL process_refund(1001, 999999.00, NULL);
-- Expected: p_status = 'FAILED'
 
-- Test: Negative refund amount
CALL process_refund(1001, -10.00, NULL);
-- Expected: p_status = 'INVALID_AMOUNT'
 
-- Test: Non-existent order
CALL process_refund(999999, 25.00, NULL);
-- Expected: p_status = 'FAILED'

-- Test that error in procedure doesn't leave partial changes
-- (verify transaction rollback behavior)
BEGIN;
CALL process_order(1001);
-- If procedure fails mid-execution, verify no partial data changes exist
ROLLBACK;
 
SELECT * FROM orders WHERE id = 1001;
-- Should be unchanged from before the failed call

-- PostgreSQL: Measure execution time
EXPLAIN ANALYZE
SELECT * FROM get_customer_orders(42);
 
-- Verify execution plan is using indexes
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT * FROM get_customer_orders(42);

-- Example: Audit trigger that logs changes
CREATE OR REPLACE FUNCTION log_order_changes()
RETURNS TRIGGER AS $$
BEGIN
    INSERT INTO order_audit_log (order_id, old_status, new_status, changed_at)
    VALUES (NEW.id, OLD.status, NEW.status, NOW());
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;
 
-- Test: Update should create audit log entry
UPDATE orders SET status = 'shipped' WHERE id = 1001;
 
-- Verify audit entry was created
SELECT * FROM order_audit_log WHERE order_id = 1001;
-- Expected: One row with old_status and new_status recorded

-- Update multiple rows and verify trigger fires for each
UPDATE orders SET status = 'cancelled' WHERE created_at < '2025-01-01';
 
-- Verify one audit entry per updated row
SELECT COUNT(*) FROM order_audit_log WHERE new_status = 'cancelled';

-- Test that updates from triggers don't cause cascading trigger loops
UPDATE orders SET status = 'processing' WHERE id = 1001;
-- Should complete without timeout or recursion error

-- Test trigger behavior when OLD or NEW values contain NULLs
INSERT INTO orders (customer_id, total_amount, status, discount_code)
VALUES (1, 100.00, 'pending', NULL);
-- Verify trigger handled NULL discount_code correctly

-- BEFORE INSERT trigger that validates/transforms data
-- Test: Insert with lowercase email should be stored as lowercase
INSERT INTO customers (name, email) VALUES ('Test', 'TEST@EXAMPLE.COM');
SELECT email FROM customers WHERE name = 'Test';
-- Expected: 'test@example.com' (if BEFORE trigger normalizes email)

-- Test: Partial failure should roll back all changes
BEGIN;
UPDATE accounts SET balance = balance - 500 WHERE id = 1;  -- Debit
UPDATE accounts SET balance = balance + 500 WHERE id = 999; -- Credit (non-existent account)
COMMIT;
 
-- Verify: First account balance should be unchanged (whole transaction rolled back)
SELECT balance FROM accounts WHERE id = 1;

-- Session 1: Start transaction, read balance
BEGIN;
SELECT balance FROM accounts WHERE id = 1; -- Returns 1000
 
-- Session 2: Update the same account
UPDATE accounts SET balance = balance - 200 WHERE id = 1;
 
-- Session 1: Read again (behavior depends on isolation level)
SELECT balance FROM accounts WHERE id = 1;
-- READ COMMITTED: Returns 800 (sees committed change)
-- REPEATABLE READ: Returns 1000 (sees snapshot)
-- SERIALIZABLE: Returns 1000 (sees snapshot)
COMMIT;

-- Insert and commit data
INSERT INTO critical_data (id, value) VALUES (1, 'important');
COMMIT;
 
-- Simulate restart (in test environment)
-- After restart, verify data persists
SELECT * FROM critical_data WHERE id = 1;
-- Expected: Row exists with value = 'important'

-- Session 1:
BEGIN;
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
-- Wait...
UPDATE accounts SET balance = balance + 100 WHERE id = 2;
 
-- Session 2 (simultaneously):
BEGIN;
UPDATE accounts SET balance = balance - 100 WHERE id = 2;
-- Wait...
UPDATE accounts SET balance = balance + 100 WHERE id = 1;
 
-- Expected: Database detects deadlock and aborts one transaction
-- Verify: Both accounts have consistent balances after resolution

-- Before migration: Verify starting state
SELECT COUNT(*) FROM orders WHERE discount_percent IS NULL;
 
-- Run migration: Add column with default value
ALTER TABLE orders ADD COLUMN discount_percent DECIMAL(5,2) DEFAULT 0.00;
 
-- After migration: Verify schema change
SELECT column_name, data_type, column_default
FROM information_schema.columns
WHERE table_name = 'orders' AND column_name = 'discount_percent';
 
-- Verify existing data was handled correctly
SELECT COUNT(*) FROM orders WHERE discount_percent = 0.00;
-- Expected: Same count as before (default applied to all existing rows)

-- Run forward migration
-- Run rollback migration
-- Verify schema matches pre-migration state
-- Verify no data was lost or corrupted during round-trip

-- Migration: Split 'full_name' into 'first_name' and 'last_name'
 
-- Before migration: Record sample data
SELECT id, full_name FROM customers WHERE id IN (1, 2, 3);
 
-- After migration: Verify transformation
SELECT id, first_name, last_name FROM customers WHERE id IN (1, 2, 3);
-- Verify: 'Jane Smith' -> first_name='Jane', last_name='Smith'

-- Time the migration on a copy of production data
\timing on
ALTER TABLE large_table ADD COLUMN new_field VARCHAR(255);
-- If this takes minutes on 10M rows, it will lock the table for minutes in production

# Reset database and run all migrations from scratch
# Verify final schema matches expected state
# This catches issues where migrations depend on data
# created by other migrations

-- PostgreSQL: Analyze query execution plan
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT c.name, COUNT(o.id) as order_count, SUM(o.total_amount) as total
FROM customers c
LEFT JOIN orders o ON c.id = o.customer_id
WHERE c.created_at > '2025-01-01'
GROUP BY c.name
ORDER BY total DESC
LIMIT 100;

-- Test query without index
DROP INDEX IF EXISTS idx_orders_customer_id;
EXPLAIN ANALYZE SELECT * FROM orders WHERE customer_id = 42;
-- Note: Sequential Scan, execution time
 
-- Create index
CREATE INDEX idx_orders_customer_id ON orders(customer_id);
EXPLAIN ANALYZE SELECT * FROM orders WHERE customer_id = 42;
-- Expected: Index Scan, faster execution time

-- Generate test data at scale
INSERT INTO orders (customer_id, total_amount, status, created_at)
SELECT
    (random() * 10000)::int,
    (random() * 1000)::decimal(10,2),
    (ARRAY['pending','shipped','delivered','cancelled'])[floor(random()*4+1)],
    NOW() - (random() * 365 || ' days')::interval
FROM generate_series(1, 1000000);
 
-- Now test your critical queries against this data volume
EXPLAIN ANALYZE
SELECT customer_id, COUNT(*), SUM(total_amount)
FROM orders
WHERE created_at > '2025-06-01'
GROUP BY customer_id
ORDER BY SUM(total_amount) DESC;

-- Common injection payloads to test (in application inputs):
' OR '1'='1
'; DROP TABLE users; --
' UNION SELECT username, password FROM users --
1; EXEC xp_cmdshell('dir')

# VULNERABLE: String concatenation
cursor.execute(f"SELECT * FROM users WHERE id = {user_input}")
 
# SAFE: Parameterized query
cursor.execute("SELECT * FROM users WHERE id = %s", (user_input,))

-- If a stored procedure builds dynamic SQL, test it:
CALL search_customers('Smith'' OR ''1''=''1');
-- Should return only customers named "Smith' OR '1'='1"
-- Not all customers in the database

-- Test: Read-only user cannot modify data
SET ROLE readonly_user;
INSERT INTO customers (name, email) VALUES ('Test', 'test@test.com');
-- Expected: ERROR permission denied
 
-- Test: Application user cannot access admin tables
SET ROLE app_user;
SELECT * FROM admin_settings;
-- Expected: ERROR permission denied
 
-- Test: Schema changes require elevated privileges
SET ROLE app_user;
DROP TABLE customers;
-- Expected: ERROR permission denied

-- Check if sensitive columns are encrypted
-- (implementation varies by database and encryption method)
 
-- Verify password hashing
SELECT password FROM users WHERE id = 1;
-- Expected: Hashed value (e.g., $2b$12$...), NOT plaintext
 
-- Verify encrypted fields can be decrypted with correct key
SELECT pgp_sym_decrypt(ssn_encrypted, 'correct_key') FROM customers WHERE id = 1;
-- Expected: Returns decrypted SSN
 
-- Verify encrypted fields cannot be decrypted with wrong key
SELECT pgp_sym_decrypt(ssn_encrypted, 'wrong_key') FROM customers WHERE id = 1;
-- Expected: ERROR or NULL

-- Perform a sensitive operation
UPDATE customers SET email = 'new@example.com' WHERE id = 1;
 
-- Verify audit log captured the change
SELECT * FROM audit_log
WHERE table_name = 'customers'
  AND record_id = 1
  AND operation = 'UPDATE'
ORDER BY timestamp DESC
LIMIT 1;
-- Expected: Log entry with old value, new value, user, timestamp

-- Application should show generic error, not:
-- "ERROR: relation 'users' has no column named 'ssn_encrypted'"
-- This reveals table and column names to potential attackers

-- Count records in source
SELECT COUNT(*) FROM source_db.raw_transactions;
-- Result: 1,000,000
 
-- Count records in target after ETL
SELECT COUNT(*) FROM target_db.processed_transactions;
-- Expected: 1,000,000 (or documented exclusion count)
 
-- If counts differ, check rejection log
SELECT COUNT(*), rejection_reason
FROM etl_rejections
GROUP BY rejection_reason;

-- Source data: amounts in various currencies
SELECT id, amount, currency FROM source_db.transactions WHERE id = 12345;
-- Returns: 12345, 100.00, 'EUR'
 
-- Target data: amounts converted to USD
SELECT id, original_amount, converted_amount_usd, exchange_rate
FROM target_db.transactions WHERE source_id = 12345;
-- Expected: Correct USD conversion using the specified exchange rate

-- Source has duplicate customer records
SELECT email, COUNT(*) FROM source_db.customers GROUP BY email HAVING COUNT(*) > 1;
 
-- After ETL, duplicates should be merged or flagged
SELECT email, COUNT(*) FROM target_db.customers GROUP BY email HAVING COUNT(*) > 1;
-- Expected: 0 duplicates (if dedup is part of the ETL spec)

-- After initial full load
SELECT MAX(last_processed_timestamp) FROM etl_watermarks;
-- Returns: 2026-01-28 10:00:00
 
-- Add new records to source after the watermark
INSERT INTO source_db.transactions VALUES (...);
 
-- Run incremental ETL
-- Verify only new records were processed
SELECT COUNT(*) FROM target_db.transactions WHERE loaded_at > '2026-01-28 10:00:00';

-- Check for NULL values in required fields
SELECT COUNT(*) FROM target_db.transactions WHERE customer_id IS NULL;
-- Expected: 0
 
-- Check for out-of-range values
SELECT COUNT(*) FROM target_db.transactions WHERE amount < 0;
-- Expected: 0 (or known count of valid negative amounts like refunds)
 
-- Check date consistency
SELECT COUNT(*) FROM target_db.transactions
WHERE transaction_date > CURRENT_DATE;
-- Expected: 0 (no future dates)

// MongoDB: Verify document schema consistency
db.orders.find({
  $or: [
    { customer_id: { $exists: false } },
    { total_amount: { $exists: false } },
    { status: { $exists: false } }
  ]
}).count();
// Expected: 0 (all documents have required fields)
 
// Test nested document operations
db.orders.updateOne(
  { _id: ObjectId("...") },
  { $push: { items: { product_id: "P001", quantity: 2, price: 29.99 } } }
);
 
// Verify nested document update
db.orders.findOne({ _id: ObjectId("...") });
// Expected: items array includes new entry

// Write data
db.collection.insertOne({ key: "test", value: "written" });
 
// Immediate read (may not see write depending on consistency level)
const result = db.collection.findOne({ key: "test" });
// Test: Does the read return the written value?
// With eventual consistency, a retry loop may be needed
 
// Test with read preference settings
const result = db.collection
  .withReadPreference("primary")
  .findOne({ key: "test" });
// Reading from primary should always return latest write

# Set with TTL
SET session:user123 "active" EX 3600

# Verify TTL is set
TTL session:user123
# Expected: ~3600 seconds

# Verify expiration (after TTL elapses)
GET session:user123
# Expected: nil

# Test atomic increment
SET counter 0
INCR counter
INCR counter
GET counter
# Expected: 2

// Verify relationship integrity
MATCH (c:Customer)-[:PLACED]->(o:Order)
WHERE o.id = 1001
RETURN c.name, o.total_amount;
 
// Test for orphaned nodes
MATCH (o:Order)
WHERE NOT (o)<-[:PLACED]-(:Customer)
RETURN COUNT(o);
// Expected: 0 (no orphaned orders)
 
// Test traversal depth
MATCH path = (start:Customer {id: 1})-[:KNOWS*1..5]-(connected)
RETURN LENGTH(path), connected.name;

# docker-compose.test.yml
services:
  test-db:
    image: postgres:16
    environment:
      POSTGRES_DB: test_db
      POSTGRES_USER: test_user
      POSTGRES_PASSWORD: test_pass
    ports:
      - "5433:5432"
    volumes:
      - ./migrations:/docker-entrypoint-initdb.d

// Using Testcontainers (Java)
@Container
static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16")
    .withDatabaseName("test_db")
    .withInitScript("schema.sql");
 
@BeforeEach
void setUp() {
    // Run data setup script
    ScriptUtils.executeSqlScript(connection, new ClassPathResource("test-data.sql"));
}
 
@AfterEach
void tearDown() {
    // Clean up or rollback
    connection.rollback();
}

# Python example with pytest
@pytest.fixture
def db_session():
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)
    yield session
    transaction.rollback()
    connection.close()
 
def test_create_order(db_session):
    order = Order(customer_id=1, total_amount=100.00)
    db_session.add(order)
    db_session.flush()
    assert order.id is not None
    # Transaction automatically rolls back after test

from faker import Faker
fake = Faker()
 
# Generate realistic customer data
for _ in range(10000):
    cursor.execute(
        "INSERT INTO customers (name, email, phone, address) VALUES (%s, %s, %s, %s)",
        (fake.name(), fake.email(), fake.phone_number(), fake.address())
    )

-- Mask PII in a test copy of production data
UPDATE customers_test SET
    name = 'Customer_' || id,
    email = 'user' || id || '@test.example.com',
    phone = '+1-555-' || LPAD(id::text, 4, '0'),
    ssn = '***-**-' || RIGHT(ssn, 4);

-- pgTAP test: Verify table structure and constraints
BEGIN;
SELECT plan(5);
 
-- Test: Table exists
SELECT has_table('orders');
 
-- Test: Column exists with correct type
SELECT has_column('orders', 'total_amount');
SELECT col_type_is('orders', 'total_amount', 'numeric(10,2)');
 
-- Test: Foreign key exists
SELECT fk_ok('orders', 'customer_id', 'customers', 'id');
 
-- Test: Check constraint exists
SELECT has_check('orders');
 
SELECT * FROM finish();
ROLLBACK;

-- tSQLt test: Verify stored procedure behavior
EXEC tSQLt.NewTestClass 'OrderTests';
 
CREATE PROCEDURE OrderTests.[test process_refund rejects negative amount]
AS
BEGIN
    EXEC tSQLt.FakeTable 'dbo.orders';
 
    INSERT INTO dbo.orders (id, total_amount, status)
    VALUES (1, 100.00, 'completed');
 
    DECLARE @status VARCHAR(20);
    EXEC process_refund @order_id = 1, @refund_amount = -10, @status = @status OUTPUT;
 
    EXEC tSQLt.AssertEquals 'INVALID_AMOUNT', @status;
END;

name: Database Tests
on: [push, pull_request]
 
jobs:
  db-tests:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_DB: test_db
          POSTGRES_USER: test_user
          POSTGRES_PASSWORD: test_pass
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
 
    steps:
      - uses: actions/checkout@v4
 
      - name: Run migrations
        run: |
          PGPASSWORD=test_pass psql -h localhost -U test_user -d test_db \
            -f migrations/001_create_tables.sql \
            -f migrations/002_add_constraints.sql
 
      - name: Run pgTAP tests
        run: |
          PGPASSWORD=test_pass psql -h localhost -U test_user -d test_db \
            -f tests/db/*.sql
 
      - name: Run application database tests
        run: npm test -- --testPathPattern=database

- name: Test migration chain
  run: |
    # Apply all migrations sequentially
    for migration in migrations/*.sql; do
      PGPASSWORD=test_pass psql -h localhost -U test_user -d test_db -f "$migration"
    done
 
    # Verify final schema matches expected state
    PGPASSWORD=test_pass pg_dump -h localhost -U test_user -d test_db --schema-only > actual_schema.sql
    diff expected_schema.sql actual_schema.sql