Bắt Đầu Nhanh: Chạy Được Trong 5 Phút
Sáu tháng trước, một lần migration thất bại đã khiến database production của chúng tôi offline suốt 45 phút. Toàn bộ unit test đều pass — vì chúng chạy trên mock. PostgreSQL thật lại có quan điểm khác về thứ tự constraint. Chính sự cố đó đã thúc đẩy tôi xây dựng một pipeline kiểm thử tích hợp đúng nghĩa với Testcontainers.
Cài đặt các dependency trước:
pip install testcontainers pytest sqlalchemy psycopg2-binary alembicSau đó là một test tối giản để khởi động container PostgreSQL thật:
import pytest
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine, text
@pytest.fixture(scope="session")
def postgres_container():
with PostgresContainer("postgres:16") as pg:
yield pg
def test_basic_connection(postgres_container):
engine = create_engine(postgres_container.get_connection_url())
with engine.connect() as conn:
result = conn.execute(text("SELECT version()"))
version = result.scalar()
assert "PostgreSQL 16" in versionChạy lệnh:
pytest test_db.py -vLần đầu chạy sẽ tải Docker image — khoảng 170MB. Sau đó, container khởi động trong 2–3 giây. Đây là PostgreSQL thật, không phải mock hay SQLite giả vờ đóng vai Postgres. Sự khác biệt này dễ bị xem nhẹ — cho đến khi một migration vượt qua mock lại làm sập database production của bạn.
Tìm Hiểu Sâu: Những Gì Bạn Thực Sự Có Thể Kiểm Thử
Kiểm Thử Migration Từ Đầu Đến Cuối
Chạy migration Alembic trên một database hoàn toàn sạch — đó chính là lúc setup này phát huy giá trị. Unit test bỏ sót rất nhiều thứ: một migration tham chiếu đến cột không tồn tại, cú pháp đặc thù của PostgreSQL mà SQLite nuốt chửng không báo lỗi, thứ tự constraint chỉ quan trọng khi chạy trên engine thật.
import pytest
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine, text
from alembic.config import Config
from alembic import command
@pytest.fixture(scope="session")
def migrated_db():
with PostgresContainer("postgres:16") as pg:
engine = create_engine(pg.get_connection_url())
alembic_cfg = Config("alembic.ini")
alembic_cfg.set_main_option("sqlalchemy.url", pg.get_connection_url())
command.upgrade(alembic_cfg, "head")
yield engine
def test_users_table_schema(migrated_db):
with migrated_db.connect() as conn:
result = conn.execute(text(
"SELECT column_name, data_type "
"FROM information_schema.columns "
"WHERE table_name = 'users' ORDER BY ordinal_position"
))
columns = {row[0]: row[1] for row in result}
assert "id" in columns
assert "email" in columns
assert columns["created_at"] == "timestamp with time zone"Tôi đã phát hiện được ba lỗi migration theo cách này — những lỗi đáng lẽ phải can thiệp thủ công trên production: một foreign key constraint trỏ sai chiều, thiếu index trên cột lookup có lượng truy cập cao, và một VARCHAR(255) ở chỗ chúng tôi cần TEXT. Không cái nào trong số đó bị phát hiện trong các test dùng mock.
Kiểm Thử Logic Truy Vấn Trên Dữ Liệu Thật
Database mock cho phép bạn xác nhận rằng một truy vấn đã được gọi. Database thật cho phép bạn xác nhận rằng truy vấn đó trả về đúng dữ liệu. Đó là hai điều hoàn toàn khác nhau.
from sqlalchemy.orm import Session
from your_app.models import User, Order
@pytest.fixture
def db_session(migrated_db):
with Session(migrated_db) as session:
yield session
session.rollback() # Dọn dẹp sau mỗi test
def test_user_total_completed_orders(db_session):
# Chuẩn bị: chèn dữ liệu test thật
user = User(email="[email protected]", name="Test User")
db_session.add(user)
for i in range(3):
order = Order(user=user, amount=10.00 * (i + 1), status="completed")
db_session.add(order)
db_session.add(Order(user=user, amount=99.00, status="pending"))
db_session.flush()
# Thực thi: chạy truy vấn thực tế từ tầng ứng dụng
result = db_session.execute(text(
"SELECT SUM(amount) FROM orders "
"WHERE user_id = :uid AND status = 'completed'"
), {"uid": user.id}).scalar()
# Kiểm tra: đơn hàng pending KHÔNG được tính vào
assert result == 60.00Lệnh session.rollback() trong teardown của fixture giúp mỗi test hoạt động độc lập mà không cần xóa và tạo lại schema. Nhanh và ổn định.
Nâng Cao: Xây Dựng Pipeline CI/CD Hoàn Chỉnh
Container Session-Scoped Kết Hợp Isolation Từng Test
Năm mươi database test chạy tuần tự sẽ làm CI của bạn bò như rùa. Một container cho cả session — migration chạy một lần — kết hợp transaction rollback từng test để tách biệt dữ liệu. Đây là pattern đó:
# conftest.py
import pytest
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from alembic.config import Config
from alembic import command
@pytest.fixture(scope="session")
def db_engine():
with PostgresContainer("postgres:16") as pg:
engine = create_engine(
pg.get_connection_url(),
pool_size=10,
max_overflow=20
)
alembic_cfg = Config("alembic.ini")
alembic_cfg.set_main_option("sqlalchemy.url", pg.get_connection_url())
command.upgrade(alembic_cfg, "head")
yield engine
@pytest.fixture
def db_session(db_engine):
connection = db_engine.connect()
transaction = connection.begin()
session = Session(bind=connection)
yield session
session.close()
transaction.rollback()
connection.close()Thêm pytest-xdist để chạy song song:
pip install pytest-xdist
pytest -n 4 tests/integration/Pattern này đã giúp bộ test của chúng tôi giảm từ 8 phút xuống còn 90 giây.
Tích Hợp GitHub Actions
Testcontainers cần Docker. GitHub Actions đã có sẵn Docker, nên cấu hình CI khá gọn:
# .github/workflows/db-tests.yml
name: Database Integration Tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Cài đặt Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Cài đặt dependency
run: pip install -r requirements.txt
- name: Chạy database integration tests
run: pytest tests/integration/ -v --tb=short
env:
TESTCONTAINERS_RYUK_DISABLED: "false"Kiểm Thử Rollback Migration
Rollback khẩn cấp vốn đã căng thẳng. Phát hiện script downgrade báo lỗi giữa chừng sự cố còn tệ hơn nữa. Hãy kiểm thử mọi migration theo cả hai chiều:
def test_migration_upgrade_and_downgrade(postgres_container):
engine = create_engine(postgres_container.get_connection_url())
alembic_cfg = Config("alembic.ini")
alembic_cfg.set_main_option("sqlalchemy.url", postgres_container.get_connection_url())
# Nâng lên phiên bản mới nhất
command.upgrade(alembic_cfg, "head")
# Hạ xuống một bước — không được báo lỗi
command.downgrade(alembic_cfg, "-1")
# Nâng lại để kiểm tra tính idempotent
command.upgrade(alembic_cfg, "head")
with engine.connect() as conn:
result = conn.execute(text("SELECT COUNT(*) FROM alembic_version"))
assert result.scalar() == 1Kinh Nghiệm Thực Tế Sau 6 Tháng Chạy Production
Phân biệt rõ scope của fixture. Dùng scope="session" cho container và migration (tốn kém nếu lặp lại), và scope="function" với transaction rollback để tách biệt dữ liệu (chi phí thấp). Sự kết hợp này chính là thứ giúp bộ test của chúng tôi giảm từ 8 phút xuống 90 giây.
Cố định phiên bản database trong test. Dùng postgres:16 thay vì postgres:latest. Test sẽ không bị vỡ ngẫu nhiên khi có phiên bản major mới ra. Khớp chính xác với phiên bản đang chạy trên production.
Viết test kiểm tra schema song song với mỗi migration. Mọi migration vào nhánh main đều nên có một test tương ứng xác nhận trạng thái schema mong đợi. Đây trở thành tài liệu sống ghi lại quá trình tiến hóa của database — và phát hiện regression trước khi thoát khỏi CI.
# Test thêm cùng với migration 005_add_user_preferences.py
def test_migration_005_schema(migrated_db):
with migrated_db.connect() as conn:
result = conn.execute(text(
"SELECT data_type FROM information_schema.columns "
"WHERE table_name = 'user_preferences' AND column_name = 'settings'"
))
data_type = result.scalar()
# Kiểm tra rõ ràng là jsonb, không phải json — cần hỗ trợ GIN index
assert data_type == "jsonb"Kiểm thử cả script seed data. Nếu ứng dụng của bạn đi kèm dữ liệu tham chiếu hoặc cấu hình mặc định, hãy tạo fixture chạy các script đó và xác nhận số lượng bản ghi. Chúng tôi đã phát hiện một script seed bị hỏng theo cách này trước khi nó lên staging.
Chuẩn bị fixture dữ liệu test cẩn thận. Đôi khi tôi cần bộ dữ liệu edge-case cho integration test — dữ liệu tham chiếu từ file CSV export hoặc snapshot production đã được ẩn danh hóa. Khi cần chuyển đổi nhanh CSV sang JSON để import dữ liệu, tôi dùng toolcraft.app/vi/tools/data/csv-to-json — chạy hoàn toàn trên trình duyệt, dữ liệu không đi đâu cả. Điều đó quan trọng khi CSV chứa dữ liệu test đã được giả danh mà bạn không thể upload lên đâu.
Thêm cơ chế retry khi khởi động container trên CI. Trên GitHub Actions, Docker đôi khi cần thêm vài giây để sẵn sàng. Hãy bọc fixture container trong một retry đơn giản nếu bạn thấy lỗi khởi động không ổn định trên runner mới.
Sáu tháng pipeline này chạy trên mọi pull request. Chúng tôi đã phát hiện lỗi schema, constraint sai chiều, thiếu index, và hai migration pass trên SQLite local nhưng nổ tung trên PostgreSQL thật ở staging. Pipeline thêm khoảng 90 giây vào CI. Nó đã giúp chúng tôi tránh được ít nhất bốn sự cố production. Hoàn toàn xứng đáng.
