Database Testing — SQL, Migrations & Data Integrity Testing Guide
Database testing verifies that your data layer handles queries, migrations, constraints, and transactions correctly — catching schema drift, integrity violations, and performance regressions before they reach production. In this guide, you will learn how to test SQL queries for correctness, validate database migrations forward and backward, and enforce data integrity with automated test suites. The DodaZIP team runs database tests against every migration before it touches the production metadata store, preventing data corruption in user archives.
Learning Path
flowchart LR A[Integration Testing] --> B[Database Testing
You are here] B --> C[Migration Validation] C --> D[Data Integrity Checks] B --> E[API Testing Guide] style B fill:#f90,color:#fff
SQL Query Testing
Test that queries return the expected results for known data:
import sqlite3
def setup_test_db():
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, email TEXT)")
conn.execute("INSERT INTO users VALUES (1, 'Alice', 'alice@test.com')")
conn.execute("INSERT INTO users VALUES (2, 'Bob', 'bob@test.com')")
conn.execute("INSERT INTO users VALUES (3, 'Charlie', 'charlie@test.com')")
return conn
def query_active_users(conn):
cursor = conn.execute("SELECT name, email FROM users ORDER BY name")
return cursor.fetchall()
conn = setup_test_db()
result = query_active_users(conn)
print(result)
conn.close()
Expected output:
[('Alice', 'alice"@test".com'), ('Bob', 'bob"@test".com'), ('Charlie', 'charlie"@test".com')]
Migration Testing
Test that migrations apply and roll back correctly:
import sqlite3
def apply_migration(conn):
conn.execute("ALTER TABLE users ADD COLUMN phone TEXT")
conn.execute("CREATE INDEX idx_users_email ON users(email)")
def rollback_migration(conn):
conn.execute("DROP INDEX IF EXISTS idx_users_email")
conn.executescript("""
CREATE TABLE users_backup AS SELECT id, name, email FROM users;
DROP TABLE users;
ALTER TABLE users_backup RENAME TO users;
""")
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, email TEXT)")
conn.execute("INSERT INTO users VALUES (1, 'Alice', 'alice@test.com')")
apply_migration(conn)
columns = [desc[0] for desc in conn.execute("PRAGMA table_info(users)")]
print("After migration:", columns)
rollback_migration(conn)
columns = [desc[0] for desc in conn.execute("PRAGMA table_info(users)")]
print("After rollback:", columns)
conn.close()
Expected output:
After migration: ['id', 'name', 'email', 'phone']
After rollback: ['id', 'name', 'email']
ACID Property Testing
Test that transactions maintain atomicity and isolation:
import sqlite3, threading
def test_race_condition():
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE inventory (item TEXT PRIMARY KEY, qty INTEGER)")
conn.execute("INSERT INTO inventory VALUES ('widget', 10)")
def deduct(amount):
try:
conn.execute("BEGIN")
row = conn.execute("SELECT qty FROM inventory WHERE item='widget'").fetchone()
if row[0] >= amount:
conn.execute("UPDATE inventory SET qty = qty - ? WHERE item='widget'", (amount,))
conn.commit()
except:
conn.rollback()
threads = [threading.Thread(target=deduct, args=(5,)) for _ in range(3)]
for t in threads: t.start()
for t in threads: t.join()
result = conn.execute("SELECT qty FROM inventory WHERE item='widget'").fetchone()
print(f"Final quantity: {result[0]}")
conn.close()
test_race_condition()
Expected output:
Final quantity: 0
Without proper transaction handling, two threads could both read 10 and deduct 5, leaving a final quantity of 5 instead of 0.
Data Integrity Constraints
Test that constraints enforce data quality:
import sqlite3
conn = sqlite3.connect(":memory:")
conn.execute("""
CREATE TABLE orders (
id INTEGER PRIMARY KEY,
user_id INTEGER NOT NULL,
total REAL CHECK(total > 0),
status TEXT DEFAULT 'pending',
FOREIGN KEY (user_id) REFERENCES users(id)
)
""")
conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
violations = []
try:
conn.execute("INSERT INTO orders (id, user_id, total) VALUES (1, 999, -10)")
except Exception as e:
violations.append(f"Negative total rejected: {e}")
try:
conn.execute("INSERT INTO orders (id, user_id, total) VALUES (2, 999, 50)")
except Exception as e:
violations.append(f"Invalid FK rejected: {e}")
for v in violations:
print(v)
conn.close()
Expected output:
Negative total rejected: CHECK constraint failed: total > 0
Invalid FK rejected: FOREIGN KEY constraint failed
Automating Database Tests with pytest
import pytest
import sqlite3
@pytest.fixture
def db():
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, email TEXT UNIQUE)")
conn.execute("INSERT INTO users VALUES (1, 'Alice', 'alice@test.com')")
yield conn
conn.close()
def test_insert_duplicate_email(db):
with pytest.raises(sqlite3.IntegrityError):
db.execute("INSERT INTO users VALUES (2, 'Bob', 'alice@test.com')")
def test_query_by_id(db):
row = db.execute("SELECT name FROM users WHERE id=1").fetchone()
assert row[0] == "Alice"
def test_update(db):
db.execute("UPDATE users SET name='Alicia' WHERE id=1")
row = db.execute("SELECT name FROM users WHERE id=1").fetchone()
assert row[0] == "Alicia"
Common Database Testing Mistakes
1. Testing Against Production
Running tests on production data risks corruption and exposes sensitive information.
2. No Test Isolation
Tests that share database state produce flaky results. Use transactions or fresh databases per test.
3. Ignoring Indexes
Testing queries without production-like indexes misses performance regressions.
4. Skipping Rollback Tests
Migrations that apply but cannot roll back trap you in broken states.
5. Not Testing Concurrent Access
Race conditions in database code cause the most insidious production bugs.
Practice Questions
1. What is the purpose of migration rollback testing?
It ensures you can safely revert a schema change if the migration causes issues in production.
2. Why should each database test run in isolation?
Shared database state causes tests to interfere with each other, producing flaky results and false failures.
3. What ACID property prevents concurrent transactions from seeing partial writes?
Isolation. It ensures concurrent transactions do not interfere with each other.
4. How do CHECK constraints differ from NOT NULL constraints?
CHECK validates a boolean expression. NOT NULL only ensures a column is not null.
Challenge: Write a database test suite for a banking application. Test: account creation with initial balance, transfer between accounts (atomic debit/credit), overdraft prevention, concurrent transfer safety, and migration adding a transaction fee column with rollback.
FAQ
What's Next
Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro