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

Prerequisite: Bài 3 (DI), Bài 5 (Layer architecture). DI là tiền đề của testability. Project reference: Project hiện chưa có test suite — bài này dùng ví dụ tương đương trên entity của project.

---

1. Tại sao viết test?

Không phải vì "best practice". Lý do cụ thể:

Lý do	Giải thích
Regression safety	Sửa code mà không phá tính năng cũ
Design pressure	Code khó test = code coupling cao = design tệ
Documentation sống	Test thể hiện rõ cách dùng và expected behavior
Refactor confidence	Dám refactor vì có safety net

Code không test được thường là code có hidden dependency bị hardcode bên trong (DB connection, SMTP server, env vars...) — không thể chạy đơn lẻ vì đòi hỏi cả infrastructure thật. Khi gặp trường hợp này, đó là dấu hiệu rõ để refactor theo DIP (bài 0).

---

2. Test Pyramid

         /\
        /E2E\          ← Ít (chậm, đắt, flaky)
       /──────\
      /  Intg  \       ← Vừa (DB thật, network)
     /──────────\
    /    Unit    \     ← Nhiều (nhanh, rẻ, stable)

Loại	Phạm vi	Tốc độ	Khi dùng
Unit	1 class/function, mock dependency	ms	Business logic, pure function
Integration	≥ 2 component thật (service + DB)	giây	Query SQL, transaction, migration
E2E	HTTP → DB → response	giây-phút	Critical user flow (login, upload)

Nguyên tắc: Test bug ở tầng thấp nhất có thể tìm ra. Bug business rule → unit test. Bug SQL → integration test. Bug auth flow → E2E test.

Pyramid ánh xạ thẳng vào 3-layer architecture (bài 5)

Project dùng kiến trúc Router → Service → Repository → 3 loại test tương ứng từng layer:

Loại test	Layer	Mock gì?	Verify gì?
Unit	Service	Toàn bộ Repository (qua ABC)	Business rule: validation, storage limit, transaction order
Integration	Repository	Không mock — DB thật (in-memory SQLite)	SQL query đúng, FK constraint, transaction commit/rollback
E2E	Router (qua httpx.AsyncClient)	Chỉ mock auth dependency	Full chain: middleware + auth + service + DB

Heuristic thực dụng: Khi viết test đầu tiên cho feature mới → bắt đầu ở Service layer (unit test). Service là nơi chứa business rule — test ở đây nhanh, ổn định, và nếu pass → 80% logic đã đúng. Repository test chỉ cần khi viết SQL phức tạp (join, window function). E2E test chỉ cho critical flow (login, upload, delete).

---

3. Tại sao DI làm test dễ?

Nhớ bài 3 — AudioService nhận AudioRepository (ABC) qua constructor:

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

Trong test, inject mock:

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

mock_repo = Mock(spec=AudioRepository)
mock_repo.get_total_size = AsyncMock(return_value=0)
svc = AudioService(mock_repo, Mock(), Mock())

→ Không chạm DB thật. Test chạy trong mili-giây.

Nếu không dùng DI, AudioService() sẽ tự tạo SqliteAudioRepository() ngay bên trong constructor — mọi test đều phải setup SQLite thật, chạy chậm hơn, dễ bị flaky, và cần cleanup database sau mỗi test.

Đây là bằng chứng thực tế cho DIP (bài 0) — không chỉ là lý thuyết.

---

4. Unit Test — Service Layer

Ví dụ: test business rule "upload file quá storage limit → raise HTTPException 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():
    # Arrange — setup mock
    repo = Mock(spec=AudioRepository)
    repo.get_total_size = AsyncMock(return_value=int(9.9 * 1024**3))  # gần đầy 10GB
    repo.get_by_filename = AsyncMock(return_value=None)

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

    # Mock file ~200MB
    file = Mock()
    file.filename = "test.mp3"
    file.read = AsyncMock(return_value=b"x" * (200 * 1024**2))

    # Act + Assert
    with pytest.raises(HTTPException) as exc_info:
        await svc.upload(
            session=Mock(),
            file=file,
            title="Test audio",
            category_id=None,
            event_id=None,
        )

    assert exc_info.value.status_code == 507  # Insufficient Storage
    repo.create.assert_not_called()            # file không được insert

Tại sao pattern này work

1. Mock(spec=AudioRepository) — mock tuân thủ interface. Gọi method không có trong interface → test fail ngay (bảo vệ khỏi typo).
2. AsyncMock thay Mock cho method async — trả coroutine.
3. Arrange-Act-Assert — cấu trúc rõ ràng, dễ đọc.

---

5. Integration Test — Repository + DB

Test SQL query thực sự chạy đúng với SQLite — không mock SQLAlchemy.

⚠ Anti-pattern phổ biến: Cố gắng mock session.execute(), result.fetchone(), insert().values()... Test đó thực ra đang verify xem bạn có gọi đúng method SQLAlchemy không, chứ không verify SQL có chạy đúng không. Kết quả: tốn 50 dòng setup mock, và khi đổi từ .fetchone() sang .scalar_one() phải rewrite test — nhưng bug SQL thật vẫn lọt.

