Kiến trúc Testing — Từ Unit đến End-to-End

Một đội ngũ có 5.000 tests, coverage đạt 92%, CI pipeline xanh lè mỗi commit. Nghe hoàn hảo — cho đến khi họ deploy lên production và phát hiện hai microservices giao tiếp sai format JSON. Không test nào bắt được lỗi đó. Coverage cao không đồng nghĩa với confidence cao — kiến trúc test mới quyết định bạn có ngủ ngon sau mỗi lần deploy hay không.

Viết một test đúng thì dễ. Thiết kế một chiến lược testing khiến toàn bộ hệ thống đáng tin cậy — đó mới là kỹ năng của senior engineer. Bài viết này trang bị cho bạn tư duy kiến trúc testing: biết đặt bao nhiêu test ở mỗi tầng, biết khi nào mock và khi nào dùng real dependency, biết thiết kế pipeline CI chạy nhanh mà vẫn phát hiện lỗi sớm.

Quick win: Nếu bạn chỉ nhớ một điều — hãy nhớ hình kim tự tháp: nhiều unit test nhanh ở đáy, ít e2e test chậm ở đỉnh. Tỷ lệ 70/20/10 là điểm khởi đầu tốt cho hầu hết dự án.

Bức tranh tư duy

Testing Pyramid — Kim tự tháp kiểm thử

Hình dung kiến trúc testing như quy trình kiểm soát chất lượng trong nhà máy sản xuất. Ở công đoạn đầu (unit test), mỗi linh kiện được kiểm tra riêng lẻ — nhanh, rẻ, phát hiện lỗi sớm. Ở công đoạn giữa (integration test), các module được lắp ghép và kiểm tra tương tác giữa chúng. Ở công đoạn cuối (e2e test), sản phẩm hoàn chỉnh chạy qua dây chuyền kiểm tra toàn diện — chậm, tốn kém, nhưng phát hiện lỗi mà các công đoạn trước bỏ sót.

              /\
             /  \
            / E2E \           ← Ít tests (≈10%), chậm, đắt
           /  Tests \            phát hiện lỗi tích hợp toàn hệ thống
          /----------\
         / Integration \      ← Vừa phải (≈20%), tốc độ trung bình
        /    Tests      \        kiểm tra tương tác giữa các module
       /----------------\
      /    Unit Tests     \   ← Nhiều tests (≈70%), cực nhanh
     /                     \     kiểm tra từng hàm/class riêng lẻ
    /───────────────────────\

Tầng	Số lượng	Tốc độ	Chi phí	Phạm vi
Unit	Nhiều (≈70%)	Nhanh (ms)	Thấp	Một hàm hoặc class
Integration	Vừa (≈20%)	Trung bình (s)	Trung bình	Nhiều component kết nối
E2E	Ít (≈10%)	Chậm (phút)	Cao	Toàn bộ hệ thống

Khi nào kim tự tháp không còn phù hợp?

Kim tự tháp là mô hình kinh điển nhưng không phải chân lý tuyệt đối. Hai mô hình thay thế đáng cân nhắc:

Testing Trophy (Kent C. Dodds) — phình ở tầng integration. Lý do: với frontend hiện đại hoặc hệ thống API-centric, integration test cho ROI (return on investment) cao nhất vì chúng kiểm tra hành vi thực tế của người dùng mà không chậm như e2e.

         /\          E2E (ít)
        /  \
       /----\
      / Integ \      ← Tầng lớn nhất trong Trophy
     / ration  \
    /    Tests  \
   /─────────────\
   \  Unit Tests /   ← Vẫn có nhưng không nhiều bằng Integration
    \───────────/
     \ Static /      ← Type checking, linting
      \──────/

Testing Honeycomb (Spotify) — dành cho microservices. Tầng integration testing giữa các service (contract tests, API tests) chiếm tỷ trọng lớn nhất, vì ranh giới giữa các service chính là nơi lỗi hay xảy ra nhất.

Nguyên tắc chọn mô hình: Không có mô hình nào đúng cho mọi kiến trúc. Phân tích xem lỗi trong hệ thống của bạn hay xảy ra ở đâu — unit logic, tích hợp module, hay giao tiếp service — rồi đầu tư test nhiều nhất ở tầng đó.

Cốt lõi kỹ thuật

Unit Tests — Kiểm tra đơn vị

Unit test kiểm tra một đơn vị logic (hàm, method, class) trong trạng thái hoàn toàn cô lập — không database, không network, không filesystem. Mục tiêu: chạy nhanh (dưới 100ms mỗi test), xác định chính xác nơi code bị hỏng.

python

# domain/pricing.py
from decimal import Decimal
from dataclasses import dataclass


@dataclass(frozen=True)
class OrderItem:
    product_id: str
    quantity: int
    unit_price: Decimal


def calculate_total(
    items: list[OrderItem],
    discount_percent: Decimal = Decimal("0"),
) -> Decimal:
    """Tính tổng đơn hàng sau giảm giá."""
    if not items:
        raise ValueError("Đơn hàng phải có ít nhất một sản phẩm")
    if not (Decimal("0") <= discount_percent <= Decimal("100")):
        raise ValueError(f"Giảm giá không hợp lệ: {discount_percent}%")

    subtotal = sum(
        item.unit_price * item.quantity for item in items
    )
    discount = subtotal * discount_percent / Decimal("100")
    return (subtotal - discount).quantize(Decimal("0.01"))

python

# tests/unit/test_pricing.py
import pytest
from decimal import Decimal
from domain.pricing import calculate_total, OrderItem


