Test Architecture Nâng cao

Test Architecture = Chiến lược testing đúng đắn = Confidence cao, maintenance thấp

Learning Outcomes

Sau khi hoàn thành trang này, bạn sẽ:

• ✅ Phân biệt Unit, Integration, E2E tests và biết khi nào dùng loại nào
• ✅ Hiểu và áp dụng Test Doubles (mocks, stubs, fakes, spies)
• ✅ Thiết kế test isolation strategies hiệu quả
• ✅ Tích hợp testing vào CI/CD pipeline
• ✅ Áp dụng Testing Pyramid trong thực tế

---

Testing Pyramid

Testing Pyramid là mô hình kinh điển để cân bằng các loại tests:

        /\
       /  \
      / E2E \        ← Ít tests, chậm, expensive
     /--------\
    /Integration\    ← Vừa phải
   /--------------\
  /     Unit       \  ← Nhiều tests, nhanh, cheap
 /------------------\

Level	Quantity	Speed	Cost	Scope
Unit	Many (70%)	Fast (ms)	Low	Single function/class
Integration	Medium (20%)	Medium (s)	Medium	Multiple components
E2E	Few (10%)	Slow (min)	High	Entire system

Tại sao Pyramid?

---

Unit Tests

Unit tests kiểm tra một đơn vị code (function, method, class) trong isolation.

Characteristics

• ✅ Fast: Chạy trong milliseconds
• ✅ Isolated: Không phụ thuộc external systems
• ✅ Deterministic: Luôn cho cùng kết quả
• ✅ Focused: Test một behavior cụ thể

Example

# src/calculator.py
class Calculator:
    def add(self, a: float, b: float) -> float:
        return a + b
    
    def divide(self, a: float, b: float) -> float:
        if b == 0:
            raise ValueError("Cannot divide by zero")
        return a / b

# tests/unit/test_calculator.py
import pytest
from src.calculator import Calculator

class TestCalculator:
    @pytest.fixture
    def calc(self):
        return Calculator()
    
    def test_add_positive_numbers(self, calc):
        assert calc.add(2, 3) == 5
    
    def test_add_negative_numbers(self, calc):
        assert calc.add(-2, -3) == -5
    
    def test_divide_normal(self, calc):
        assert calc.divide(10, 2) == 5
    
    def test_divide_by_zero_raises(self, calc):
        with pytest.raises(ValueError, match="Cannot divide by zero"):
            calc.divide(10, 0)

What to Unit Test

# ✅ GOOD: Pure functions
def calculate_discount(price: float, discount_percent: float) -> float:
    """Pure function - easy to unit test."""
    return price * (1 - discount_percent / 100)

def test_calculate_discount():
    assert calculate_discount(100, 20) == 80
    assert calculate_discount(50, 10) == 45

# ✅ GOOD: Business logic
class OrderValidator:
    def validate(self, order: Order) -> list[str]:
        errors = []
        if order.total <= 0:
            errors.append("Total must be positive")
        if not order.items:
            errors.append("Order must have items")
        return errors

def test_order_validator_empty_items():
    validator = OrderValidator()
    order = Order(total=100, items=[])
    errors = validator.validate(order)
    assert "Order must have items" in errors

# ❌ AVOID: Testing implementation details
def test_internal_cache_structure():
    # Don't test private implementation
    service = MyService()
    assert service._cache == {}  # ❌ Testing private attribute

---

Integration Tests

Integration tests kiểm tra interaction giữa multiple components.

Characteristics

• ✅ Real dependencies: Database, APIs, file system
• ✅ Component interaction: Multiple classes working together
• ✅ Slower than unit tests: Seconds instead of milliseconds
• ✅ More realistic: Closer to production behavior

Example: Database Integration

# tests/integration/test_user_repository.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from src.models import Base, User
from src.repositories import UserRepository