Cách đúng: Dùng SQLite in-memory làm DB thật cho test. Nhanh (ms), sạch (:memory: tự drop khi session close), và test chạy đúng 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:
    """In-memory SQLite cho mỗi test — clean state."""
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")

    # Tạo schema
    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()

    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,
    )

    assert created["filename"] == "test.mp3"
    assert created["id"] > 0

    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

Lưu ý:

• In-memory SQLite (:memory:) cho test — không file I/O, không cleanup
• Fresh session mỗi test — không có state leak
• await engine.dispose() — cleanup connection pool

Khi nào dùng Postgres thật thay vì SQLite in-memory?

• SQLite không support: JSON path queries, full-text search advanced, array columns
• Project dùng Postgres-specific feature → integration test phải chạy Postgres (testcontainers hoặc CI service)
• Project này dùng SQLite production → test SQLite đủ

---

6. E2E Test — FastAPI TestClient

Test toàn bộ request/response qua HTTP, với dependency_overrides:

# 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

# Override dependency trong test
async def _fake_session():
    # ... tạo in-memory session như fixture 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 override get_current_user → mặc định raise 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():
    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()  # cleanup

Tại sao dependency_overrides mạnh

• Không cần mock HTTP client
• Không cần mock DB
• Framework tự resolve dependency chain → thay 1 dependency, cả chain đổi

Đây là lý do bài 3 phân chia Depends(get_session) tách bạch — test mới override được.

---

7. Test cái gì, không test cái gì

Nên test:

• Business rule: storage limit, validation, role check
• Transaction order: file trước DB sau, rollback khi lỗi
• SQL query trả đúng shape/filter/sort
• Critical path: login flow, upload flow, schedule trigger
• Error handling: 404, 403, 422, 507

Không cần test:

• Framework behavior (FastAPI tự parse Pydantic — đã có test của FastAPI)
• Library third-party (discord.py, SQLAlchemy internals)
• Getter/setter trivial
• Code sẽ thay đổi trong vài ngày tới

Heuristic: Bug này xảy ra → ai mất ngủ? Nếu là bạn → viết test. Nếu là người viết library → không cần.

---

8. Fixture — Setup/Teardown tái dùng

# tests/conftest.py — pytest tự load
import pytest_asyncio

@pytest_asyncio.fixture
async def session():
    # 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
    # teardown
    await engine.dispose()


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


@pytest_asyncio.fixture
async def sample_audio(session, audio_repo):
    return await audio_repo.create(
        session, filename="sample.mp3", title="Sample", size_bytes=1024, ...
    )


# Dùng ở test:
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ự tìm fixture theo tên parameter. Scope mặc định là function — recreate mỗi test.

---

9. Async Test với pytest-asyncio

# pyproject.toml hoặc pytest.ini
[tool.pytest.ini_options]
asyncio_mode = "auto"    # tự động mark async test

Hoặc explicit:

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

Gotcha: Event loop scope. Mặc định pytest-asyncio tạo event loop mới mỗi test. Nếu fixture async được share session-scope, bạn cần @pytest_asyncio.fixture(scope="session") và config loop_scope.

---

10. Debugging async test

Async test treo (hang)

• Thiếu await → coroutine tạo nhưng không chạy
• Deadlock do share session giữa tests mà không cleanup

RuntimeError: Event loop is closed

• Fixture scope sai — tái dùng engine giữa event loop khác nhau
• Giải pháp: scope="function" hoặc align event_loop fixture

Test pass local, fail CI

• Random order (pytest-randomly) → state leak giữa test
• Giải pháp: mỗi test fixture fresh session, không dùng biến module-level

---

11. Nguyên lý tổng quát

Nguyên lý (bài 0)	Hiện thực qua testing
DIP	Mock repo qua constructor injection
DI	app.dependency_overrides
SoC	Unit test 1 layer, integration test 2+ layer, E2E full stack
SRP	Test code giống production — 1 test, 1 reason to fail
Fail Fast	Mock(spec=X) crash khi gọi method không có trong interface

Test design là mirror của code design. Code tuân thủ SOLID → test viết dễ. Test viết khó → code vi phạm nguyên lý.

Chuyển giao: Kể cả framework khác (Django, Flask, GraphQL, gRPC), 3 loại test (unit/integration/E2E) và ý tưởng "mock qua interface" không đổi. Chỉ đổi API của test framework.

---

Bài tập

1. Viết unit test cho AudioService.delete — mock repo, test "xóa DB trước, file sau" và "file lỗi không raise".
2. Viết integration test cho SqliteAudioRepository.list_all với filter q — verify LIKE pattern work với tiếng Việt (không dấu).
3. Viết E2E test cho POST /api/audio — override auth + session, upload file MP3 mock, assert 200 và data.id > 0.
4. Code dưới đây khó test — refactor theo DIP:

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

1. Tại sao unit test nhiều hơn E2E test?
2. Khi nào nên dùng SQLite in-memory, khi nào nên dùng Postgres thật (testcontainers)?
3. Mock(spec=X) khác Mock() chỗ nào? Cái nào an toàn hơn?
4. dependency_overrides giải quyết vấn đề gì mà mock thường không?
5. Test pass local fail CI — 3 nguyên nhân phổ biến?