class TestCalculateTotal:
    """Unit tests cho calculate_total — không dependency ngoài."""

    def test_single_item_no_discount(self):
        items = [OrderItem("SKU-001", quantity=2, unit_price=Decimal("49.99"))]
        assert calculate_total(items) == Decimal("99.98")

    def test_multiple_items_with_discount(self):
        items = [
            OrderItem("SKU-001", quantity=1, unit_price=Decimal("100.00")),
            OrderItem("SKU-002", quantity=3, unit_price=Decimal("25.00")),
        ]
        result = calculate_total(items, discount_percent=Decimal("10"))
        assert result == Decimal("157.50")

    def test_empty_order_raises(self):
        with pytest.raises(ValueError, match="ít nhất một sản phẩm"):
            calculate_total([])

    def test_invalid_discount_raises(self):
        items = [OrderItem("SKU-001", quantity=1, unit_price=Decimal("10.00"))]
        with pytest.raises(ValueError, match="không hợp lệ"):
            calculate_total(items, discount_percent=Decimal("150"))

Integration Tests — Kiểm tra tích hợp

Integration test kiểm tra tương tác giữa nhiều component thực: code kết nối database thật (thường qua container), code gọi API thật (sandbox), code đọc/ghi filesystem. Chậm hơn unit test nhưng phát hiện lỗi ở ranh giới giữa các module.

python

# tests/integration/test_order_repository.py
import pytest
from decimal import Decimal
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from domain.pricing import OrderItem
from infrastructure.database import Base
from infrastructure.repositories import OrderRepository


@pytest.fixture(scope="module")
def db_engine():
    """Tạo SQLite in-memory database cho integration test."""
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    yield engine
    engine.dispose()


@pytest.fixture
def db_session(db_engine):
    """Mỗi test nhận transaction riêng, rollback sau khi chạy."""
    connection = db_engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()


class TestOrderRepository:
    """Integration test — code tương tác với database thật."""

    def test_save_and_retrieve_order(self, db_session):
        repo = OrderRepository(db_session)
        items = [
            OrderItem("SKU-001", quantity=2, unit_price=Decimal("49.99")),
        ]
        order_id = repo.create_order(customer_id="CUST-123", items=items)

        retrieved = repo.get_order(order_id)
        assert retrieved is not None
        assert retrieved.customer_id == "CUST-123"
        assert len(retrieved.items) == 1

    def test_list_orders_by_customer(self, db_session):
        repo = OrderRepository(db_session)
        items = [OrderItem("SKU-001", quantity=1, unit_price=Decimal("10.00"))]

        repo.create_order(customer_id="CUST-A", items=items)
        repo.create_order(customer_id="CUST-A", items=items)
        repo.create_order(customer_id="CUST-B", items=items)

        orders = repo.list_orders(customer_id="CUST-A")
        assert len(orders) == 2

E2E Tests — Kiểm tra toàn diện

E2E test chạy toàn bộ hệ thống từ đầu đến cuối, mô phỏng hành vi thực tế của người dùng hoặc client. Với backend API, e2e test gọi HTTP endpoint thật, qua tất cả middleware, xử lý logic, lưu database, và kiểm tra response.

python

# tests/e2e/test_order_workflow.py
import pytest
import httpx


@pytest.fixture(scope="session")
def api_client():
    """HTTP client kết nối tới server chạy trong test environment."""
    with httpx.Client(base_url="http://localhost:8000") as client:
        yield client


class TestOrderWorkflowE2E:
    """E2E test — chạy toàn bộ flow từ API đến database."""

    def test_complete_order_lifecycle(self, api_client):
        # 1. Tạo đơn hàng
        create_response = api_client.post("/api/orders", json={
            "customer_id": "CUST-E2E-001",
            "items": [
                {"product_id": "SKU-001", "quantity": 2, "unit_price": "49.99"},
            ],
        })
        assert create_response.status_code == 201
        order_id = create_response.json()["order_id"]

        # 2. Lấy thông tin đơn hàng
        get_response = api_client.get(f"/api/orders/{order_id}")
        assert get_response.status_code == 200
        order_data = get_response.json()
        assert order_data["customer_id"] == "CUST-E2E-001"
        assert order_data["status"] == "pending"
        assert order_data["total"] == "99.98"

        # 3. Xác nhận đơn hàng
        confirm_response = api_client.post(
            f"/api/orders/{order_id}/confirm"
        )
        assert confirm_response.status_code == 200
        assert confirm_response.json()["status"] == "confirmed"

Test Doubles — Mock, Stub, Fake, Spy

Test doubles là các đối tượng thay thế dependency thật trong quá trình testing. Mỗi loại phục vụ mục đích khác nhau:

Loại	Mục đích	Ví dụ
Stub	Trả về dữ liệu cố định, không kiểm tra cách gọi	Trả về user giả từ database
Mock	Kiểm tra hàm được gọi đúng cách (đúng args, đúng số lần)	Verify email service được gọi 1 lần
Fake	Triển khai đơn giản hóa của dependency thật	In-memory database thay SQLite
Spy	Ghi lại thông tin gọi hàm nhưng vẫn chạy logic thật	Đếm số lần cache hit

python

# Fake — triển khai đơn giản thay thế dependency thật
from typing import Protocol


class EmailSender(Protocol):
    """Interface cho email service."""
    def send(self, to: str, subject: str, body: str) -> bool: ...


class FakeEmailSender:
    """Fake implementation — lưu email vào memory thay vì gửi thật."""

    def __init__(self):
        self.sent_emails: list[dict] = []

    def send(self, to: str, subject: str, body: str) -> bool:
        self.sent_emails.append({
            "to": to, "subject": subject, "body": body,
        })
        return True

    @property
    def last_email(self) -> dict | None:
        return self.sent_emails[-1] if self.sent_emails else None