@pytest.fixture(scope="module")
def engine():
    """Create test database engine."""
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    yield engine
    engine.dispose()

@pytest.fixture
def session(engine):
    """Create fresh session for each test."""
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.rollback()
    session.close()

@pytest.fixture
def user_repo(session):
    return UserRepository(session)

class TestUserRepository:
    def test_create_user(self, user_repo, session):
        user = user_repo.create(name="Alice", email="alice@example.com")
        
        assert user.id is not None
        assert user.name == "Alice"
        
        # Verify in database
        db_user = session.query(User).filter_by(id=user.id).first()
        assert db_user is not None
        assert db_user.email == "alice@example.com"
    
    def test_find_by_email(self, user_repo):
        user_repo.create(name="Bob", email="bob@example.com")
        
        found = user_repo.find_by_email("bob@example.com")
        
        assert found is not None
        assert found.name == "Bob"
    
    def test_find_by_email_not_found(self, user_repo):
        found = user_repo.find_by_email("nonexistent@example.com")
        assert found is None
)

Example: API Integration

# tests/integration/test_api.py
import pytest
from fastapi.testclient import TestClient
from src.main import app
from src.database import get_db, Base, engine

@pytest.fixture(scope="module")
def client():
    """Create test client with test database."""
    Base.metadata.create_all(bind=engine)
    with TestClient(app) as client:
        yield client
    Base.metadata.drop_all(bind=engine)

class TestUserAPI:
    def test_create_user(self, client):
        response = client.post("/users/", json={
            "name": "Alice",
            "email": "alice@example.com"
        })
        
        assert response.status_code == 201
        data = response.json()
        assert data["name"] == "Alice"
        assert "id" in data
    
    def test_get_user(self, client):
        # Create user first
        create_response = client.post("/users/", json={
            "name": "Bob",
            "email": "bob@example.com"
        })
        user_id = create_response.json()["id"]
        
        # Get user
        response = client.get(f"/users/{user_id}")
        
        assert response.status_code == 200
        assert response.json()["name"] == "Bob"
    
    def test_get_nonexistent_user(self, client):
        response = client.get("/users/99999")
        assert response.status_code == 404

---

End-to-End (E2E) Tests

E2E tests kiểm tra entire user journey từ đầu đến cuối.

Characteristics

• ✅ Full stack: Frontend → Backend → Database
• ✅ User perspective: Test như user thực sự sử dụng
• ✅ Slowest: Minutes to run
• ✅ Most realistic: Catches integration issues

Example: Playwright E2E

# tests/e2e/test_user_flow.py
import pytest
from playwright.sync_api import Page, expect

@pytest.fixture(scope="session")
def browser_context(browser):
    """Create browser context for tests."""
    context = browser.new_context()
    yield context
    context.close()

@pytest.fixture
def page(browser_context):
    """Create new page for each test."""
    page = browser_context.new_page()
    yield page
    page.close()

class TestUserRegistrationFlow:
    def test_user_can_register_and_login(self, page: Page):
        # Navigate to registration
        page.goto("http://localhost:3000/register")
        
        # Fill registration form
        page.fill("[data-testid=name-input]", "Alice")
        page.fill("[data-testid=email-input]", "alice@example.com")
        page.fill("[data-testid=password-input]", "SecurePass123!")
        page.click("[data-testid=register-button]")
        
        # Verify redirect to login
        expect(page).to_have_url("http://localhost:3000/login")
        
        # Login with new account
        page.fill("[data-testid=email-input]", "alice@example.com")
        page.fill("[data-testid=password-input]", "SecurePass123!")
        page.click("[data-testid=login-button]")
        
        # Verify logged in
        expect(page).to_have_url("http://localhost:3000/dashboard")
        expect(page.locator("[data-testid=welcome-message]")).to_contain_text("Alice")
    
    def test_user_can_create_order(self, page: Page, logged_in_user):
        page.goto("http://localhost:3000/products")
        
        # Add product to cart
        page.click("[data-testid=product-1] [data-testid=add-to-cart]")
        
        # Go to checkout
        page.click("[data-testid=cart-icon]")
        page.click("[data-testid=checkout-button]")
        
        # Complete order
        page.fill("[data-testid=address-input]", "123 Main St")
        page.click("[data-testid=place-order-button]")
        
        # Verify order confirmation
        expect(page.locator("[data-testid=order-confirmation]")).to_be_visible()

