Bài 9: Testing — Unit, Integration, End-to-End¶

Điều kiện tiên quyết: Bạn cần nắm vững Bài 3 (Dependency Injection) và Bài 5 (Kiến trúc 3 lớp). DI chính là "chìa khóa vàng" để code có thể test được. Ghi chú: Project hiện tại chưa dựng sẵn test suite — bài này sẽ dùng các ví dụ test mẫu mô phỏng chính xác những gì đang chạy trong project để bạn dễ hình dung.

1. Tại sao lại phải nai lưng ra viết test?¶

Đừng viết test chỉ vì giang hồ đồn đó là "best practice". Đây mới là những lý do thực dụng nhất:

Lý do	Ý nghĩa thực chiến
Chống phá hoại (Regression safety)	Bạn thoải mái sửa code, thêm tính năng mà không sợ vô tình làm gãy những tính năng cũ đang chạy ngon.
Ép thiết kế tốt (Design pressure)	Code nào khó viết test = code đó đang bị dính chặt vào nhau (high coupling) = thiết kế tồi. Viết test ép bạn phải thiết kế lại cho sạch.
Tài liệu sống (Documentation)	Test là bộ tài liệu không bao giờ nói dối. Đọc test là biết ngay hàm này dùng để làm gì và kết quả mong đợi ra sao.
Tự tin đập đi xây lại (Refactor)	Dám mạnh tay dọn dẹp code rác vì đã có lưới an toàn bảo vệ.

Những đoạn code "không thể test được" thường là do nó chứa phụ thuộc ngầm (hidden dependency) bị gắn chết bên trong (ví dụ tự gọi thẳng DB, tự gọi SMTP server gửi mail, tự đọc biến môi trường). Bạn không thể test chúng nếu không setup cả một hệ thống y như thật. Khi gặp cảnh này, đó là tiếng chuông báo động yêu cầu bạn phải refactor theo nguyên lý DIP (Bài 0) ngay lập tức.

2. Kim tự tháp Testing¶

           .
          / \
         /   \
        / E2E \          ← Ít nhất (Chạy chậm, tốn kém, dễ bị lỗi vặt)
       /───────\
      /  Intg   \       ← Vừa phải (Chạm vào DB thật, mạng thật)
     /───────────\
    /    Unit     \     ← Nhiều nhất (Chạy cực nhanh, rẻ, siêu ổn định)

Loại Test	Phạm vi phủ sóng	Tốc độ	Khi nào nên xài?
Unit	1 class/function, tráo (mock) toàn bộ đồ giả vào	Mili-giây	Dùng cho Business logic, luật nghiệp vụ.
Integration	Kết nối ≥ 2 cục lại với nhau (Ví dụ: Service + DB)	Giây	Test câu query SQL, tính toàn vẹn của Transaction.
E2E (End-to-End)	Đi từ ngoài vào trong: HTTP → DB → Response	Vài giây - phút	Những luồng sinh tử: Login, Mua hàng, Upload file.

Nguyên tắc cốt lõi: Bắt bug ở tầng thấp nhất có thể. Lỗi quy tắc nghiệp vụ → Bắt bằng Unit test. Lỗi viết sai câu SQL → Bắt bằng Integration test. Lỗi rớt mạng, sai luồng Auth → Bắt bằng E2E.

Ánh xạ Kim tự tháp vào Kiến trúc 3 lớp (Bài 5)¶

Với kiến trúc Router → Service → Repository, chúng ta cũng có 3 loại test tương ứng:

Loại test	Tầng áp dụng	Mock (Tráo đồ giả) cái gì?	Mục đích kiểm tra
Unit	Service	Tráo toàn bộ Repository (Nhờ có Abstract Class)	Kiểm tra luật nghiệp vụ: validation, logic giới hạn dung lượng, thứ tự lưu DB vs File.
Integration	Repository	Không tráo gì cả — Chơi DB thật (In-memory SQLite)	Kiểm tra SQL gõ đúng không, khóa ngoại có ăn không, Transaction có commit/rollback chuẩn không.
E2E	Router (dùng `httpx.AsyncClient`)	Chỉ tráo mỗi Dependency Auth (bỏ qua bước login)	Test một lèo: Middleware + Auth + Service + DB.