# Spy — wrap implementation thật để ghi lại thông tin
class SpyCache:
    """Spy wrapper — delegate sang cache thật, đếm hit/miss."""

    def __init__(self, real_cache):
        self._real = real_cache
        self.hits = 0
        self.misses = 0

    def get(self, key: str):
        result = self._real.get(key)
        if result is not None:
            self.hits += 1
        else:
            self.misses += 1
        return result

    def set(self, key: str, value, ttl: int = 300):
        return self._real.set(key, value, ttl)

Mocking Strategies — unittest.mock và pytest-mock

unittest.mock là thư viện chuẩn của Python để tạo mock objects. pytest-mock cung cấp fixture mocker tiện lợi hơn. Nguyên tắc vàng: mock ở ranh giới, không mock logic nội bộ.

python

# service/order_service.py
from domain.pricing import calculate_total, OrderItem
from infrastructure.repositories import OrderRepository
from infrastructure.email import EmailSender


class OrderService:
    def __init__(
        self,
        repository: OrderRepository,
        email_sender: EmailSender,
    ):
        self._repo = repository
        self._email = email_sender

    def place_order(
        self,
        customer_id: str,
        customer_email: str,
        items: list[OrderItem],
    ) -> str:
        total = calculate_total(items)
        order_id = self._repo.create_order(customer_id, items)
        self._email.send(
            to=customer_email,
            subject=f"Xác nhận đơn hàng #{order_id}",
            body=f"Tổng: {total} VNĐ",
        )
        return order_id

python

# tests/unit/test_order_service.py
from unittest.mock import MagicMock, patch
from decimal import Decimal

from domain.pricing import OrderItem
from service.order_service import OrderService


class TestOrderService:
    """Mock external dependencies, test business logic."""

    def test_place_order_calls_repository_and_email(self):
        # Arrange — tạo mock cho dependencies
        mock_repo = MagicMock()
        mock_repo.create_order.return_value = "ORD-001"
        mock_email = MagicMock()

        service = OrderService(
            repository=mock_repo,
            email_sender=mock_email,
        )
        items = [OrderItem("SKU-001", quantity=1, unit_price=Decimal("100.00"))]

        # Act
        order_id = service.place_order(
            customer_id="CUST-001",
            customer_email="user@example.com",
            items=items,
        )

        # Assert — kiểm tra hành vi, không kiểm tra implementation
        assert order_id == "ORD-001"
        mock_repo.create_order.assert_called_once_with("CUST-001", items)
        mock_email.send.assert_called_once()
        email_call = mock_email.send.call_args
        assert email_call.kwargs["to"] == "user@example.com"
        assert "ORD-001" in email_call.kwargs["subject"]


class TestOrderServiceWithPytestMock:
    """Sử dụng pytest-mock fixture — cú pháp gọn hơn."""

    def test_place_order_with_mocker(self, mocker):
        mock_repo = mocker.MagicMock()
        mock_repo.create_order.return_value = "ORD-002"
        mock_email = mocker.MagicMock()

        service = OrderService(repository=mock_repo, email_sender=mock_email)
        items = [OrderItem("SKU-001", quantity=1, unit_price=Decimal("50.00"))]

        order_id = service.place_order("CUST-002", "buyer@example.com", items)
        assert order_id == "ORD-002"
        assert mock_email.send.call_count == 1

Test Isolation Patterns

Test isolation đảm bảo mỗi test không ảnh hưởng lẫn nhau — thất bại của test A không gây thất bại test B. Ba pattern chính:

python

# Pattern 1: Transaction Rollback — mỗi test chạy trong transaction riêng
@pytest.fixture
def isolated_session(db_engine):
    """Rollback toàn bộ thay đổi sau mỗi test."""
    connection = db_engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()


# Pattern 2: Temporary Directory — filesystem isolation
@pytest.fixture
def work_dir(tmp_path):
    """Mỗi test nhận thư mục tạm riêng biệt."""
    config_file = tmp_path / "config.yaml"
    config_file.write_text("debug: true\nlog_level: INFO\n")
    return tmp_path


def test_read_config(work_dir):
    config = load_config(work_dir / "config.yaml")
    assert config["debug"] is True


# Pattern 3: Environment Variable Isolation
@pytest.fixture
def clean_env(monkeypatch):
    """Đảm bảo environment variables không bị leak giữa các test."""
    monkeypatch.setenv("DATABASE_URL", "sqlite:///:memory:")
    monkeypatch.setenv("API_KEY", "test-key-12345")
    monkeypatch.delenv("PRODUCTION_SECRET", raising=False)

Thực chiến

Scenario: Testing hệ thống microservices đặt hàng

Giả sử hệ thống có 3 services: Order Service (nhận đơn), Inventory Service (kiểm kho), Payment Service (thanh toán). Mỗi service giao tiếp qua HTTP API. Thách thức: test sao cho phát hiện lỗi ở ranh giới service mà không cần chạy toàn bộ hệ thống.

Tầng 1 — Unit test logic nghiệp vụ

python

# order_service/domain/validators.py
from dataclasses import dataclass
from decimal import Decimal


@dataclass
class OrderValidationResult:
    is_valid: bool
    errors: list[str]


def validate_order_request(
    customer_id: str,
    items: list[dict],
    payment_method: str,
) -> OrderValidationResult:
    """Validate đơn hàng — pure function, không dependency ngoài."""
    errors = []

    if not customer_id or not customer_id.startswith("CUST-"):
        errors.append("customer_id phải bắt đầu bằng 'CUST-'")

    if not items:
        errors.append("Đơn hàng phải có ít nhất một sản phẩm")

    for idx, item in enumerate(items):
        qty = item.get("quantity", 0)
        if qty <= 0:
            errors.append(f"Sản phẩm #{idx}: quantity phải > 0")

        price = Decimal(str(item.get("unit_price", "0")))
        if price <= 0:
            errors.append(f"Sản phẩm #{idx}: unit_price phải > 0")

    valid_methods = {"credit_card", "bank_transfer", "e_wallet"}
    if payment_method not in valid_methods:
        errors.append(f"payment_method không hợp lệ: {payment_method}")

    return OrderValidationResult(
        is_valid=len(errors) == 0,
        errors=errors,
    )


