Các mẫu pytest fixtures có khả năng mở rộng thực tế: Bài học từ 2 năm vận hành CI

Mô hình pytest fixtures giúp mở rộng quy mô thực tế: Các mẫu thiết kế từ 2 năm vận hành pipeline Python CI

Tôi đã dành hai năm để chứng kiến cùng một loạt thất bại trong bộ kiểm thử (test suite) trên CI.

Đó không phải là lỗi logic — logic của bài kiểm tra thì hoàn toàn ổn. Đó là lỗi cơ sở hạ tầng. Là fixture tạo ra bảng cơ sở dữ liệu nhưng không dọn dẹp nó sau khi chạy xong. Là mock S3 chia sẻ trạng thái (state) giữa các file kiểm thử, khiến bộ test chạy locally tốt nhưng trên CI lại thất bại không rõ nguyên nhân. Là fixture mất tới 4 giây để thiết lập vì nó tạo ra toàn bộ schema, và lại chạy tới 200 lần trên toàn bộ bộ kiểm thử.

Nguyên nhân gốc rễ của những thất bại này luôn như nhau: các fixtures được tối ưu hóa cho tiêu chí "chạy được" thay vì tập trung vào sự cô lập, tốc độ và độ tin cậy dưới mô hình thực thi song song của CI. Một khi tôi hiểu rõ các mẫu (patterns) gây ra vấn đề này, tôi có thể nhận thấy chúng xuất hiện trong mọi mã nguồn mà tôi chạm vào.

Dưới đây là bốn mẫu mà tôi hiện đang sử dụng. Chúng không yêu cầu bất kỳ phụ thuộc mới nào — chỉ cần một mô hình rõ ràng về cách hoạt động của cơ chế phạm vi (scope) và dọn dẹp (teardown) trong pytest.

Vấn đề về phạm vi (Scope) và lý do phạm vi Function nên là mặc định

Pytest fixtures có bốn cấp độ phạm vi: function (hàm), class (lớp), module (mô-đun) và session (phiên làm việc). Mặc định là function. Sai lầm phổ biến nhất tôi thấy là nâng cấp lên phạm vi module hoặc session quá sớm để "tăng tốc độ".

Lợi ích về hiệu suất là có thật. Một kết nối cơ sở dữ liệu ở phạm vi session nhanh hơn việc tạo lại nó 200 lần. Tuy nhiên, chi phí ẩn nằm ở sự cô lập của bài kiểm thử — giả định rằng mỗi bài kiểm thử đều bắt đầu với một trạng thái sạch sẽ.

Các fixtures phạm vi session chia sẻ trạng thái trên toàn bộ quá trình chạy kiểm thử. Nếu kiểm thử A sửa đổi trạng thái chia sẻ và kiểm thử B đọc nó, bạn sẽ có một sự phụ thuộc vào thứ tự chạy. Các bài kiểm thử sẽ chạy lẻ (pass) khi chạy riêng lẻ nhưng thất bại khi chạy cùng nhau — hoặc tệ hơn, chạy được ở trình tự này nhưng thất bại ở trình tự khác. Tính song song của CI làm cho điều này trở nên khó dự đoán.

Quy tắc tôi tuân theo:

• Thiết lập tốn kém, chỉ đọc (read-only) → phạm vi session (tải file cấu hình, thiết lập pool kết nối, đọc dữ liệu fixture từ ổ đĩa)
• Thiết lập có trạng thái (stateful) → phạm vi function (bất kỳ thứ gì tạo, sửa đổi hoặc xóa bản ghi)
• Đặt lại (reset) rõ ràng thay vì nâng cấp phạm vi → luôn luôn

Nếu bạn muốn phạm vi module hoặc session để tối ưu hiệu suất nhưng cần trạng thái sạch cho mỗi kiểm thử, yield + teardown là công cụ phù hợp:

@pytest.fixture(scope="module")
def db_connection():
    conn = create_connection()
    yield conn
    conn.close()  # dọn dẹp sau khi module hoàn thành

Đối với tài nguyên có trạng thái, phạm vi function + thiết lập nhanh tốt hơn là phạm vi session + sự cô lập kém:

@pytest.fixture(scope="function")
def clean_table(db_connection):
    """Xóa (truncate) và điền lại dữ liệu kiểm thử cho mỗi test."""
    db_connection.execute("TRUNCATE test_records")
    db_connection.execute("INSERT INTO test_records VALUES ...")
    yield db_connection
    # teardown ngầm định — việc xóa ở đầu test sau sẽ xử lý nó

Layered fixtures: Xây dựng sự phức tạp từ các mảnh đơn giản

