Data API 7 — Kiểm thử, tài liệu & versioning

Sau khi đã dựng được endpoint (FastAPI căn bản), validate dữ liệu (validation), nối database (DB integration), bảo mật (auth) và tối ưu (async & performance), bài này trả lời câu hỏi khó nhất khi một API dữ liệu đã có người dùng: làm sao đổi code mà không làm hỏng consumer? Với NCB, một Data API phục vụ cùng lúc nhiều nhóm — Risk, AML, báo cáo NHNN, app mobile nội bộ. Một thay đổi vô ý đổi tên trường JSON có thể làm gãy job của ba đội mà bạn không hề biết cho đến sáng hôm sau. Bộ ba test tự động + tài liệu chính xác + versioning kỷ luật chính là cái phanh an toàn đó.

1. Kim tự tháp test cho một Data API

Không phải test nào cũng đáng giá như nhau. Ưu tiên nhiều test nhanh ở đáy, ít test chậm ở đỉnh.

Nguyên tắc: logic phức tạp (tính toán, quy tắc nghiệp vụ) nên tách ra hàm thuần để test unit rẻ; endpoint chỉ còn là lớp mỏng ghép validation + query + serialize, và ta test nó ở tầng integration.

2. TestClient và AsyncClient

FastAPI cung cấp TestClient (dựa trên httpx, chạy đồng bộ) — đủ cho phần lớn test endpoint kể cả khi endpoint là async def, vì nó tự chạy event loop bên trong.

# minh hoạ, không chạy được ở sandbox
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_health():
    resp = client.get("/health")
    assert resp.status_code == 200
    assert resp.json() == {"status": "ok"}

Khi cần test hành vi bất đồng bộ thật sự (ví dụ nhiều request đồng thời, hoặc muốn await trong test), dùng httpx.AsyncClient với ASGITransport để gọi thẳng app trong tiến trình, không mở cổng mạng:

# minh hoạ, không chạy được ở sandbox
import pytest
import httpx
from httpx import ASGITransport
from app.main import app

@pytest.mark.anyio
async def test_list_transactions_async():
    transport = ASGITransport(app=app)
    async with httpx.AsyncClient(
        transport=transport, base_url="http://test"
    ) as ac:
        resp = await ac.get("/v1/accounts/12/transactions?limit=10")
    assert resp.status_code == 200
    body = resp.json()
    assert len(body["items"]) <= 10

TestClient dùng cho hầu hết case; AsyncClient khi bạn cần kiểm soát concurrency hoặc code test bản thân nó phải async.

3. pytest: fixture và parametrize

Fixture là cách pytest cung cấp "vật liệu" cho test và tự dọn dẹp sau đó. Đặt các fixture dùng chung trong conftest.py.

# minh hoạ, không chạy được ở sandbox
import pytest
from fastapi.testclient import TestClient
from app.main import app

@pytest.fixture
def client():
    with TestClient(app) as c:   # kích hoạt lifespan startup/shutdown
        yield c

@pytest.fixture
def auth_headers(client):
    token = client.post(
        "/v1/auth/login",
        json={"username": "svc_risk", "password": "..."},
    ).json()["access_token"]
    return {"Authorization": f"Bearer {token}"}

Parametrize giúp một hàm test chạy nhiều bộ dữ liệu — cực hợp để quét các trường hợp validation biên:

# minh hoạ, không chạy được ở sandbox
import pytest

@pytest.mark.parametrize(
    "limit, expected_status",
    [
        (1, 200),
        (100, 200),
        (0, 422),      # dưới ngưỡng cho phép
        (1001, 422),   # vượt trần phân trang
        (-5, 422),
    ],
)
def test_pagination_limit_bounds(client, auth_headers, limit, expected_status):
    resp = client.get(
        f"/v1/accounts/12/transactions?limit={limit}",
        headers=auth_headers,
    )
    assert resp.status_code == expected_status

4. Integration test với DB thật (nhưng cách ly khỏi prod)

Test integration cần một database — nhưng tuyệt đối không chạm production. Ba chiến lược, từ nhẹ tới nặng:

1. Test DB riêng: một Postgres tách biệt (thường qua Docker, xem Docker), tạo schema mới mỗi lần chạy CI. Sạch nhất nhưng chậm.
2. Transaction rollback per-test: mở transaction đầu mỗi test, rollback ở cuối. Nhanh vì không phải tái tạo dữ liệu, và mọi thay đổi biến mất — hợp cho suite lớn.
3. Fixture nạp dữ liệu mẫu (seed): chèn vài record cố định trước khi test.

Mẫu rollback per-test — điểm mấu chốt là ghi đè dependency lấy session để endpoint dùng đúng session đang trong transaction:

# minh hoạ, không chạy được ở sandbox
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.db import get_session

TEST_DB_URL = "postgresql+psycopg://tester:...@localhost:5433/api_test"
engine = create_engine(TEST_DB_URL)
TestingSession = sessionmaker(bind=engine)

@pytest.fixture
def db_session():
    connection = engine.connect()
    trans = connection.begin()          # transaction bao ngoài
    session = TestingSession(bind=connection)
    yield session
    session.close()
    trans.rollback()                    # xoá sạch mọi thay đổi của test
    connection.close()

@pytest.fixture
def client(db_session):
    app.dependency_overrides[get_session] = lambda: db_session
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

Bảo vệ thêm bằng một chốt chặn: fail ngay nếu URL DB trỏ về host prod.

# minh hoạ, không chạy được ở sandbox
assert "prod" not in TEST_DB_URL, "Từ chối chạy test trên DB production!"

5. Bao phủ các trường hợp cần test

Với mỗi endpoint dữ liệu, checklist tối thiểu:

# minh hoạ, không chạy được ở sandbox
def test_not_found(client, auth_headers):
    resp = client.get("/v1/accounts/999999999/transactions",
                      headers=auth_headers)
    assert resp.status_code == 404
    assert resp.json()["detail"] == "Account not found"

def test_missing_token(client):
    resp = client.get("/v1/accounts/12/transactions")
    assert resp.status_code == 401

def test_forbidden_wrong_scope(client, teller_headers):
    # teller không có scope đọc dữ liệu Risk
    resp = client.get("/v1/risk/exposure", headers=teller_headers)
    assert resp.status_code == 403

6. Mock dependency qua dependency_overrides

FastAPI cho phép thay bất kỳ dependency nào lúc test qua app.dependency_overrides. Đây là công cụ mạnh nhất để cách ly: giả lập user đã đăng nhập, thay client dịch vụ ngoài bằng bản giả, hoặc thay session DB.

# minh hoạ, không chạy được ở sandbox
from app.deps import get_current_user
from app.models import User

def fake_risk_user():
    return User(id=1, username="svc_risk", scopes=["risk:read"])

def test_endpoint_with_fake_user(client):
    app.dependency_overrides[get_current_user] = fake_risk_user
    resp = client.get("/v1/risk/exposure")   # bỏ qua login thật
    assert resp.status_code == 200
    app.dependency_overrides.clear()

Lưu ý: chỉ override những dependency là ranh giới ra ngoài (auth, DB, HTTP client tới hệ thống khác). Đừng mock logic nghiệp vụ của chính mình — nếu mock hết thì test không còn kiểm chứng gì.

7. Đo coverage

pytest-cov báo phần trăm dòng code được test chạm tới:

pytest --cov=app --cov-report=term-missing --cov-fail-under=80

Coverage là chỉ báo phủ đến đâu, không phải đúng đến đâu — 100% coverage vẫn có thể sai logic. Với Data API, ưu tiên phủ các nhánh xử lý lỗi (422/401/403/404) và các bộ chuyển đổi dữ liệu, đừng chạy theo con số. Đặt ngưỡng tối thiểu (--cov-fail-under) để tránh tụt lùi. Chi tiết về chất lượng test và CI xem production quality.

8. Tài liệu OpenAPI: tuỳ biến để consumer đọc được