# Unit test — nhanh, cô lập, xác định rõ lỗi
class TestValidateOrderRequest:
    def test_valid_order(self):
        result = validate_order_request(
            customer_id="CUST-001",
            items=[{"product_id": "P1", "quantity": 2, "unit_price": "50.00"}],
            payment_method="credit_card",
        )
        assert result.is_valid is True
        assert result.errors == []

    def test_invalid_customer_id(self):
        result = validate_order_request(
            customer_id="USER-001",
            items=[{"product_id": "P1", "quantity": 1, "unit_price": "10.00"}],
            payment_method="credit_card",
        )
        assert result.is_valid is False
        assert any("CUST-" in e for e in result.errors)

    def test_multiple_validation_errors(self):
        result = validate_order_request(
            customer_id="",
            items=[{"product_id": "P1", "quantity": -1, "unit_price": "0"}],
            payment_method="bitcoin",
        )
        assert result.is_valid is False
        assert len(result.errors) >= 3

Tầng 2 — Integration test với service boundaries

python

# tests/integration/test_inventory_client.py
import pytest
import respx
import httpx
from order_service.clients.inventory import InventoryClient


@pytest.fixture
def inventory_client():
    return InventoryClient(base_url="http://inventory-service:8001")


class TestInventoryClient:
    """Test HTTP client gọi Inventory Service — mock HTTP, test serialization."""

    @respx.mock
    def test_check_stock_available(self, inventory_client):
        respx.post("http://inventory-service:8001/api/stock/check").mock(
            return_value=httpx.Response(200, json={
                "available": True,
                "items": [
                    {"product_id": "SKU-001", "requested": 2, "in_stock": 50},
                ],
            })
        )

        result = inventory_client.check_stock(
            items=[{"product_id": "SKU-001", "quantity": 2}]
        )
        assert result.available is True
        assert result.items[0].in_stock == 50

    @respx.mock
    def test_check_stock_insufficient(self, inventory_client):
        respx.post("http://inventory-service:8001/api/stock/check").mock(
            return_value=httpx.Response(200, json={
                "available": False,
                "items": [
                    {"product_id": "SKU-002", "requested": 100, "in_stock": 3},
                ],
            })
        )

        result = inventory_client.check_stock(
            items=[{"product_id": "SKU-002", "quantity": 100}]
        )
        assert result.available is False

    @respx.mock
    def test_inventory_service_timeout(self, inventory_client):
        respx.post("http://inventory-service:8001/api/stock/check").mock(
            side_effect=httpx.ConnectTimeout("Connection timed out")
        )

        with pytest.raises(httpx.ConnectTimeout):
            inventory_client.check_stock(
                items=[{"product_id": "SKU-001", "quantity": 1}]
            )

Tầng 3 — Contract test giữa các services

python

# tests/contract/test_order_payment_contract.py
"""
Contract test: đảm bảo Order Service gửi đúng format
mà Payment Service mong đợi. Chạy mà KHÔNG cần Payment Service thật.
"""
import pytest
from pydantic import BaseModel, ValidationError
from decimal import Decimal


class PaymentRequest(BaseModel):
    """Schema mà Payment Service mong đợi — source of truth."""
    order_id: str
    amount: Decimal
    currency: str
    payment_method: str
    customer_id: str
    idempotency_key: str


class TestOrderToPaymentContract:
    """Verify payload Order Service tạo ra khớp schema Payment Service."""

    def test_payment_request_schema_valid(self):
        payload = {
            "order_id": "ORD-20240115-001",
            "amount": Decimal("299.99"),
            "currency": "VND",
            "payment_method": "credit_card",
            "customer_id": "CUST-001",
            "idempotency_key": "idem-abc-123",
        }
        request = PaymentRequest(**payload)
        assert request.order_id == "ORD-20240115-001"
        assert request.amount == Decimal("299.99")

    def test_missing_idempotency_key_rejected(self):
        """Payment Service yêu cầu idempotency_key — thiếu phải lỗi."""
        payload = {
            "order_id": "ORD-001",
            "amount": Decimal("100.00"),
            "currency": "VND",
            "payment_method": "credit_card",
            "customer_id": "CUST-001",
        }
        with pytest.raises(ValidationError):
            PaymentRequest(**payload)

    def test_amount_must_be_decimal(self):
        """amount phải là Decimal, không chấp nhận string tùy ý."""
        payload = {
            "order_id": "ORD-001",
            "amount": "not_a_number",
            "currency": "VND",
            "payment_method": "credit_card",
            "customer_id": "CUST-001",
            "idempotency_key": "idem-001",
        }
        with pytest.raises(ValidationError):
            PaymentRequest(**payload)

Sai lầm điển hình

Sai lầm 1: Đảo ngược kim tự tháp — quá nhiều E2E, quá ít Unit

python

# ❌ SAI — Viết E2E test cho mọi thứ, kể cả logic tính toán đơn giản
class TestPricingE2E:
    """Khởi động toàn bộ server chỉ để test phép tính."""

    def test_discount_calculation(self, api_client):
        # Chạy mất 3 giây vì phải boot server + database
        response = api_client.post("/api/orders", json={
            "customer_id": "CUST-001",
            "items": [{"product_id": "P1", "quantity": 1, "unit_price": "100"}],
            "discount_percent": 10,
        })
        assert response.json()["total"] == "90.00"

python