Bí kíp thực dụng: Khi code tính năng mới, hãy bắt đầu viết test từ Tầng Service (Unit test). Đây là nơi chứa bộ não của nghiệp vụ — test chạy siêu nhanh, và nếu nó pass → 80% tính năng của bạn đã hoạt động đúng. Test Repository chỉ cần khi bạn gõ mấy câu SQL phức tạp (JOIN, Window function). E2E thì để dành chốt hạ các luồng quan trọng (Login, Upload).

3. Tại sao DI (Dependency Injection) lại là "vị cứu tinh"?¶

Nhớ lại Bài 3 — AudioService của chúng ta nhận AudioRepository (Bản hợp đồng ABC) thông qua constructor chứ không tự tạo ra nó:

# src/services/audio_service.py
class AudioService:
    def __init__(self, repo: AudioRepository, ...):
        self.repo = repo

Khi vào viết test, việc "tráo đồ" trở nên dễ như lật bàn tay:

# test_audio_service.py
from unittest.mock import AsyncMock, Mock

# Nặn ra một thằng đệ giả danh AudioRepository
mock_repo = Mock(spec=AudioRepository)
mock_repo.get_total_size = AsyncMock(return_value=0)

# Nhét thằng đệ giả này vào Service
svc = AudioService(mock_repo, Mock(), Mock())

→ Tuyệt vời! Test chạy cái vèo trong vài mili-giây mà không thèm đụng tới ổ cứng hay Database thật.

Hãy tưởng tượng nếu không dùng DI, hàm __init__ tự rước cái SqliteAudioRepository() vào nhà — lúc đó mỗi lần chạy test bạn lại phải lọ mọ dọn dẹp, khởi tạo lại cái DB SQLite cho nó. Cực kỳ thảm họa! Đây chính là minh chứng sống của nguyên lý DIP (Bài 0).

4. Unit Test — Bắt bệnh ở Tầng Service¶

Ví dụ: Test luật nghiệp vụ "Nếu user upload file vượt quá giới hạn ổ cứng → Quăng lỗi 507".

# tests/test_audio_service.py
import pytest
from unittest.mock import Mock, AsyncMock
from fastapi import HTTPException

from src.services.audio_service import AudioService
from src.repositories.base import AudioRepository, CategoryRepository, EventRepository

@pytest.mark.asyncio
async def test_upload_rejects_when_storage_full():
    # Bước 1: Arrange — Dàn cảnh (Chuẩn bị đồ giả)
    repo = Mock(spec=AudioRepository)
    # Giả đò báo cho Service biết là ổ cứng đã chứa 9.9GB (Sắp đầy)
    repo.get_total_size = AsyncMock(return_value=int(9.9 * 1024**3))
    repo.get_by_filename = AsyncMock(return_value=None)

    svc = AudioService(
        repo=repo,
        category_repo=Mock(spec=CategoryRepository),
        event_repo=Mock(spec=EventRepository),
    )

    # Nặn ra một cái file âm thanh giả bự chà bá (~200MB)
    file = Mock()
    file.filename = "test.mp3"
    file.read = AsyncMock(return_value=b"x" * (200 * 1024**2))

    # Bước 2 & 3: Act + Assert — Bóp cò và Kiểm tra kết quả
    with pytest.raises(HTTPException) as exc_info:
        await svc.upload(
            session=Mock(),
            file=file,
            title="Test audio",
            category_id=None,
            event_id=None,
        )

    # Khẳng định: Hệ thống phải quăng lỗi 507 (Hết chỗ chứa)
    assert exc_info.value.status_code == 507
    # Khẳng định: DB tuyệt đối chưa được gọi hàm insert file này
    repo.create.assert_not_called()

Tại sao kiểu viết này lại lợi hại?¶

Mock(spec=AudioRepository) — Ép đồ giả phải tuân thủ đúng hợp đồng. Nếu bạn gõ sai tên hàm, test sẽ chửi ngay lập tức.
Dùng AsyncMock thay cho Mock để giả lập các hàm async (Cần dùng await).
Cấu trúc Arrange-Act-Assert (Dàn cảnh - Hành động - Khẳng định) làm code test trong vắt, đọc như văn xuôi.

5. Integration Test — Đụng chạm Database thật¶