---

Test Doubles

Test Doubles là objects thay thế real dependencies trong tests:

1. Dummy

Dummy: Placeholder object, không được sử dụng thực sự.

class DummyLogger:
    """Dummy logger - methods do nothing."""
    def info(self, msg): pass
    def error(self, msg): pass
    def debug(self, msg): pass

def test_service_with_dummy_logger():
    # Logger is required but not used in this test
    dummy_logger = DummyLogger()
    service = MyService(logger=dummy_logger)
    
    result = service.calculate(10, 20)
    assert result == 30

2. Stub

Stub: Returns pre-configured values.

class StubUserRepository:
    """Stub that returns fixed data."""
    
    def get_user(self, user_id: int) -> dict:
        return {"id": user_id, "name": "Stub User", "email": "stub@example.com"}
    
    def get_all_users(self) -> list[dict]:
        return [
            {"id": 1, "name": "User 1"},
            {"id": 2, "name": "User 2"},
        ]

def test_user_service_with_stub():
    stub_repo = StubUserRepository()
    service = UserService(repository=stub_repo)
    
    user = service.get_user(1)
    assert user["name"] == "Stub User"

3. Spy

Spy: Records calls for later verification.

class SpyEmailService:
    """Spy that records all calls."""
    
    def __init__(self):
        self.sent_emails = []
    
    def send_email(self, to: str, subject: str, body: str):
        self.sent_emails.append({
            "to": to,
            "subject": subject,
            "body": body
        })

def test_registration_sends_welcome_email():
    spy_email = SpyEmailService()
    service = RegistrationService(email_service=spy_email)
    
    service.register("alice@example.com", "password123")
    
    # Verify email was sent
    assert len(spy_email.sent_emails) == 1
    assert spy_email.sent_emails[0]["to"] == "alice@example.com"
    assert "Welcome" in spy_email.sent_emails[0]["subject"]

4. Mock

Mock: Spy + behavior verification.

from unittest.mock import Mock, call

def test_order_processing_with_mock():
    mock_payment = Mock()
    mock_payment.charge.return_value = {"status": "success", "transaction_id": "123"}
    
    mock_inventory = Mock()
    mock_inventory.reserve.return_value = True
    
    service = OrderService(payment=mock_payment, inventory=mock_inventory)
    
    order = service.process_order(user_id=1, items=[{"id": 1, "qty": 2}])
    
    # Verify interactions
    mock_inventory.reserve.assert_called_once_with(items=[{"id": 1, "qty": 2}])
    mock_payment.charge.assert_called_once()
    
    # Verify call order
    assert mock_inventory.reserve.call_count == 1
    assert mock_payment.charge.call_count == 1

5. Fake

Fake: Working implementation for testing.

class FakeDatabase:
    """In-memory database for testing."""
    
    def __init__(self):
        self.users = {}
        self._counter = 0
    
    def create_user(self, name: str, email: str) -> dict:
        self._counter += 1
        user = {"id": self._counter, "name": name, "email": email}
        self.users[self._counter] = user
        return user
    
    def get_user(self, user_id: int) -> dict | None:
        return self.users.get(user_id)
    
    def delete_user(self, user_id: int) -> bool:
        if user_id in self.users:
            del self.users[user_id]
            return True
        return False

@pytest.fixture
def fake_db():
    return FakeDatabase()