# ✅ ĐÚNG — Pure logic test bằng unit test, nhanh gấp 1000 lần
class TestPricingUnit:
    def test_discount_calculation(self):
        items = [OrderItem("P1", quantity=1, unit_price=Decimal("100"))]
        result = calculate_total(items, discount_percent=Decimal("10"))
        assert result == Decimal("90.00")
        # Chạy trong < 1ms, không cần server hay database

Sai lầm 2: Over-mocking — Mock quá nhiều, test mất ý nghĩa

python

# ❌ SAI — Mock hết mọi thứ, test chẳng kiểm tra được gì thật
def test_place_order_over_mocked(mocker):
    mocker.patch("service.order_service.calculate_total", return_value=Decimal("100"))
    mocker.patch("service.order_service.OrderRepository")
    mocker.patch("service.order_service.EmailSender")

    # Test này pass ngay cả khi calculate_total có bug
    # vì ta đã mock nó trả về giá trị cố định!
    service = OrderService(mocker.MagicMock(), mocker.MagicMock())
    order_id = service.place_order("CUST-1", "a@b.com", [])
    assert order_id is not None  # Assertion vô nghĩa

python

# ✅ ĐÚNG — Chỉ mock external dependencies, giữ business logic thật
def test_place_order_correct_mocking(mocker):
    mock_repo = mocker.MagicMock()
    mock_repo.create_order.return_value = "ORD-001"
    mock_email = mocker.MagicMock()

    # calculate_total chạy THẬT — phát hiện bug nếu có
    service = OrderService(repository=mock_repo, email_sender=mock_email)
    items = [OrderItem("P1", quantity=2, unit_price=Decimal("50.00"))]

    order_id = service.place_order("CUST-1", "a@b.com", items)
    assert order_id == "ORD-001"
    mock_email.send.assert_called_once()

Sai lầm 3: Test implementation details thay vì behavior

python

# ❌ SAI — Test gắn chặt vào cách triển khai nội bộ
def test_user_service_internal_cache(mocker):
    service = UserService()
    # Test kiểm tra biến private — sẽ fail khi refactor
    assert service._cache == {}
    service.get_user("USR-001")
    assert "USR-001" in service._cache
    assert service._cache["USR-001"]["fetched_at"] is not None

python

# ✅ ĐÚNG — Test hành vi quan sát được từ bên ngoài
def test_user_service_caches_result(mocker):
    mock_db = mocker.MagicMock()
    mock_db.find_user.return_value = {"id": "USR-001", "name": "Minh"}

    service = UserService(database=mock_db)
    result_1 = service.get_user("USR-001")
    result_2 = service.get_user("USR-001")

    # Kiểm tra behavior: gọi 2 lần nhưng database chỉ query 1 lần
    assert result_1 == result_2
    assert mock_db.find_user.call_count == 1

Sai lầm 4: Flaky tests — Tests lúc pass lúc fail

python

# ❌ SAI — Phụ thuộc thời gian thực, fail vào nửa đêm
def test_welcome_message_flaky():
    message = generate_welcome("Minh")
    assert message == "Chào buổi sáng, Minh!"  # Fail sau 12h trưa!

python

# ✅ ĐÚNG — Inject dependency thời gian, kiểm soát hoàn toàn
from datetime import datetime


def test_welcome_message_deterministic():
    morning = datetime(2024, 1, 15, 8, 0)  # 8:00 AM cố định
    message = generate_welcome("Minh", current_time=morning)
    assert message == "Chào buổi sáng, Minh!"

    evening = datetime(2024, 1, 15, 20, 0)  # 8:00 PM cố định
    message = generate_welcome("Minh", current_time=evening)
    assert message == "Chào buổi tối, Minh!"

Sai lầm 5: Shared mutable state giữa các tests

python

# ❌ SAI — Tests chia sẻ state, thứ tự chạy ảnh hưởng kết quả
_shared_users = []

def test_add_user():
    _shared_users.append({"id": "U1", "name": "An"})
    assert len(_shared_users) == 1

def test_list_users():
    # Giả định _shared_users rỗng — FAIL vì test trước đã thêm data
    assert len(_shared_users) == 0  # AssertionError!

python

# ✅ ĐÚNG — Mỗi test có state riêng qua fixture
@pytest.fixture
def user_store():
    """Factory mới cho mỗi test — hoàn toàn cô lập."""
    return []

def test_add_user(user_store):
    user_store.append({"id": "U1", "name": "An"})
    assert len(user_store) == 1

def test_list_users(user_store):
    assert len(user_store) == 0  # Luôn pass — state riêng biệt

Under the Hood

Cơ chế thực thi và isolation của pytest

Khi chạy pytest, quá trình xử lý diễn ra theo các bước:

Collection phase: pytest quét toàn bộ file test_*.py, thu thập tất cả hàm test_* và class Test*
Fixture resolution: Xây dựng dependency graph giữa fixtures, xác định scope (function, class, module, session)
Execution phase: Chạy từng test trong process riêng biệt (hoặc song song với pytest-xdist)
Teardown phase: Gọi finalizer/cleanup theo thứ tự ngược với setup

python

# Minh họa fixture scope và lifecycle
import pytest

@pytest.fixture(scope="session")
def expensive_resource():
    """Chỉ tạo MỘT LẦN cho toàn bộ test session."""
    print("  [SESSION] Setup expensive resource")
    resource = {"connection_pool": "initialized"}
    yield resource
    print("  [SESSION] Teardown expensive resource")

@pytest.fixture(scope="module")
def module_state(expensive_resource):
    """Tạo mới cho mỗi MODULE (file test)."""
    print("    [MODULE] Setup module state")
    yield {"module_data": True, "pool": expensive_resource}
    print("    [MODULE] Teardown module state")

@pytest.fixture  # scope="function" mặc định
def clean_state(module_state):
    """Tạo mới cho MỖI TEST FUNCTION."""
    print("      [FUNCTION] Setup clean state")
    yield {"test_data": [], "module": module_state}
    print("      [FUNCTION] Teardown clean state")