Mục tiêu là kiểm tra xem câu SQL mình viết có chọc đúng vào DB không. Tuyệt đối KHÔNG ĐƯỢC MOCK SQLAlchemy.

⚠ Hố tử thần của Newbie: Thấy bảo test là phải mock, thế là bạn hì hục đi mock session.execute(), result.fetchone(), insert().values()... Rốt cuộc, cái test đó chỉ đang chứng minh "bạn có gọi đúng cái hàm của SQLAlchemy không", chứ chả chứng minh được câu SQL có chạy ra đúng data hay không. Đổi code một phát là test gãy tùm lum, nhưng SQL gõ sai thì test vẫn pass!

Tuyệt chiêu: Dùng SQLite in-memory (:memory:) làm DB thật. Nó chạy nhanh như điện, sạch sẽ (tự bay màu khi tắt session), và quan trọng là nó chạy SQL thật!

# tests/test_audio_repo.py
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

from src.db.engine import metadata
from src.repositories.audio_repo import SqliteAudioRepository

@pytest_asyncio.fixture
async def session() -> AsyncSession:
    """Tạo 1 cái DB SQLite ảo trên RAM, dùng xong tự vứt cho mỗi bài test."""
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")

    # Tạo cấu trúc bảng
    async with engine.begin() as conn:
        await conn.run_sync(metadata.create_all)

    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
    async with SessionLocal() as s:
        yield s

    await engine.dispose()


@pytest.mark.asyncio
async def test_create_and_get_by_id(session):
    repo = SqliteAudioRepository()

    # Bóp cò
    created = await repo.create(
        session,
        filename="test.mp3",
        title="Thiền sáng",
        title_normalized="thien sang",
        size_bytes=1024,
        duration_seconds=60,
        category_id=None,
        event_id=None,
    )

    # Chắc chắn đã thêm được vào DB
    assert created["filename"] == "test.mp3"
    assert created["id"] > 0

    # Lôi đầu ra xem có đúng thông tin không
    fetched = await repo.get_by_id(session, created["id"])
    assert fetched["title"] == "Thiền sáng"


@pytest.mark.asyncio
async def test_get_total_size_returns_sum(session):
    repo = SqliteAudioRepository()

    await repo.create(
        session, filename="a.mp3", title="A", title_normalized="a",
        size_bytes=100, duration_seconds=30, category_id=None, event_id=None,
    )
    await repo.create(
        session, filename="b.mp3", title="B", title_normalized="b",
        size_bytes=200, duration_seconds=60, category_id=None, event_id=None,
    )

    total = await repo.get_total_size(session)
    assert total == 300

Khi nào thì không được xài SQLite in-memory mà phải xài Postgres thật?¶

Khi dự án bạn chơi đồ "độc quyền" của Postgres: Query mảng (Array), Full-text search (TSVector), hay JSON path.
Khi đó, bạn bắt buộc phải dùng testcontainers hoặc mồi 1 con DB Postgres thật lên CI/CD để chạy Integration Test. Còn với project KCDS này xài SQLite ở Production, test in-memory là quá đẹp rồi.

6. E2E Test — Đánh sập mọi rào cản với TestClient¶

Kiểm tra một mạch từ ngoài vào trong bằng HTTP Client, lợi dụng tính năng dependency_overrides của FastAPI để tráo quyền:

# tests/test_api_audio.py
import pytest
from httpx import AsyncClient, ASGITransport
from src.main import app
from src.db.engine import get_session
from src.auth import get_current_user

# Tạo ra đồ giả để tráo
async def _fake_session():
    # ... tạo in-memory session như trên
    yield session

def _fake_user():
    return {"email": "test@example.com", "role": "admin", "enabled": True}

@pytest.mark.asyncio
async def test_list_audio_requires_auth():
    # Không tráo user → Mặc định phải bị chửi 401
    app.dependency_overrides.clear()

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
        res = await ac.get("/api/audio")

    assert res.status_code == 401


@pytest.mark.asyncio
async def test_list_audio_returns_data_when_authed():
    # Tráo đồ: Nhét DB ảo và User ảo vào
    app.dependency_overrides[get_session] = _fake_session
    app.dependency_overrides[get_current_user] = _fake_user

    try:
        async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
            res = await ac.get("/api/audio")

        assert res.status_code == 200
        assert "data" in res.json()
    finally:
        app.dependency_overrides.clear()  # Dùng xong nhớ dọn rác