def test_user_crud_with_fake_db(fake_db):
    # Create
    user = fake_db.create_user("Alice", "alice@example.com")
    assert user["id"] == 1
    
    # Read
    found = fake_db.get_user(1)
    assert found["name"] == "Alice"
    
    # Delete
    assert fake_db.delete_user(1) is True
    assert fake_db.get_user(1) is None

When to Use Which?

Double	Use When
Dummy	Dependency required but not used
Stub	Need specific return values
Spy	Need to verify calls were made
Mock	Need to verify call arguments and order
Fake	Need working implementation without real dependency

---

Test Isolation Strategies

1. Database Isolation

import pytest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker

@pytest.fixture(scope="session") def engine(): """Create engine once per test session.""" return create_engine("postgresql://test:test@localhost/testdb")

@pytest.fixture(scope="function") def session(engine): """Create transaction for each test, rollback after.""" connection = engine.connect() transaction = connection.begin() Session = sessionmaker(bind=connection) session = Session()

yield session

session.close()
transaction.rollback()
connection.close()

Alternative: Truncate tables

@pytest.fixture(scope="function") def clean_db(session): yield session # Truncate all tables after test for table in reversed(Base.metadata.sorted_tables): session.execute(table.delete()) session.commit() ) session.commit()

### 2. External Service Isolation
import pytest
import responses
import httpx

@pytest.fixture
def mock_external_api():
    """Mock external API calls."""
    with responses.RequestsMock() as rsps:
        rsps.add(
            responses.GET,
            "https://api.external.com/users/1",
            json={"id": 1, "name": "External User"},
            status=200
        )
        yield rsps

def test_fetch_external_user(mock_external_api):
    response = httpx.get("https://api.external.com/users/1")
    assert response.json()["name"] == "External User"
)
    response = httpx.get("https://api.external.com/users/1")
    assert response.json()["name"] == "External User"

3. Time Isolation

import pytest
from datetime import datetime
from freezegun import freeze_time

@freeze_time("2024-01-15 10:00:00")
def test_with_frozen_time():
    assert datetime.now() == datetime(2024, 1, 15, 10, 0, 0)

@pytest.fixture
def fixed_time(mocker):
    """Fixture for controlled time."""
    mock_datetime = mocker.patch("myapp.services.datetime")
    mock_datetime.now.return_value = datetime(2024, 1, 15, 10, 0, 0)
    return mock_datetime

4. File System Isolation

import pytest
from pathlib import Path

@pytest.fixture
def temp_config(tmp_path):
    """Create temporary config file."""
    config_file = tmp_path / "config.json"
    config_file.write_text('{"debug": true}')
    return config_file

def test_read_config(temp_config):
    import json
    config = json.loads(temp_config.read_text())
    assert config["debug"] is True

@pytest.fixture
def isolated_filesystem(tmp_path, monkeypatch):
    """Change working directory to temp path."""
    monkeypatch.chdir(tmp_path)
    return tmp_path

---

CI/CD Integration

GitHub Actions

# .github/workflows/test.yml
name: Tests

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    
    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_USER: test
          POSTGRES_PASSWORD: test
          POSTGRES_DB: testdb
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      
      - name: Install dependencies
        run: |
          pip install -e ".[dev]"
      
      - name: Run unit tests
        run: |
          pytest tests/unit -v --cov=src --cov-report=xml
      
      - name: Run integration tests
        run: |
          pytest tests/integration -v
        env:
          DATABASE_URL: postgresql://test:test@localhost:5432/testdb
      
      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          files: ./coverage.xml

Test Stages

# Multi-stage testing
jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run unit tests
        run: pytest tests/unit -v --tb=short
  
  integration-tests:
    needs: unit-tests  # Run after unit tests pass
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run integration tests
        run: pytest tests/integration -v
  
  e2e-tests:
    needs: integration-tests  # Run after integration tests pass
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run E2E tests
        run: pytest tests/e2e -v

pyproject.toml Configuration