Hiệu năng từng tầng test

Bảng so sánh đặc điểm hiệu năng dựa trên dự án thực tế (order management system, ~50 endpoints):

Đặc điểm	Unit Test	Integration Test	E2E Test
Thời gian trung bình/test	1-10 ms	100-500 ms	2-30 giây
Số test chạy song song	Hàng trăm	Hàng chục	3-5
Dependency ngoài	Không	Database, cache	Toàn bộ stack
Nguyên nhân fail phổ biến	Bug logic	Schema mismatch	Timeout, race condition
Tần suất chạy	Mỗi save (watch mode)	Mỗi commit	Mỗi PR hoặc nightly
Chi phí maintain	Thấp	Trung bình	Cao

Thiết kế CI/CD Pipeline cho test suites

Pipeline tối ưu chạy test theo tầng — fail fast ở tầng nhanh, tránh lãng phí thời gian cho tầng chậm:

yaml

# .github/workflows/test-pipeline.yml
name: Test Pipeline

on: [push, pull_request]

jobs:
  # Stage 1: Chạy đầu tiên — nhanh nhất (< 1 phút)
  static-analysis:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install ruff mypy
      - run: ruff check .
      - run: mypy --strict src/

  # Stage 2: Chạy sau static analysis pass (< 3 phút)
  unit-tests:
    needs: static-analysis
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install -e ".[test]"
      - run: pytest tests/unit/ -x --tb=short -q
      # -x: dừng ngay test đầu tiên fail

  # Stage 3: Chạy sau unit tests pass (< 10 phút)
  integration-tests:
    needs: unit-tests
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_DB: testdb
          POSTGRES_PASSWORD: testpass
        ports:
          - 5432:5432
    steps:
      - uses: actions/checkout@v4
      - run: pip install -e ".[test]"
      - run: pytest tests/integration/ --tb=short
        env:
          DATABASE_URL: postgresql://postgres:testpass@localhost:5432/testdb

  # Stage 4: Chạy cuối cùng, chỉ trên main branch hoặc PR
  e2e-tests:
    needs: integration-tests
    if: github.event_name == 'pull_request' || github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: docker compose -f docker-compose.test.yml up -d
      - run: pip install -e ".[test]"
      - run: pytest tests/e2e/ --timeout=60 --tb=long
      - run: docker compose -f docker-compose.test.yml down

Bảng trade-offs giữa các chiến lược testing

Chiến lược	Ưu điểm	Nhược điểm	Phù hợp khi
Pyramid (70/20/10)	Feedback nhanh, chi phí thấp	Bỏ sót lỗi tích hợp	Monolith, logic nghiệp vụ phức tạp
Trophy (nhiều integration)	Phát hiện lỗi thực tế	Chậm hơn, setup phức tạp	Frontend, API-centric apps
Honeycomb (nhiều contract)	Tin cậy ranh giới service	Cần maintain schema contracts	Microservices
Ice cream cone (nhiều E2E)	Dễ viết ban đầu	Chậm, flaky, tốn CI resources	Tránh — đây là anti-pattern

Checklist ghi nhớ

✅ Checklist triển khai

[ ] Phân bổ test theo kim tự tháp: ~70% unit, ~20% integration, ~10% e2e
[ ] Unit test không có dependency ngoài — không database, không network, không filesystem
[ ] Mock ở ranh giới hệ thống (I/O, API calls), giữ business logic chạy thật
[ ] Mỗi test phải independent — không phụ thuộc thứ tự chạy hoặc state từ test khác
[ ] Sử dụng fixture với scope phù hợp: function cho isolation, session cho resource đắt
[ ] Transaction rollback pattern cho database integration tests
[ ] Test behavior (output, side effects), không test implementation details
[ ] Inject dependencies thay vì hardcode — đặc biệt thời gian, random, I/O
[ ] Contract test cho ranh giới giữa microservices
[ ] CI pipeline chạy test theo tầng: static → unit → integration → e2e
[ ] Flaky test phải được fix hoặc quarantine ngay — không bao giờ ignore
[ ] Tên test mô tả hành vi: test_expired_coupon_raises_validation_error
[ ] Review test code nghiêm túc như production code

Bài tập luyện tập

Bài 1 — Thiết kế test suite cho UserService (Intermediate)

Cho UserService sau, hãy viết đầy đủ unit tests và integration tests:

python

class UserService:
    def __init__(self, db_repository, cache, email_sender):
        self._db = db_repository
        self._cache = cache
        self._email = email_sender

    def register(self, email: str, name: str) -> str:
        if not email or "@" not in email:
            raise ValueError("Email không hợp lệ")
        if self._db.find_by_email(email):
            raise ValueError("Email đã tồn tại")

        user_id = self._db.create_user(email=email, name=name)
        self._cache.invalidate("user_list")
        self._email.send(
            to=email,
            subject="Chào mừng bạn!",
            body=f"Xin chào {name}, tài khoản của bạn đã được tạo.",
        )
        return user_id

    def get_user(self, user_id: str) -> dict | None:
        cached = self._cache.get(f"user:{user_id}")
        if cached:
            return cached
        user = self._db.find_by_id(user_id)
        if user:
            self._cache.set(f"user:{user_id}", user, ttl=300)
        return user

Yêu cầu:

Viết ít nhất 5 unit tests cho register() và get_user()
Mock đúng dependencies (db, cache, email)
Test cả happy path lẫn error cases
Đảm bảo không test implementation details

🧠 Quiz

Câu hỏi: Trong register(), nên mock ValueError check ("@" not in email) không?

[ ] A. Có — mock hết để test nhanh
[x] B. Không — đây là pure logic, để chạy thật
[ ] C. Tùy — nếu logic phức tạp thì mock
[ ] D. Không cần test validation