Mẫu anti-pattern phổ biến nhất mà tôi thấy là một fixture khổng lồ thiết lập mọi thứ mà một bài kiểm thử có thể cần, ngay cả khi kiểm thử đó chỉ cần một phần nhỏ.

# Anti-pattern: monolithic fixture (độc khối)
@pytest.fixture
def full_environment(aws_credentials, s3_bucket, dynamodb_table, sqs_queue, test_user):
    # 40 dòng thiết lập
    yield FullEnv(s3=s3_bucket, db=dynamodb_table, ...)

Vấn đề ở đây: mọi bài kiểm thử sử dụng full_environment đều phải trả chi phí thiết lập đầy đủ, ngay cả khi nó chỉ cần S3. Và khi thiết lập SQS thất bại, nó làm hỏng những bài kiểm thử thậm chí không dùng đến SQS.

Giải pháp thay thế: hãy phân tầng fixtures từ hẹp đến rộng, và để pytest xử lý việc kết hợp:

@pytest.fixture
def aws_credentials(monkeypatch):
    """Mock thông tin xác thực AWS tối thiểu. Mọi test AWS đều cần cái này."""
    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
    monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")


@pytest.fixture
@mock_aws
def s3_client(aws_credentials):
    """Client S3. Phụ thuộc vào thông tin xác thực, không có gì khác."""
    return boto3.client("s3", region_name="us-east-1")


@pytest.fixture
@mock_aws
def test_bucket(s3_client):
    """Bucket S3 được tạo sẵn. Phụ thuộc vào s3_client."""
    s3_client.create_bucket(Bucket="test-bucket")
    return "test-bucket"

Bây giờ, một bài kiểm thử chỉ cần S3 sẽ yêu cầu test_bucket. Một bài kiểm thử cần DynamoDB sẽ yêu cầu dynamodb_table. Chúng nhận được đúng những gì chúng cần, không thừa không thiếu. pytest tự động giải quyết cây phụ thuộc.

Lợi ích: các lỗi được cô lập. Nếu thiết lập SQS bị lỗi, chỉ những bài kiểm thử phụ thuộc vào SQS thất bại. Phần còn lại của bộ test vẫn tiếp tục chạy.

Vấn đề về ranh giới của Mock (Mock boundary)

Với moto (hoặc bất kỳ thư viện mock nào), phạm vi của mock quan trọng ngang với phạm vi của fixture.

Decorator @mock_aws chặn các gọi boto3 ở cấp độ hàm. Nếu bạn áp dụng nó cho fixture, mock chỉ hoạt động bên trong mã thiết lập của fixture đó:

# Lỗi: mock_aws chỉ hoạt động trong quá trình thiết lập fixture, không phải trong test
@pytest.fixture
@mock_aws
def s3_client(aws_credentials):
    client = boto3.client("s3")
    client.create_bucket(Bucket="test-bucket")  # cái này hoạt động
    return client
    # context mock_aws kết thúc ở đây

def test_upload(s3_client):
    s3_client.put_object(...)  # các gọi boto3 ở đây KHÔNG được mock — gọi AWS thật

Mẫu đúng là quản lý ngữ cảnh (context) mock trong fixture bằng cách sử dụng with hoặc chuyển context manager vào test một cách rõ ràng:

@pytest.fixture
def s3_mock():
    """Cung cấp ngữ cảnh mock_aws hoạt động trong suốt thời gian chạy test."""
    with mock_aws():
        yield


@pytest.fixture
def s3_client(aws_credentials, s3_mock):
    """Client S3 nằm trong ngữ cảnh mock đang hoạt động."""
    client = boto3.client("s3", region_name="us-east-1")
    client.create_bucket(Bucket="test-bucket")
    return client


def test_upload(s3_client):
    # s3_mock hoạt động trong suốt thời gian của test này
    s3_client.put_object(Bucket="test-bucket", Key="key.txt", Body=b"content")
    response = s3_client.get_object(Bucket="test-bucket", Key="key.txt")
    assert response["Body"].read() == b"content"

Fixture s3_mock giữ ngữ cảnh mock_aws() mở cho toàn bộ bài kiểm thử. s3_client phụ thuộc vào s3_mock, vì vậy pytest đảm bảo mock được kích hoạt trước khi thiết lập client.

Mẫu này loại bỏ lớp lỗi "chạy tốt trên máy local, thất bại trên CI khi thứ tự test thay đổi".

Kiến trúc conftest.py: Tập trung hóa các fixtures có thể chia sẻ