Quyền năng của `dependency_overrides`¶

Nhờ nó, bạn:

Không cần tốn công mock cái thư viện gởi HTTP request.
Tự do thay thế bất cứ mắt xích nào trong dây chuyền mà không cần chọc vào code chính.
Đây chính là phần thưởng xứng đáng nhất cho việc chịu khó viết Depends() ở Bài 3.

7. Test cái gì và tha cho cái gì?¶

Nên dồn sức test:

Luật nghiệp vụ sinh tử: Tính toán giới hạn, logic phân quyền.
Thứ tự thực thi: Có chắc là ghi file rồi mới cắm vô DB không? Có tự lùi lại (rollback) khi hỏng hóc không?
Truy vấn hóc búa: SQL gõ tay có chạy ra đúng cái list như mong muốn không?
Các luồng sống còn: Quá trình đăng nhập, upload file, tự động kích hoạt bot phát nhạc.

Không rảnh đi test:

Tính năng của Framework: Không cần test xem FastAPI có bóc tách được Pydantic không (cha đẻ FastAPI test giùm bạn rồi).
Thư viện bên thứ ba: Không test discord.py hay xem lõi SQLAlchemy hoạt động thế nào.
Mấy cái hàm Get/Set nhảm nhí, hoặc API nào chắc chắn tuần sau sếp đòi đập đi làm lại.

Câu thần chú: Lỗi này mà nổ ra thì tối nay ai mất ngủ? Nếu là bạn → Hãy viết test. Nếu là lão dev nào đó trên Github → Khỏi test.

8. Fixture — Bộ công cụ dùng một lần¶

Pytest hỗ trợ Fixture giúp bạn setup một lần rồi tái sử dụng ở mọi nơi:

# tests/conftest.py — File đặc biệt pytest tự động múc lên
import pytest_asyncio

@pytest_asyncio.fixture
async def session():
    # Bày biện (setup)
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(metadata.create_all)
    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
    async with SessionLocal() as s:
        yield s
    # Dọn dẹp (teardown)
    await engine.dispose()


@pytest_asyncio.fixture
async def audio_repo():
    return SqliteAudioRepository()


@pytest_asyncio.fixture
async def sample_audio(session, audio_repo):
    # Tạo sẵn 1 cái mồi cho các test khác xài
    return await audio_repo.create(
        session, filename="sample.mp3", title="Sample", size_bytes=1024, ...
    )


# Vô test gọi tên nó ra là có xài luôn:
async def test_delete_removes_audio(session, audio_repo, sample_audio):
    deleted = await audio_repo.delete(session, sample_audio["id"])
    assert deleted is True
    assert await audio_repo.get_by_id(session, sample_audio["id"]) is None

Pytest tự động phân tích tham số và bơm đúng đồ nghề vào cho bạn. Mặc định xài xong một cái là nó quăng sọt rác nặn lại cái mới để đảm bảo độ sạch sẽ (clean state).

9. Async Test với pytest-asyncio¶

Nhớ cấu hình để pytest biết cách chạy code async:

# pyproject.toml hoặc pytest.ini
[tool.pytest.ini_options]
asyncio_mode = "auto"    # Tự động nhận diện hàm async và chạy giùm

Hoặc gắn bùa thủ công cho từng test:

@pytest.mark.asyncio
async def test_something():
    ...

Bẫy ngầm: Mặc định pytest-asyncio đẻ ra một cái event loop mới cho mỗi hàm test. Nếu bạn định xài một fixture dùng chung cho toàn phiên chạy (scope="session"), bạn phải báo cho nó biết bằng @pytest_asyncio.fixture(scope="session") kèm cấu hình loop_scope.

10. Bắt bệnh (Debug) Async Test¶

Test treo lơ lửng không chịu chạy (Hang)¶

Quên gõ chữ await ở đâu đó → Lệnh được gọi nhưng không bao giờ nhường quyền.
Lỗi Deadlock: Cho nhiều bài test cùng bâu vào xài chung 1 cái session DB mà không chịu cleanup đoàng hoàng.

Lỗi `RuntimeError: Event loop is closed`¶