Giải thích: Validation logic là pure function (không I/O, không side effect). Không bao giờ mock pure logic — để nó chạy thật trong unit test. Chỉ mock external dependencies như database, cache, email service.

✅ Lời giải Bài 1

python

import pytest
from unittest.mock import MagicMock


class TestUserServiceRegister:
    """Unit tests cho register() — mock external dependencies."""

    @pytest.fixture
    def deps(self):
        return {
            "db": MagicMock(),
            "cache": MagicMock(),
            "email": MagicMock(),
        }

    @pytest.fixture
    def service(self, deps):
        return UserService(
            db_repository=deps["db"],
            cache=deps["cache"],
            email_sender=deps["email"],
        )

    def test_register_success(self, service, deps):
        deps["db"].find_by_email.return_value = None
        deps["db"].create_user.return_value = "USR-001"

        user_id = service.register("minh@example.com", "Minh")

        assert user_id == "USR-001"
        deps["db"].create_user.assert_called_once_with(
            email="minh@example.com", name="Minh",
        )
        deps["cache"].invalidate.assert_called_once_with("user_list")
        deps["email"].send.assert_called_once()

    def test_register_invalid_email_raises(self, service):
        with pytest.raises(ValueError, match="không hợp lệ"):
            service.register("invalid-email", "Minh")

    def test_register_empty_email_raises(self, service):
        with pytest.raises(ValueError, match="không hợp lệ"):
            service.register("", "Minh")

    def test_register_duplicate_email_raises(self, service, deps):
        deps["db"].find_by_email.return_value = {"id": "USR-EXIST"}

        with pytest.raises(ValueError, match="đã tồn tại"):
            service.register("taken@example.com", "Minh")

        deps["db"].create_user.assert_not_called()
        deps["email"].send.assert_not_called()


class TestUserServiceGetUser:
    """Unit tests cho get_user() — kiểm tra cache behavior."""

    @pytest.fixture
    def deps(self):
        return {
            "db": MagicMock(),
            "cache": MagicMock(),
            "email": MagicMock(),
        }

    @pytest.fixture
    def service(self, deps):
        return UserService(
            db_repository=deps["db"],
            cache=deps["cache"],
            email_sender=deps["email"],
        )

    def test_get_user_cache_hit(self, service, deps):
        deps["cache"].get.return_value = {"id": "USR-001", "name": "Minh"}

        result = service.get_user("USR-001")

        assert result["name"] == "Minh"
        deps["db"].find_by_id.assert_not_called()

    def test_get_user_cache_miss_fetches_from_db(self, service, deps):
        deps["cache"].get.return_value = None
        deps["db"].find_by_id.return_value = {"id": "USR-002", "name": "An"}

        result = service.get_user("USR-002")

        assert result["name"] == "An"
        deps["db"].find_by_id.assert_called_once_with("USR-002")
        deps["cache"].set.assert_called_once_with(
            "user:USR-002", {"id": "USR-002", "name": "An"}, ttl=300,
        )

    def test_get_user_not_found(self, service, deps):
        deps["cache"].get.return_value = None
        deps["db"].find_by_id.return_value = None

        result = service.get_user("USR-NONE")

        assert result is None
        deps["cache"].set.assert_not_called()

Bài 2 — Contract test cho Payment API (Advanced)

Hai team phát triển độc lập: Team A (Order Service) gửi payment request, Team B (Payment Service) xử lý. Viết contract test đảm bảo hai bên đồng bộ schema.

Payment Service mong đợi:

python

{
    "order_id": str,       # format: "ORD-YYYYMMDD-NNN"
    "amount_cents": int,   # số nguyên, đơn vị xu (VNĐ không có xu nhưng giả lập)
    "currency": str,       # ISO 4217: "VND"
    "method": str,         # "credit_card" | "bank_transfer" | "e_wallet"
    "metadata": {
        "customer_id": str,
        "idempotency_key": str,  # UUID v4
    }
}

Yêu cầu:

Viết Pydantic model cho schema trên
Viết ≥ 3 contract tests: valid payload, thiếu field bắt buộc, sai format order_id
Thêm custom validator cho order_id format

🧠 Quiz

Câu hỏi: Contract test khác gì integration test?

[ ] A. Không khác — chỉ là tên gọi khác
[x] B. Contract test chạy không cần service thật, chỉ validate schema
[ ] C. Contract test chậm hơn vì phải gọi API thật
[ ] D. Contract test chỉ dùng cho frontend

Giải thích: Contract test validate rằng hai bên (producer và consumer) đồng ý về cùng một schema/format. Không cần chạy service thật — chỉ cần schema definition. Điều này cho phép hai team phát triển độc lập mà vẫn đảm bảo tương thích khi tích hợp.

✅ Lời giải Bài 2

python

import re
import uuid
import pytest
from pydantic import BaseModel, field_validator, ValidationError


class PaymentMetadata(BaseModel):
    customer_id: str
    idempotency_key: str

    @field_validator("idempotency_key")
    @classmethod
    def validate_uuid(cls, v: str) -> str:
        try:
            uuid.UUID(v, version=4)
        except ValueError:
            raise ValueError("idempotency_key phải là UUID v4 hợp lệ")
        return v


class PaymentRequest(BaseModel):
    order_id: str
    amount_cents: int
    currency: str
    method: str
    metadata: PaymentMetadata

    @field_validator("order_id")
    @classmethod
    def validate_order_id_format(cls, v: str) -> str:
        pattern = r"^ORD-\d{8}-\d{3,}$"
        if not re.match(pattern, v):
            raise ValueError(
                f"order_id phải có format 'ORD-YYYYMMDD-NNN', nhận: {v}"
            )
        return v

    @field_validator("method")
    @classmethod
    def validate_method(cls, v: str) -> str:
        allowed = {"credit_card", "bank_transfer", "e_wallet"}
        if v not in allowed:
            raise ValueError(f"method phải là một trong {allowed}")
        return v