Nếu bạn có các AWS fixtures rải rác trên nhiều file kiểm thử, bạn đang tạo lại cùng một mã thiết lập ở khắp nơi. Bất kỳ thay đổi nào — thêm vùng mới, đổi phạm vi mock, thêm dọn dẹp — đều phải thực hiện ở nhiều nơi.

Giải pháp là một file conftest.py ở mức phù hợp. pytest tự động phát hiện các file conftest.py và làm cho các fixtures của chúng khả dụng cho tất cả các kiểm thử trong cùng thư mục và các thư mục con.

Đối với cấu trúc dự án điển hình:

project/
├── conftest.py          # Fixtures phạm vi dự án: aws_credentials, s3_mock, sqs_mock
├── tests/
│   ├── conftest.py      # Fixtures phạm vi bộ test cần root dự án
│   ├── test_ingestion/
│   │   ├── conftest.py  # Fixtures cụ thể mô-đun: bảng điền sẵn, dữ liệu test
│   │   └── test_*.py
│   └── test_processing/
│       ├── conftest.py
│       └── test_*.py

File conftest.py gốc chứa các fixtures cơ sở hạ tầng — mock AWS, kết nối cơ sở dữ liệu, thiết lập tốn kém phạm vi session. Các file conftest.py cấp mô-đun chứa thiết lập cụ thể theo lĩnh vực — dữ liệu test, bảng điền sẵn, các client cụ thể theo dịch vụ.

Việc phân tầng này có nghĩa là:

• Thêm một file test mới trong test_ingestion/ tự động nhận được các mocks AWS.
• Thay đổi cấu hình mock chỉ cần sửa một file.
• Các fixtures cụ thể mô-đun không làm ô nhiễm không gian tên (namespace) của các mô-đun khác.

# conftest.py (gốc dự án)
import boto3
import pytest
from moto import mock_aws


@pytest.fixture(scope="function")
def aws_credentials(monkeypatch):
    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
    monkeypatch.setenv("AWS_SECURITY_TOKEN", "testing")
    monkeypatch.setenv("AWS_SESSION_TOKEN", "testing")
    monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")


@pytest.fixture(scope="function")
def aws_mock(aws_credentials):
    """Ngữ cảnh mock_aws hoạt động cho toàn bộ test."""
    with mock_aws():
        yield


@pytest.fixture(scope="function")
def s3_client(aws_mock):
    return boto3.client("s3", region_name="us-east-1")


@pytest.fixture(scope="function")
def dynamodb_client(aws_mock):
    return boto3.client("dynamodb", region_name="us-east-1")


@pytest.fixture(scope="function")
def sqs_client(aws_mock):
    return boto3.client("sqs", region_name="us-east-1")

Tổng hợp: Một bài kiểm thử đáng tin cậy trên CI

Một bài kiểm thử sử dụng cả bốn mẫu — phạm vi function, fixtures phân tầng, ranh giới mock đúng, kiến trúc conftest:

# tests/test_ingestion/conftest.py
@pytest.fixture
def ingestion_bucket(s3_client):
    """Bucket đã tạo sẵn với cấu trúc tiền tố test."""
    bucket_name = "test-ingestion-bucket"
    s3_client.create_bucket(Bucket=bucket_name)
    s3_client.put_object(
        Bucket=bucket_name,
        Key="raw/2026-03-24/event_001.json",
        Body=b'{"event": "test", "timestamp": 1711238400}'
    )
    return bucket_name

# tests/test_ingestion/test_processor.py
def test_processes_event_file(s3_client, ingestion_bucket):
    """Kiểm thử xử lý một file sự kiện từ S3."""
    # Arrange: file đã có trong bucket thông qua fixture

    # Act
    result = process_s3_event(
        bucket=ingestion_bucket,
        key="raw/2026-03-24/event_001.json",
        s3_client=s3_client
    )

    # Assert
    assert result.status == "processed"
    assert result.event_count == 1

    # Kiểm tra tác động phụ trong S3
    response = s3_client.list_objects_v2(
        Bucket=ingestion_bucket, Prefix="processed/"
    )
    assert response["KeyCount"] == 1

Bài kiểm thử này:

• Có môi trường S3 sạch, cô lập (phạm vi function).
• Mock bao phủ toàn bộ thực thi kiểm thử (ranh giới mock đúng).
• Thiết lập cụ thể theo lĩnh vực nằm trong conftest mô-đun (fixtures phân tầng).
• Mock AWS chung được kế thừa từ conftest gốc (kiến trúc conftest).

Nếu bạn chạy bài kiểm thử này 50 lần song song trên CI, nó sẽ tạo ra cùng một kết quả mỗi lần.