Prerequisites
- Basic understanding of programming concepts 📝
- Python installation (3.8+) 🐍
- VS Code or preferred IDE 💻
What you'll learn
- Understand the concept fundamentals 🎯
- Apply the concept in real projects 🏗️
- Debug common issues 🐛
- Write clean, Pythonic code ✨
🎯 Introduction
Welcome to this exciting tutorial on testing database code! 🎉 Have you ever wondered how to test your database-driven applications without messing up production data? Or struggled with tests that fail randomly because of database state? You’re in the right place!
Testing database code is like having a sandbox 🏖️ where you can build and destroy castles without worrying about the real beach. You’ll discover how test databases give you the freedom to experiment, validate your code, and sleep peacefully knowing your tests won’t accidentally delete important data! 😴
By the end of this tutorial, you’ll be confidently setting up test databases, writing reliable database tests, and following best practices that professional developers use every day. Let’s dive in! 🏊♂️
📚 Understanding Test Databases
🤔 What are Test Databases?
Test databases are like practice rooms for musicians 🎸. Just as musicians need a safe space to practice without disturbing others, developers need isolated databases to test their code without affecting real data.
In Python terms, a test database is a separate database instance used exclusively for running tests. This means you can:
- ✨ Create, modify, and delete data freely
- 🚀 Run tests in parallel without conflicts
- 🛡️ Protect production data from test operations
- 🔄 Reset to a clean state for each test
💡 Why Use Test Databases?
Here’s why developers love test databases:
- Isolation 🏝️: Each test runs in its own clean environment
- Reproducibility 🔄: Tests produce consistent results every time
- Safety 🛡️: No risk of corrupting production data
- Speed ⚡: Tests can run in parallel without interference
- Confidence 💪: Test real database operations, not mocks
Real-world example: Imagine testing an e-commerce system 🛒. With test databases, you can simulate thousands of orders, test edge cases like payment failures, and verify inventory updates—all without touching a single real customer’s data!
🔧 Basic Syntax and Usage
📝 Setting Up a Test Database with SQLite
Let’s start with a simple example using SQLite (perfect for testing!):
import sqlite3
import pytest
from pathlib import Path
# 🎨 Create a test database fixture
@pytest.fixture
def test_db():
# 👋 Create a temporary database
db_path = Path("test_shop.db")
connection = sqlite3.connect(db_path)
# 🏗️ Set up schema
cursor = connection.cursor()
cursor.execute("""
CREATE TABLE products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
price REAL NOT NULL,
stock INTEGER DEFAULT 0
)
""")
connection.commit()
# 🎯 Yield the connection for tests
yield connection
# 🧹 Clean up after test
connection.close()
db_path.unlink() # Delete test database
# 🎮 Use the test database
def test_add_product(test_db):
cursor = test_db.cursor()
# ➕ Add a product
cursor.execute(
"INSERT INTO products (name, price, stock) VALUES (?, ?, ?)",
("Python Book 📘", 29.99, 10)
)
test_db.commit()
# ✅ Verify it was added
cursor.execute("SELECT * FROM products WHERE name LIKE '%Python%'")
product = cursor.fetchone()
assert product is not None
assert product[1] == "Python Book 📘"
assert product[2] == 29.99💡 Explanation: Notice how we create a fresh database for each test and clean it up afterward. This ensures tests don’t interfere with each other!
🎯 Using PostgreSQL for Testing
For more complex applications, you might use PostgreSQL:
import psycopg2
import pytest
from contextlib import contextmanager
# 🏗️ Database configuration
TEST_DB_CONFIG = {
'host': 'localhost',
'port': 5432,
'user': 'test_user',
'password': 'test_pass',
'database': 'test_shop_db'
}
# 🎨 Context manager for test transactions
@contextmanager
def test_transaction():
conn = psycopg2.connect(**TEST_DB_CONFIG)
conn.autocommit = False
try:
yield conn
conn.rollback() # 🔄 Always rollback to keep DB clean!
finally:
conn.close()
# 🧪 Test with automatic rollback
def test_inventory_update():
with test_transaction() as conn:
cursor = conn.cursor()
# 📦 Insert test product
cursor.execute("""
INSERT INTO products (name, price, stock)
VALUES (%s, %s, %s) RETURNING id
""", ("Gaming Mouse 🖱️", 49.99, 50))
product_id = cursor.fetchone()[0]
# 🛒 Simulate a purchase
cursor.execute("""
UPDATE products
SET stock = stock - %s
WHERE id = %s AND stock >= %s
""", (5, product_id, 5))
# ✅ Verify stock was updated
cursor.execute("SELECT stock FROM products WHERE id = %s", (product_id,))
new_stock = cursor.fetchone()[0]
assert new_stock == 45
# 🎉 Transaction will rollback, keeping DB clean!💡 Practical Examples
🛒 Example 1: E-Commerce Order System
Let’s build a complete test suite for an order system:
import sqlite3
from datetime import datetime
from decimal import Decimal
import pytest
# 🏪 Our e-commerce system
class OrderSystem:
def __init__(self, db_connection):
self.conn = db_connection
self.cursor = db_connection.cursor()
# 🛒 Create a new order
def create_order(self, customer_id, items):
# 📝 Create order record
self.cursor.execute("""
INSERT INTO orders (customer_id, total, status, created_at)
VALUES (?, 0, 'pending', ?)
""", (customer_id, datetime.now()))
order_id = self.cursor.lastrowid
total = Decimal('0')
# 📦 Add order items
for item in items:
product_id, quantity = item['product_id'], item['quantity']
# 🔍 Get product details
self.cursor.execute(
"SELECT price, stock FROM products WHERE id = ?",
(product_id,)
)
price, stock = self.cursor.fetchone()
if stock < quantity:
raise ValueError(f"Insufficient stock! 😱 Only {stock} available")
# ➕ Add to order
self.cursor.execute("""
INSERT INTO order_items (order_id, product_id, quantity, price)
VALUES (?, ?, ?, ?)
""", (order_id, product_id, quantity, price))
# 📉 Update stock
self.cursor.execute(
"UPDATE products SET stock = stock - ? WHERE id = ?",
(quantity, product_id)
)
total += Decimal(str(price)) * quantity
# 💰 Update order total
self.cursor.execute(
"UPDATE orders SET total = ? WHERE id = ?",
(float(total), order_id)
)
self.conn.commit()
return order_id
# 🧪 Test fixtures
@pytest.fixture
def test_database():
# 🎨 Create test database with schema
conn = sqlite3.connect(':memory:') # In-memory for speed! ⚡
cursor = conn.cursor()
# 📊 Create tables
cursor.executescript("""
CREATE TABLE products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
price REAL NOT NULL,
stock INTEGER DEFAULT 0
);
CREATE TABLE orders (
id INTEGER PRIMARY KEY,
customer_id INTEGER NOT NULL,
total REAL DEFAULT 0,
status TEXT DEFAULT 'pending',
created_at TIMESTAMP
);
CREATE TABLE order_items (
id INTEGER PRIMARY KEY,
order_id INTEGER REFERENCES orders(id),
product_id INTEGER REFERENCES products(id),
quantity INTEGER NOT NULL,
price REAL NOT NULL
);
""")
# 🎮 Add test data
test_products = [
("Gaming Keyboard ⌨️", 89.99, 20),
("Wireless Mouse 🖱️", 49.99, 15),
("USB-C Cable 🔌", 12.99, 100),
("Webcam 📹", 79.99, 5)
]
cursor.executemany(
"INSERT INTO products (name, price, stock) VALUES (?, ?, ?)",
test_products
)
conn.commit()
yield conn
conn.close()
# 🎯 Test successful order
def test_create_order_success(test_database):
order_system = OrderSystem(test_database)
# 🛍️ Create an order
order_id = order_system.create_order(
customer_id=123,
items=[
{'product_id': 1, 'quantity': 2}, # 2 keyboards
{'product_id': 3, 'quantity': 3} # 3 cables
]
)
# ✅ Verify order was created
cursor = test_database.cursor()
cursor.execute("SELECT total, status FROM orders WHERE id = ?", (order_id,))
total, status = cursor.fetchone()
expected_total = (89.99 * 2) + (12.99 * 3)
assert abs(total - expected_total) < 0.01 # Float comparison
assert status == 'pending'
# 📦 Verify stock was updated
cursor.execute("SELECT stock FROM products WHERE id IN (1, 3)")
stocks = cursor.fetchall()
assert stocks[0][0] == 18 # 20 - 2 keyboards
assert stocks[1][0] == 97 # 100 - 3 cables
# 😱 Test insufficient stock
def test_create_order_insufficient_stock(test_database):
order_system = OrderSystem(test_database)
# 🚫 Try to order more than available
with pytest.raises(ValueError) as exc_info:
order_system.create_order(
customer_id=456,
items=[{'product_id': 4, 'quantity': 10}] # Only 5 webcams!
)
assert "Insufficient stock!" in str(exc_info.value)
assert "Only 5 available" in str(exc_info.value)
# ✅ Verify stock wasn't changed
cursor = test_database.cursor()
cursor.execute("SELECT stock FROM products WHERE id = 4")
assert cursor.fetchone()[0] == 5 # Stock unchanged! 🛡️🎯 Try it yourself: Add a test for canceling orders and restoring stock!
🎮 Example 2: User Authentication System
Let’s test a user authentication system with proper database isolation:
import hashlib
import secrets
from datetime import datetime, timedelta
import pytest
# 🔐 Authentication system
class AuthSystem:
def __init__(self, db_connection):
self.conn = db_connection
self.cursor = db_connection.cursor()
# 🔒 Hash password securely
def _hash_password(self, password, salt):
return hashlib.pbkdf2_hmac(
'sha256',
password.encode('utf-8'),
salt.encode('utf-8'),
100000
).hex()
# 👤 Register new user
def register_user(self, username, email, password):
# 🎲 Generate salt
salt = secrets.token_hex(32)
password_hash = self._hash_password(password, salt)
try:
self.cursor.execute("""
INSERT INTO users (username, email, password_hash, salt, created_at)
VALUES (?, ?, ?, ?, ?)
""", (username, email, password_hash, salt, datetime.now()))
self.conn.commit()
return True, "Registration successful! 🎉"
except sqlite3.IntegrityError as e:
if "username" in str(e):
return False, "Username already taken! 😕"
elif "email" in str(e):
return False, "Email already registered! 📧"
raise
# 🔓 Login user
def login(self, username, password):
self.cursor.execute("""
SELECT id, password_hash, salt, failed_attempts, locked_until
FROM users WHERE username = ?
""", (username,))
user = self.cursor.fetchone()
if not user:
return False, "Invalid credentials! 🚫"
user_id, password_hash, salt, failed_attempts, locked_until = user
# 🔒 Check if account is locked
if locked_until and datetime.fromisoformat(locked_until) > datetime.now():
return False, "Account locked! Try again later 🔒"
# ✅ Verify password
if self._hash_password(password, salt) == password_hash:
# 🎉 Reset failed attempts
self.cursor.execute("""
UPDATE users
SET failed_attempts = 0, locked_until = NULL, last_login = ?
WHERE id = ?
""", (datetime.now(), user_id))
self.conn.commit()
return True, "Welcome back! 👋"
else:
# 😱 Track failed attempt
failed_attempts += 1
locked_until = None
if failed_attempts >= 3:
locked_until = datetime.now() + timedelta(minutes=15)
message = "Too many attempts! Account locked for 15 minutes 🔒"
else:
message = f"Invalid credentials! {3 - failed_attempts} attempts remaining ⚠️"
self.cursor.execute("""
UPDATE users
SET failed_attempts = ?, locked_until = ?
WHERE id = ?
""", (failed_attempts, locked_until, user_id))
self.conn.commit()
return False, message
# 🧪 Test database setup
@pytest.fixture
def auth_test_db():
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()
# 👥 Create users table
cursor.execute("""
CREATE TABLE users (
id INTEGER PRIMARY KEY,
username TEXT UNIQUE NOT NULL,
email TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL,
salt TEXT NOT NULL,
failed_attempts INTEGER DEFAULT 0,
locked_until TEXT,
created_at TIMESTAMP,
last_login TIMESTAMP
)
""")
conn.commit()
yield conn
conn.close()
# ✅ Test successful registration
def test_register_success(auth_test_db):
auth = AuthSystem(auth_test_db)
success, message = auth.register_user(
"python_lover",
"[email protected]",
"SecurePass123! 🔐"
)
assert success is True
assert "successful" in message
# 🔍 Verify user was created
cursor = auth_test_db.cursor()
cursor.execute("SELECT username, email FROM users WHERE username = ?", ("python_lover",))
user = cursor.fetchone()
assert user[0] == "python_lover"
assert user[1] == "[email protected]"
# ❌ Test duplicate username
def test_register_duplicate_username(auth_test_db):
auth = AuthSystem(auth_test_db)
# 👤 First registration
auth.register_user("coder123", "[email protected]", "Pass123!")
# 🚫 Try same username
success, message = auth.register_user(
"coder123", # Same username!
"[email protected]",
"DifferentPass!"
)
assert success is False
assert "Username already taken" in message
# 🔐 Test login and lockout
def test_login_with_lockout(auth_test_db):
auth = AuthSystem(auth_test_db)
# 👤 Register user
auth.register_user("test_user", "[email protected]", "CorrectPass123!")
# ❌ Try wrong password 3 times
for i in range(3):
success, message = auth.login("test_user", "WrongPass!")
assert success is False
if i < 2:
assert f"{2-i} attempts remaining" in message
else:
assert "Account locked" in message
# 🔒 Try correct password while locked
success, message = auth.login("test_user", "CorrectPass123!")
assert success is False
assert "Account locked" in message
# ✅ Successful login (in real scenario, wait 15 minutes)
# For testing, we'll manually unlock
cursor = auth_test_db.cursor()
cursor.execute("UPDATE users SET locked_until = NULL WHERE username = ?", ("test_user",))
auth_test_db.commit()
success, message = auth.login("test_user", "CorrectPass123!")
assert success is True
assert "Welcome back" in message🚀 Advanced Concepts
🧙♂️ Database Migrations in Tests
When testing evolving schemas, manage migrations properly:
import alembic.config
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 🎯 Advanced test database with migrations
class TestDatabaseManager:
def __init__(self, db_url="sqlite:///:memory:"):
self.db_url = db_url
self.engine = None
self.SessionLocal = None
# 🏗️ Set up database with migrations
def setup(self):
self.engine = create_engine(self.db_url)
# 🔄 Run migrations
alembic_cfg = alembic.config.Config("alembic.ini")
alembic_cfg.set_main_option("sqlalchemy.url", self.db_url)
alembic.command.upgrade(alembic_cfg, "head")
self.SessionLocal = sessionmaker(bind=self.engine)
return self
# 🧹 Clean up
def teardown(self):
if self.engine:
self.engine.dispose()
# 🎨 Get a test session
def get_session(self):
session = self.SessionLocal()
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
# 🧪 Use in tests
@pytest.fixture(scope="function")
def test_db_manager():
manager = TestDatabaseManager()
manager.setup()
yield manager
manager.teardown()🏗️ Test Data Factories
Create realistic test data easily:
from faker import Faker
import factory
from factory.alchemy import SQLAlchemyModelFactory
fake = Faker()
# 🏭 User factory
class UserFactory(SQLAlchemyModelFactory):
class Meta:
model = User
sqlalchemy_session_persistence = 'commit'
username = factory.LazyAttribute(lambda x: fake.user_name())
email = factory.LazyAttribute(lambda x: fake.email())
full_name = factory.LazyAttribute(lambda x: fake.name())
created_at = factory.LazyFunction(datetime.now)
# 🎮 Product factory
class ProductFactory(SQLAlchemyModelFactory):
class Meta:
model = Product
name = factory.Sequence(lambda n: f"Product {n} {fake.emoji()}")
price = factory.LazyFunction(lambda: round(fake.random.uniform(10, 200), 2))
stock = factory.LazyFunction(lambda: fake.random_int(0, 100))
category = factory.Faker('random_element', elements=['Electronics 💻', 'Books 📚', 'Games 🎮'])
# 🧪 Use in tests
def test_bulk_order_processing(test_session):
# 🎨 Create test data quickly!
users = UserFactory.create_batch(5)
products = ProductFactory.create_batch(20)
# 🛒 Test bulk operations
for user in users:
# Create orders with random products
selected_products = fake.random_elements(products, length=3)
# ... test order processing⚠️ Common Pitfalls and Solutions
😱 Pitfall 1: Shared Test Database State
# ❌ Wrong way - tests affect each other!
class TestProduct:
db = setup_test_database() # Shared across all tests! 😰
def test_create_product(self):
# This creates data that affects other tests
create_product(self.db, "Test Product")
def test_list_products(self):
# This might see products from other tests! 💥
products = list_products(self.db)
assert len(products) == 0 # Fails if test_create_product ran first!
# ✅ Correct way - isolated test databases!
@pytest.fixture
def test_db():
db = setup_test_database()
yield db
cleanup_database(db) # 🧹 Clean after each test
def test_create_product(test_db):
create_product(test_db, "Test Product")
# Database is fresh for this test only! ✨
def test_list_products(test_db):
products = list_products(test_db)
assert len(products) == 0 # Always passes! 🎉🤯 Pitfall 2: Forgetting to Clean Up
# ❌ Dangerous - leaves test data behind!
def test_user_creation():
conn = psycopg2.connect("postgresql://localhost/main_db") # 😱 Production DB!
cursor = conn.cursor()
cursor.execute("INSERT INTO users (email) VALUES ('[email protected]')")
conn.commit()
# Oops! Test data in production! 💥
# ✅ Safe - use test database with cleanup!
@pytest.fixture
def test_db():
# 🛡️ Create isolated test database
conn = psycopg2.connect("postgresql://localhost/test_db")
yield conn
# 🧹 Clean up everything
conn.rollback() # Undo any uncommitted changes
conn.close()
def test_user_creation(test_db):
cursor = test_db.cursor()
cursor.execute("INSERT INTO users (email) VALUES ('[email protected]')")
# No commit needed - will be rolled back! ✅🛠️ Best Practices
- 🎯 Use Transactions: Rollback after each test for speed and isolation
- 📝 Seed Consistent Data: Use factories or fixtures for predictable test data
- 🛡️ Never Touch Production: Always use separate test databases
- 🎨 Name Test Databases Clearly: Use prefixes like
test_or_test - ✨ Keep Tests Fast: Use in-memory databases when possible
- 🔄 Reset Between Tests: Start each test with a clean slate
- 📊 Use Migrations: Test your schema changes too!
🧪 Hands-On Exercise
🎯 Challenge: Build a Library Management System Test Suite
Create a comprehensive test suite for a library system:
📋 Requirements:
- ✅ Books with ISBN, title, author, and availability
- 📚 Members who can borrow books
- 🔄 Borrowing system with due dates
- 💰 Fine calculation for overdue books
- 📊 Reports on most borrowed books
- 🎨 Each book category needs an emoji!
🚀 Bonus Points:
- Implement reservation system
- Add book recommendations based on history
- Create admin functions for managing fines
💡 Solution
🔍 Click to see solution
import sqlite3
from datetime import datetime, timedelta
from decimal import Decimal
import pytest
# 📚 Library Management System
class Library:
def __init__(self, db_connection):
self.conn = db_connection
self.cursor = db_connection.cursor()
self.fine_per_day = Decimal('0.50')
# 📖 Add a new book
def add_book(self, isbn, title, author, category):
emoji_map = {
'fiction': '📖',
'science': '🔬',
'history': '📜',
'technology': '💻',
'children': '🧸'
}
self.cursor.execute("""
INSERT INTO books (isbn, title, author, category, emoji, available)
VALUES (?, ?, ?, ?, ?, 1)
""", (isbn, title, author, category, emoji_map.get(category, '📚')))
self.conn.commit()
return self.cursor.lastrowid
# 👤 Register a member
def register_member(self, name, email):
self.cursor.execute("""
INSERT INTO members (name, email, joined_date, active)
VALUES (?, ?, ?, 1)
""", (name, email, datetime.now()))
self.conn.commit()
return self.cursor.lastrowid
# 📖 Borrow a book
def borrow_book(self, member_id, isbn, days=14):
# 🔍 Check if book is available
self.cursor.execute("""
SELECT id, title, available FROM books WHERE isbn = ?
""", (isbn,))
book = self.cursor.fetchone()
if not book:
raise ValueError("Book not found! 😕")
book_id, title, available = book
if not available:
raise ValueError(f"'{title}' is not available! 📚")
# 🔍 Check member's current loans
self.cursor.execute("""
SELECT COUNT(*) FROM loans
WHERE member_id = ? AND returned_date IS NULL
""", (member_id,))
active_loans = self.cursor.fetchone()[0]
if active_loans >= 3:
raise ValueError("Maximum 3 books allowed! 📚📚📚")
# ✅ Create loan
due_date = datetime.now() + timedelta(days=days)
self.cursor.execute("""
INSERT INTO loans (book_id, member_id, loan_date, due_date)
VALUES (?, ?, ?, ?)
""", (book_id, member_id, datetime.now(), due_date))
# 📉 Mark book as unavailable
self.cursor.execute("""
UPDATE books SET available = 0 WHERE id = ?
""", (book_id,))
self.conn.commit()
return f"Borrowed '{title}'! Due: {due_date.strftime('%Y-%m-%d')} 📅"
# 📚 Return a book
def return_book(self, member_id, isbn):
# 🔍 Find the loan
self.cursor.execute("""
SELECT l.id, l.due_date, b.title
FROM loans l
JOIN books b ON l.book_id = b.id
WHERE b.isbn = ? AND l.member_id = ? AND l.returned_date IS NULL
""", (isbn, member_id))
loan = self.cursor.fetchone()
if not loan:
raise ValueError("No active loan found! 🤔")
loan_id, due_date, title = loan
due_date = datetime.fromisoformat(due_date)
# 💰 Calculate fine if overdue
fine = Decimal('0')
if datetime.now() > due_date:
days_overdue = (datetime.now() - due_date).days
fine = self.fine_per_day * days_overdue
# ✅ Complete return
self.cursor.execute("""
UPDATE loans
SET returned_date = ?, fine = ?
WHERE id = ?
""", (datetime.now(), float(fine), loan_id))
# 📈 Mark book as available
self.cursor.execute("""
UPDATE books SET available = 1
WHERE isbn = ?
""", (isbn,))
self.conn.commit()
if fine > 0:
return f"'{title}' returned. Fine: ${fine:.2f} 💸"
return f"'{title}' returned on time! 🎉"
# 📊 Get most borrowed books
def get_popular_books(self, limit=5):
self.cursor.execute("""
SELECT b.title, b.author, b.emoji, COUNT(l.id) as borrow_count
FROM books b
JOIN loans l ON b.id = l.book_id
GROUP BY b.id
ORDER BY borrow_count DESC
LIMIT ?
""", (limit,))
return self.cursor.fetchall()
# 🧪 Test database fixture
@pytest.fixture
def library_db():
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()
# 📊 Create schema
cursor.executescript("""
CREATE TABLE books (
id INTEGER PRIMARY KEY,
isbn TEXT UNIQUE NOT NULL,
title TEXT NOT NULL,
author TEXT NOT NULL,
category TEXT,
emoji TEXT,
available INTEGER DEFAULT 1
);
CREATE TABLE members (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
joined_date TIMESTAMP,
active INTEGER DEFAULT 1
);
CREATE TABLE loans (
id INTEGER PRIMARY KEY,
book_id INTEGER REFERENCES books(id),
member_id INTEGER REFERENCES members(id),
loan_date TIMESTAMP NOT NULL,
due_date TIMESTAMP NOT NULL,
returned_date TIMESTAMP,
fine REAL DEFAULT 0
);
""")
conn.commit()
yield conn
conn.close()
# ✅ Test successful book borrowing
def test_borrow_book_success(library_db):
library = Library(library_db)
# 📚 Add a book
library.add_book("978-0-13-110362-7", "The Python Book", "Guido van Rossum", "technology")
# 👤 Register a member
member_id = library.register_member("Alice Python", "[email protected]")
# 📖 Borrow the book
result = library.borrow_book(member_id, "978-0-13-110362-7")
assert "Borrowed 'The Python Book'" in result
assert "Due:" in result
# ✅ Verify book is now unavailable
cursor = library_db.cursor()
cursor.execute("SELECT available FROM books WHERE isbn = ?", ("978-0-13-110362-7",))
assert cursor.fetchone()[0] == 0
# 😱 Test borrowing limit
def test_borrow_limit_exceeded(library_db):
library = Library(library_db)
# 👤 Create member
member_id = library.register_member("Bob Reader", "[email protected]")
# 📚 Add and borrow 3 books
for i in range(3):
isbn = f"978-0-{i:02d}-000000-0"
library.add_book(isbn, f"Book {i+1}", "Author", "fiction")
library.borrow_book(member_id, isbn)
# 🚫 Try to borrow 4th book
library.add_book("978-0-99-000000-0", "One Too Many", "Author", "fiction")
with pytest.raises(ValueError) as exc_info:
library.borrow_book(member_id, "978-0-99-000000-0")
assert "Maximum 3 books allowed" in str(exc_info.value)
# 💰 Test overdue fines
def test_overdue_fine_calculation(library_db):
library = Library(library_db)
# Setup
library.add_book("978-1-23-456789-0", "Always Late", "Procrastinator", "fiction")
member_id = library.register_member("Charlie Late", "[email protected]")
# 📅 Manually set loan date to past
cursor = library_db.cursor()
past_date = datetime.now() - timedelta(days=20)
due_date = past_date + timedelta(days=14) # 6 days overdue
cursor.execute("""
INSERT INTO loans (book_id, member_id, loan_date, due_date)
VALUES (1, ?, ?, ?)
""", (member_id, past_date, due_date))
cursor.execute("UPDATE books SET available = 0 WHERE id = 1")
library_db.commit()
# 💸 Return late book
result = library.return_book(member_id, "978-1-23-456789-0")
assert "Fine: $3.00" in result # 6 days × $0.50
# ✅ Verify fine was recorded
cursor.execute("SELECT fine FROM loans WHERE member_id = ?", (member_id,))
assert cursor.fetchone()[0] == 3.0
# 📊 Test popular books report
def test_popular_books_report(library_db):
library = Library(library_db)
# 📚 Create books and members
books = [
("978-0-00-000001-0", "Python Mastery", "Expert Dev", "technology"),
("978-0-00-000002-0", "Web Dev 101", "Full Stack", "technology"),
("978-0-00-000003-0", "Data Science", "ML Expert", "science")
]
for isbn, title, author, category in books:
library.add_book(isbn, title, author, category)
# 👥 Create borrowing history
for i in range(3):
member_id = library.register_member(f"Member {i}", f"member{i}@lib.com")
# Python book is most popular (borrowed by all)
library.borrow_book(member_id, "978-0-00-000001-0")
library.return_book(member_id, "978-0-00-000001-0")
# Web Dev borrowed by 2 members
if i < 2:
library.borrow_book(member_id, "978-0-00-000002-0")
library.return_book(member_id, "978-0-00-000002-0")
# 📊 Get report
popular = library.get_popular_books(limit=2)
assert len(popular) == 2
assert popular[0][0] == "Python Mastery" # Most borrowed
assert popular[0][3] == 3 # Borrowed 3 times
assert popular[1][0] == "Web Dev 101"
assert popular[1][3] == 2 # Borrowed 2 times🎓 Key Takeaways
You’ve learned so much! Here’s what you can now do:
- ✅ Set up test databases for safe, isolated testing 💪
- ✅ Write reliable database tests that don’t interfere with each other 🛡️
- ✅ Use fixtures and factories for consistent test data 🎯
- ✅ Handle transactions properly to keep tests fast and clean 🐛
- ✅ Test complex database operations with confidence! 🚀
Remember: Test databases are your safety net—use them liberally to catch bugs before they reach production! 🤝
🤝 Next Steps
Congratulations! 🎉 You’ve mastered testing database code with test databases!
Here’s what to do next:
- 💻 Practice with the library management exercise above
- 🏗️ Set up test databases for your current projects
- 📚 Learn about database mocking for unit tests
- 🌟 Explore continuous integration with database tests
Keep testing, keep learning, and most importantly, keep your data safe! 🚀
Happy testing! 🎉🚀✨