FastAPI tự sinh tài liệu OpenAPI tại /docs (Swagger UI) và /redoc. Nhưng tài liệu "auto" thì khô khan — thêm mô tả, tags, và examples để consumer copy được ngay.

# minh hoạ, không chạy được ở sandbox
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

app = FastAPI(
    title="NCB Data API",
    version="1.3.0",
    description="Dịch vụ tra cứu dữ liệu giao dịch nội bộ NCB.",
    openapi_tags=[
        {"name": "transactions", "description": "Truy vấn giao dịch tài khoản"},
        {"name": "accounts", "description": "Thông tin tài khoản"},
    ],
)

class TransactionOut(BaseModel):
    id: int = Field(..., examples=[9001])
    amount: float = Field(..., examples=[1500000.0])
    kind: str = Field(..., examples=["debit"])

@app.get("/v1/accounts/{account_id}/transactions",
         tags=["transactions"],
         summary="Lịch sử giao dịch theo tài khoản",
         response_model=list[TransactionOut])
def list_tx(account_id: int,
            limit: int = Query(50, ge=1, le=1000)):
    ...

Ta có thể ghi đè hàm sinh schema (app.openapi) khi cần chèn thông tin tuỳ biến (ví dụ mã lỗi chung, thông tin liên hệ team). tags, summary, description, examples là những thứ rẻ mà tăng chất lượng tài liệu nhiều nhất.

9. Export openapi.json cho client codegen

Điểm mạnh lớn nhất của OpenAPI: consumer không phải viết tay client. Xuất file openapi.json rồi sinh SDK cho Python/TypeScript/Java.

# minh hoạ, không chạy được ở sandbox
import json
from app.main import app

with open("openapi.json", "w", encoding="utf-8") as f:
    json.dump(app.openapi(), f, ensure_ascii=False, indent=2)

Sau đó nhóm khác chạy codegen (ví dụ openapi-generator hoặc openapi-python-client) để có client type-safe. Quy trình đề nghị: commit openapi.json vào repo và kiểm trong CI rằng nó khớp code hiện tại — nếu code đổi mà quên regenerate, CI báo lỗi. Đây chính là cách biến "tài liệu" thành "hợp đồng có thể kiểm tra được".

10. API versioning và tương thích ngược

Khi đã có consumer, mọi thay đổi phá vỡ (breaking change) đều phải qua versioning. Hai cách phổ biến:

Đa số team dữ liệu Việt Nam chọn URL path /v1 vì minh bạch và dễ debug qua log/gateway. Trong FastAPI, tách mỗi version thành một APIRouter riêng.

# minh hoạ, không chạy được ở sandbox
from fastapi import APIRouter

v1 = APIRouter(prefix="/v1", tags=["v1"])
v2 = APIRouter(prefix="/v2", tags=["v2"])
app.include_router(v1)
app.include_router(v2)

Phân biệt thay đổi:

• Non-breaking (thêm field optional, thêm endpoint mới, nới lỏng validation): không cần version mới.
• Breaking (đổi tên/xoá field, đổi kiểu, siết validation, đổi ý nghĩa mã lỗi): bắt buộc /v2.

Deprecation cần có quy trình, không tắt đột ngột: đánh dấu endpoint cũ deprecated=True (hiển thị trong OpenAPI), trả header Deprecation/Sunset báo ngày ngừng, thông báo consumer, chạy song song /v1 và /v2 trong một khoảng chuyển tiếp (ví dụ 3 tháng), rồi mới gỡ.

11. Contract/schema stability cho consumer

Điều consumer sợ nhất là schema âm thầm đổi. Ba lớp bảo vệ:

1. Snapshot test schema: lưu openapi.json làm mốc, test so sánh; nếu khác mà không cố ý thì fail — buộc developer nhận thức được mình đang đổi hợp đồng.
2. Consumer-driven contract: nhóm dùng đăng ký kỳ vọng (trường nào, kiểu gì) thành test chạy trên API; provider không được phá.
3. Chỉ mở rộng, không co lại: thêm field thoải mái (client cũ bỏ qua), nhưng xoá/đổi field thì phải qua version mới.

