Mình dùng Python làm automation tool cho hầu hết task hàng ngày, từ deploy script đến monitoring alert. Khi project lớn dần, test suite từ 50 test nhảy lên 300+ test, và mỗi lần git push là ngồi nhìn CI/CD quay vòng gần 10 phút chỉ để chờ test chạy xong. Không phải vì code lỗi — mà vì test chậm thật sự.
Vấn đề thực tế: Test suite 300 bài ngốn 8 phút mỗi lần push
Project cụ thể: một bộ tool Python gồm nhiều module — xử lý file, gọi API bên thứ ba, query database. Test suite khoảng 320 bài, trong đó 60% là integration test kiểm tra logic phức tạp với mock HTTP và file I/O.
Kết quả khi chạy đầy đủ:
$ pytest --tb=no -q
320 passed in 487.23s (8 minutes 7 seconds)8 phút. Nhân lên 10–15 lần push mỗi ngày là hơn 1 tiếng đồng hồ chỉ để chờ test. Trên CI/CD còn tệ hơn vì máy CI thường ít core hơn máy dev.
Phân tích nguyên nhân: Pytest chạy tuần tự theo thiết kế
Pytest mặc định chạy từng test một — test này xong mới tới test kế tiếp. Đây là thiết kế có chủ đích để đảm bảo tính nhất quán và dễ debug, nhưng không tận dụng được CPU đa nhân.
Nhìn vào resource monitor khi test đang chạy trên máy 8 core:
- CPU tổng: 12–15% (1 core đang bận, 7 core nhàn rỗi)
- Disk I/O: spike theo từng test đọc/ghi file
- Network: thỉnh thoảng gọi mock HTTP response
Vấn đề rõ ràng: test không tốn CPU nhiều, chủ yếu tốn I/O wait time — chờ mock response, chờ file write, chờ DB query trả về. Trong khoảng chờ đó, 7 CPU core còn lại không làm gì cả. Đây là lãng phí tài nguyên điển hình.
Các cách giải quyết
Cách 1: Tối ưu từng test riêng lẻ
Bước đầu tiên nên làm bất kể có dùng xdist hay không. Các trick phổ biến:
import pytest
# Đánh dấu test chậm để có thể skip khi cần
@pytest.mark.slow
def test_heavy_integration():
result = call_real_api() # tốn 2-3 giây
assert result["status"] == "ok"# Dev local: bỏ qua test slow
pytest -m "not slow"
# CI full: chạy tất cả
pytestTối ưu xong thì giảm từ 8 phút xuống còn ~6 phút. Vẫn còn chậm, chưa đủ.
Cách 2: Chạy một phần test liên quan đến thay đổi
# Chạy chỉ test trong module vừa sửa
pytest tests/test_api_client.py
# Chạy theo tên keyword
pytest -k "upload or download"
# Chạy test thất bại lần trước trước tiên
pytest --lf # last failed
pytest --lf --nf # last failed, sau đó new filesHữu ích khi dev local nhưng CI/CD vẫn cần chạy full suite. Không giải quyết được vấn đề gốc.
Cách 3: Chạy song song với pytest-xdist
pytest-xdist spawn nhiều worker process, mỗi worker nhận một tập con của test suite và chạy song song với nhau. Đây là cách giải quyết đúng chỗ — thay vì chờ tuần tự, tất cả core CPU cùng làm việc.
Cách tốt nhất: Pytest-xdist từ cài đặt đến thực chiến
Cài đặt
pip install pytest-xdist
# Hoặc thêm vào requirements-dev.txt
pytest-xdist>=3.0Sử dụng cơ bản
# Auto detect số CPU core — đây là lệnh mình dùng nhất
pytest -n auto
# Chỉ định số worker cố định
pytest -n 4
# Kết hợp với verbose output
pytest -n auto -v --tb=shortKết quả ngay sau khi bật:
320 passed in 127.41s (2 minutes 7 seconds)
# Từ 8 phút → 2 phút, nhanh hơn ~4 lầnChiến lược phân phối test với –dist
Đây là thứ mình mất thời gian nhất để tìm hiểu. pytest-xdist có nhiều chiến lược phân phối:
# load (mặc định): worker nào rảnh nhận test tiếp theo
pytest -n 4 --dist=load
# loadscope: group theo class hoặc module — tốt khi dùng class-scoped fixture
pytest -n 4 --dist=loadscope
# loadfile: tất cả test trong cùng file chạy trên cùng 1 worker
pytest -n 4 --dist=loadfileMình dùng --dist=loadfile cho project có nhiều class fixture phức tạp, vì nó đảm bảo test cùng file không bị tách ra worker khác nhau — tránh được khá nhiều race condition.
Fixture và xdist: điểm cần chú ý nhất
Đây là chỗ hay gây bug nhất khi mới dùng xdist. Fixture scope="session" sẽ chạy độc lập trên mỗi worker, không phải shared toàn bộ session.
# Cần port riêng cho mỗi worker để tránh conflict
@pytest.fixture(scope="session")
def server_port(worker_id):
# worker_id: "gw0", "gw1", "gw2"... hoặc "master" khi không dùng xdist
base_port = 8000
if worker_id == "master":
return base_port
worker_num = int(worker_id.replace("gw", ""))
return base_port + worker_num # gw0=8000, gw1=8001, gw2=8002...Với database test, mình dùng tmp_path để mỗi test có DB riêng:
@pytest.fixture
def db(tmp_path):
db_path = tmp_path / "test.db"
conn = sqlite3.connect(str(db_path))
yield conn
conn.close()
# tmp_path tự xóa sau khi test xongKết hợp với pytest-cov
pip install pytest-covFile .coveragerc:
[run]
concurrency = multiprocessing
parallel = true
[report]
omit =
*/tests/*
*/venv/*# Chạy test với coverage
pytest -n auto --cov=src --cov-report=html
# Combine data từ nhiều worker rồi xuất report
coverage combine
coverage report -mTips thực chiến
Test phải hoàn toàn stateless — đây là điều kiện tiên quyết:
# SAI: global state bị share giữa các worker
counter = 0
def test_increment():
global counter
counter += 1
assert counter == 1 # Fail khi chạy song song!
# ĐÚNG: mỗi test tự quản lý state riêng
def test_increment():
counter = 0
counter += 1
assert counter == 1Tránh hardcode port hoặc file path cố định:
import socket
def get_free_port() -> int:
with socket.socket() as s:
s.bind(('', 0))
return s.getsockname()[1]
@pytest.fixture
def app_port():
return get_free_port() # mỗi worker nhận port ngẫu nhiên, không conflictDùng tmp_path thay vì tạo file cố định:
def test_file_processing(tmp_path):
input_file = tmp_path / "input.csv"
input_file.write_text("col1,col2\n1,2\n3,4")
result = process_csv(input_file)
assert len(result) == 2 # 2 hàng dataProfiling để biết test nào chậm nhất:
# Hiển thị 20 test chậm nhất
pytest --durations=20
# Cài pytest-randomly để phát hiện test phụ thuộc thứ tự
pip install pytest-randomly
pytest -n auto # test sẽ được shuffle ngẫu nhiênKết quả thực tế và checklist trước khi bật xdist
Sau khi áp dụng pytest -n auto --dist=loadfile, test suite của mình giảm từ 8 phút → khoảng 2 phút trên máy 8 core. CI pipeline từ 30 phút xuống còn ~12 phút (phần còn lại là build Docker và deploy).
Trước khi bật xdist, check nhanh các điểm này:
- Mỗi test không dùng shared mutable state (global variable, class attribute)
- Không hardcode port hay file path cố định trong test
- Fixture
scope="session"không có side effect ảnh hưởng test khác - Test database dùng transaction rollback hoặc in-memory DB / tmp_path riêng biệt
- Không có test nào phụ thuộc vào thứ tự chạy của test khác
Nếu có test break khi chạy song song, debug bằng cách chạy lại với -n 1 để xác nhận test pass khi tuần tự — nếu -n 1 pass mà -n auto fail thì chắc chắn là do shared state hoặc port/file conflict.