class TestPaymentContract:
    """Contract tests — đảm bảo Order Service tạo payload đúng schema."""

    def test_valid_payment_request(self):
        payload = {
            "order_id": "ORD-20240115-001",
            "amount_cents": 29999,
            "currency": "VND",
            "method": "credit_card",
            "metadata": {
                "customer_id": "CUST-001",
                "idempotency_key": str(uuid.uuid4()),
            },
        }
        request = PaymentRequest(**payload)
        assert request.amount_cents == 29999
        assert request.currency == "VND"

    def test_missing_metadata_rejected(self):
        payload = {
            "order_id": "ORD-20240115-001",
            "amount_cents": 10000,
            "currency": "VND",
            "method": "bank_transfer",
        }
        with pytest.raises(ValidationError) as exc_info:
            PaymentRequest(**payload)
        assert "metadata" in str(exc_info.value)

    def test_invalid_order_id_format_rejected(self):
        payload = {
            "order_id": "ORDER-001",
            "amount_cents": 5000,
            "currency": "VND",
            "method": "e_wallet",
            "metadata": {
                "customer_id": "CUST-002",
                "idempotency_key": str(uuid.uuid4()),
            },
        }
        with pytest.raises(ValidationError) as exc_info:
            PaymentRequest(**payload)
        assert "ORD-YYYYMMDD-NNN" in str(exc_info.value)

    def test_invalid_payment_method_rejected(self):
        payload = {
            "order_id": "ORD-20240115-002",
            "amount_cents": 15000,
            "currency": "VND",
            "method": "bitcoin",
            "metadata": {
                "customer_id": "CUST-003",
                "idempotency_key": str(uuid.uuid4()),
            },
        }
        with pytest.raises(ValidationError):
            PaymentRequest(**payload)

    def test_invalid_idempotency_key_rejected(self):
        payload = {
            "order_id": "ORD-20240115-003",
            "amount_cents": 8000,
            "currency": "VND",
            "method": "credit_card",
            "metadata": {
                "customer_id": "CUST-004",
                "idempotency_key": "not-a-uuid",
            },
        }
        with pytest.raises(ValidationError) as exc_info:
            PaymentRequest(**payload)
        assert "UUID v4" in str(exc_info.value)

Kiến trúc Testing — Từ Unit đến End-to-End ​

Bức tranh tư duy ​

Testing Pyramid — Kim tự tháp kiểm thử ​

Khi nào kim tự tháp không còn phù hợp? ​

Cốt lõi kỹ thuật ​

Unit Tests — Kiểm tra đơn vị ​

Integration Tests — Kiểm tra tích hợp ​

E2E Tests — Kiểm tra toàn diện ​

Test Doubles — Mock, Stub, Fake, Spy ​

Mocking Strategies — unittest.mock và pytest-mock ​

Test Isolation Patterns ​

Thực chiến ​

Scenario: Testing hệ thống microservices đặt hàng ​

Tầng 1 — Unit test logic nghiệp vụ ​

Tầng 2 — Integration test với service boundaries ​

Tầng 3 — Contract test giữa các services ​

Sai lầm điển hình ​

Sai lầm 1: Đảo ngược kim tự tháp — quá nhiều E2E, quá ít Unit ​

Sai lầm 2: Over-mocking — Mock quá nhiều, test mất ý nghĩa ​

Sai lầm 3: Test implementation details thay vì behavior ​

Sai lầm 4: Flaky tests — Tests lúc pass lúc fail ​

Sai lầm 5: Shared mutable state giữa các tests ​

Under the Hood ​

Cơ chế thực thi và isolation của pytest ​

Hiệu năng từng tầng test ​

Thiết kế CI/CD Pipeline cho test suites ​

Bảng trade-offs giữa các chiến lược testing ​

Checklist ghi nhớ ​

Bài tập luyện tập ​

Bài 1 — Thiết kế test suite cho UserService (Intermediate) ​

Bài 2 — Contract test cho Payment API (Advanced) ​

Liên kết học tiếp ​

Kiến trúc Testing — Từ Unit đến End-to-End

Bức tranh tư duy

Testing Pyramid — Kim tự tháp kiểm thử

Khi nào kim tự tháp không còn phù hợp?

Cốt lõi kỹ thuật

Unit Tests — Kiểm tra đơn vị

Integration Tests — Kiểm tra tích hợp

E2E Tests — Kiểm tra toàn diện

Test Doubles — Mock, Stub, Fake, Spy

Mocking Strategies — unittest.mock và pytest-mock

Test Isolation Patterns

Thực chiến

Scenario: Testing hệ thống microservices đặt hàng

Tầng 1 — Unit test logic nghiệp vụ

Tầng 2 — Integration test với service boundaries

Tầng 3 — Contract test giữa các services

Sai lầm điển hình

Sai lầm 1: Đảo ngược kim tự tháp — quá nhiều E2E, quá ít Unit

Sai lầm 2: Over-mocking — Mock quá nhiều, test mất ý nghĩa

Sai lầm 3: Test implementation details thay vì behavior

Sai lầm 4: Flaky tests — Tests lúc pass lúc fail

Sai lầm 5: Shared mutable state giữa các tests

Under the Hood

Cơ chế thực thi và isolation của pytest

Hiệu năng từng tầng test

Thiết kế CI/CD Pipeline cho test suites

Bảng trade-offs giữa các chiến lược testing

Checklist ghi nhớ

Bài tập luyện tập

Bài 1 — Thiết kế test suite cho UserService (Intermediate)

Bài 2 — Contract test cho Payment API (Advanced)

Liên kết học tiếp