Ví dụ snapshot đơn giản:

# minh hoạ, không chạy được ở sandbox
import json
from app.main import app

def test_openapi_schema_unchanged():
    with open("tests/snapshots/openapi.json") as f:
        expected = json.load(f)
    assert app.openapi() == expected, (
        "Schema thay đổi! Nếu cố ý: cập nhật snapshot và bump version."
    )

12. CI chạy test tự động

Test chỉ có giá trị khi chạy trên mọi thay đổi, không phụ thuộc ai đó nhớ gõ lệnh. Cấu hình pipeline (xem CI/CD) dựng test DB, chạy pytest + coverage, và kiểm openapi.json còn khớp.

Sandbox của NCB là PostgreSQL chỉ đọc, nên minh hoạ một truy vấn kiểm thử dữ liệu tham chiếu (đếm số giao dịch mỗi tài khoản) — đúng loại dữ liệu integration test của endpoint transactions sẽ so khớp:

-- ▶ Chạy được
SELECT account_id, COUNT(*) AS n_tx, SUM(amount) AS total
FROM transactions
GROUP BY account_id
ORDER BY n_tx DESC
LIMIT 10;

Use case thực tế

Bối cảnh. Data API của NCB cung cấp endpoint GET /v1/accounts/{id}/transactions cho ba nhóm: đội AML (giám sát giao dịch), đội báo cáo NHNN, và app mobile nội bộ. Team backend muốn đổi trường amount (số) thành object {value, currency} để hỗ trợ đa tệ.

Nếu làm ẩu: đổi thẳng trên /v1. Kết quả: job AML nightly parse amount như số → crash lúc 2 giờ sáng; báo cáo NHNN sai số liệu; app mobile hiện [object Object].

Quy trình đúng đã áp dụng:

1. Xác định đây là breaking change → tạo /v2 với schema mới.
2. Viết integration test cho /v2 (happy path 200, phân trang, 404, 401/403) trên test DB rollback, không chạm prod.
3. Cập nhật openapi.json snapshot; CI cảnh báo schema đổi → buộc bump version 1.3.0 → 2.0.0.
4. Đánh dấu /v1 deprecated=True, trả header Sunset: Wed, 30 Sep 2026.
5. Export openapi.json cho ba nhóm tự sinh client; chạy song song /v1 + /v2 trong 3 tháng.
6. Theo dõi log gateway: khi lưu lượng /v1 về 0, gỡ bỏ.

Kết quả: không nhóm nào bị gãy; coverage suite giữ ở 84%; CI chặn được 2 PR vô tình đổi schema mà quên bump version.

Ghi nhớ

• Kim tự tháp: nhiều unit (validator/logic thuần) → ít integration (endpoint+DB+auth) → rất ít E2E.
• TestClient cho phần lớn test; httpx.AsyncClient khi cần kiểm soát concurrency hoặc test bản thân phải async.
• pytest: fixture cấp vật liệu + dọn dẹp, parametrize quét nhiều case biên (nhất là validation).
• Integration test dùng test DB riêng hoặc transaction rollback per-test; luôn có chốt chặn từ chối URL prod.
• Phủ đủ: happy path 200, validation 422, auth 401/403, not found 404, phân trang.
• app.dependency_overrides để mock auth/DB/HTTP ngoài — chỉ mock ranh giới ra ngoài, đừng mock logic của mình.
• Coverage là chỉ báo phủ, không phải chỉ báo đúng; đặt --cov-fail-under để chống tụt lùi.
• Tuỳ biến OpenAPI bằng tags/summary/examples; export openapi.json và kiểm khớp trong CI để biến tài liệu thành hợp đồng kiểm tra được.
• Versioning: URL /v1 minh bạch; breaking change → version mới; deprecation có Sunset + chạy song song rồi mới gỡ.
• Chỉ mở rộng (thêm field optional), không co lại; dùng snapshot/contract test để bảo vệ consumer.
• Mọi test phải chạy tự động trong CI trên mỗi PR, kèm dựng test DB và kiểm schema.