[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
python_functions = ["test_*"]
markers = [
    "unit: Unit tests",
    "integration: Integration tests",
    "e2e: End-to-end tests",
    "slow: Slow tests",
]
addopts = [
    "-ra",
    "--strict-markers",
    "--strict-config",
]

# Coverage configuration
[tool.coverage.run]
source = ["src"]
branch = true
omit = ["tests/*", "*/__init__.py"]

[tool.coverage.report]
exclude_lines = [
    "pragma: no cover",
    "def __repr__",
    "raise NotImplementedError",
    "if TYPE_CHECKING:",
]
fail_under = 80

---

Production Pitfalls ⚠️

1. Testing Pyramid Inversion

# ❌ BAD: Too many E2E tests, few unit tests
# tests/
# ├── e2e/           # 100 tests (slow, flaky)
# ├── integration/   # 20 tests
# └── unit/          # 10 tests (should be most!)

# ✅ GOOD: Proper pyramid
# tests/
# ├── e2e/           # 10 tests (critical paths only)
# ├── integration/   # 30 tests
# └── unit/          # 100 tests (fast, reliable)

2. Flaky Tests

# ❌ BAD: Test depends on timing
def test_flaky_timeout():
    start = time.time()
    result = slow_operation()
    assert time.time() - start < 1.0  # Flaky!

# ✅ GOOD: Mock time or use generous timeout
def test_stable_timeout(mocker):
    mocker.patch("time.time", side_effect=[0, 0.5])
    start = time.time()
    result = slow_operation()
    assert time.time() - start < 1.0

3. Test Pollution

# ❌ BAD: Tests share state
class TestSharedState:
    items = []  # Shared across all tests!
    
    def test_add_item(self):
        self.items.append(1)
        assert len(self.items) == 1  # Might fail!
    
    def test_add_another(self):
        self.items.append(2)
        assert len(self.items) == 1  # Fails!

# ✅ GOOD: Fresh state per test
class TestIsolatedState:
    @pytest.fixture
    def items(self):
        return []
    
    def test_add_item(self, items):
        items.append(1)
        assert len(items) == 1
    
    def test_add_another(self, items):
        items.append(2)
        assert len(items) == 1  # Fresh list!

4. Over-Mocking

# ❌ BAD: Mock everything
def test_over_mocked(mocker):
    mocker.patch("myapp.validate")
    mocker.patch("myapp.process")
    mocker.patch("myapp.save")
    mocker.patch("myapp.notify")
    
    result = myapp.handle_request(data)
    # What are we even testing?

# ✅ GOOD: Mock only external dependencies
def test_properly_mocked(mocker, fake_db):
    mocker.patch("myapp.send_email")  # External service
    
    result = myapp.handle_request(data, db=fake_db)
    
    assert fake_db.get_user(result["id"]) is not None

---

Bảng Tóm tắt

# === TEST TYPES ===
# Unit: Single function/class, fast, isolated
# Integration: Multiple components, real dependencies
# E2E: Full user journey, slowest

# === TEST DOUBLES ===
# Dummy: Placeholder, not used
# Stub: Returns fixed values
# Spy: Records calls
# Mock: Spy + verification
# Fake: Working implementation

# === ISOLATION ===
# Database: Transaction rollback
# External APIs: Mock responses
# Time: freeze_time
# Filesystem: tmp_path

# === CI/CD ===
# Run unit tests first (fast feedback)
# Run integration tests after unit pass
# Run E2E tests last (expensive)
# Fail fast on first failure

# === PYRAMID ===
# 70% Unit tests
# 20% Integration tests
# 10% E2E tests

---

Cross-links

• Prerequisites: Pytest Fundamentals, Fixtures & Mocking
• Related: Property-Based Testing - Complement to example-based tests
• Related: FastAPI Deep Dive (Phase 2) - API testing patterns
• Related: Production Deployment (Phase 2) - CI/CD integration