Vòng lặp chết rồi mà vẫn đòi xài. Lỗi này 90% là do bạn dùng chung cái engine cho nhiều bài test ở các event loop khác nhau.
Cách chữa: Ép scope="function" để mỗi bài test tự dọn dẹp sạch sẽ ổ của mình.

Test chạy local thì Xanh, lên CI thì Đỏ (Flaky)¶

Thứ tự test bị đảo lộn (do dùng pytest-randomly) làm lộ ra việc các bài test đang âm thầm xài chung biến trạng thái (state leak).
Cách chữa: Code test sao cho "Tao chạy một mình hay tao chạy thứ 100 thì kết quả vẫn y chang". Không dùng biến global trong test.

11. Ánh xạ vào Nguyên lý tổng quát¶

Nguyên lý (Bài 0)	Chuyển hóa thành Testing
DIP (Đảo ngược phụ thuộc)	Là khả năng bơm `Mock Repo` vào Service qua Constructor. Không có DIP thì đừng mơ test!
DI (Tiêm phụ thuộc)	Khả năng xài `app.dependency_overrides` để tráo nguyên liệu dễ như ăn kẹo.
SoC (Tách biệt quan tâm)	Được phản chiếu qua Kim Tự Tháp: Test Unit cho 1 lớp, Test Intg cho 2 lớp kết dính, Test E2E cho nguyên khối.
SRP (Đơn trách nhiệm)	Nguyên tắc test tốt nhất: 1 bài test chỉ nên có 1 lý do để thất bại (Ví dụ: Test hàm A bắt lỗi dung lượng thì chỉ bắt duy nhất mã 507).
Fail Fast (Đứt cầu chì)	Đặt `Mock(spec=...)`, hễ gõ nhầm tên hàm chưa khai báo là test nổ banh xác ngay, đỡ mất công mò lỗi!

Tóm lại: Kiến trúc Test chính là tấm gương phản chiếu kiến trúc Code. Code viết tuân thủ SOLID thì test viết nhàn tênh. Ngược lại, thấy test viết rườm rà, nhọc não → Tự hiểu là code đang vi phạm nguyên lý.

Giá trị chuyển giao: Dù ngày mai bạn chuyển sang code Django, Go hay NestJS, thì tư tưởng Kim tự tháp 3 bậc (Unit/Intg/E2E) và nghệ thuật "tráo đồ giả qua hợp đồng" (mock via interface) vẫn không hề thay đổi. Việc của bạn chỉ là tìm đọc API của thư viện test mới mà thôi.

Bài tập¶

Viết unit test cho AudioService.delete — mock repo, đảm bảo quy tắc "xóa DB trước, xóa file sau" và "xóa file bị lỗi cũng không được raise exception".
Viết integration test cho SqliteAudioRepository.list_all test cái bộ lọc q — đảm bảo nó nhận diện được việc gõ tiếng Việt không dấu (LIKE pattern).
Viết E2E test cho API POST /api/audio — tráo quyền bằng dependency, tống một file MP3 rỗng lên, kiểm tra xem có ói ra HTTP 200 và cái ID mới không.
Cục code dưới đây đang dính chết với nhau rất khó test. Bạn hãy "giải phẫu" nó theo nguyên lý DIP để viết test dễ hơn:

class EmailService:
    def send(self, to: str, subject: str):
        import smtplib
        smtp = smtplib.SMTP("mail.example.com", 587)
        smtp.login(os.getenv("SMTP_USER"), os.getenv("SMTP_PASS"))
        smtp.sendmail("noreply@example.com", to, subject)
        smtp.quit()

Câu hỏi tự kiểm tra¶

Tại sao trên tháp Kim Tự Tháp, lượng Unit Test phải dày đặc mà E2E Test lại lưa thưa?
Khi nào thì xài in-memory SQLite là chân ái, khi nào thì bắt buộc phải triệu hồi con Postgres thật lên test?
Bỏ chữ spec=X vào trong Mock() có tác dụng phong thủy gì?
Đám dependency_overrides giải quyết được ác mộng gì ở cấp API mà trò mock thông thường phải bó tay?
Chạy ở máy mình thì pass, đẩy lên server CI thì sấp mặt đỏ lòm — Kể ra 3 nguyên do tội lỗi nhất?