Python hiện đại 7 — Testing & Chất lượng cho code data

Vì sao code data cũng CẦN test

Có một hiểu lầm phổ biến trong nhiều đội dữ liệu: "test là việc của mấy anh backend viết API, còn tôi làm pipeline báo cáo thì cần gì test." Thực tế ngược lại — code data có rủi ro âm thầm mà code ứng dụng không có. Một API lỗi thì trả về mã 500, người dùng kêu ngay. Một pipeline biến đổi sai thì vẫn chạy trơn tru, vẫn ra một con số trông hợp lý, và con số sai đó đi thẳng vào báo cáo lãnh đạo, vào mô hình rủi ro, vào quyết định cấp tín dụng. Chuỗi hậu quả rất rõ: pipeline sai → số liệu sai → quyết định sai. Không ai thấy stack trace, chỉ thấy tiền.

Với một ngân hàng như NCB, một phép join nhân đôi dòng, một bộ lọc debit bị đảo chiều, một cột NULL bị sum thành 0 thay vì loại bỏ — tất cả đều là lỗi lặng lẽ, không "nổ", nhưng làm lệch tổng dư nợ hay tổng giao dịch của cả một chi nhánh. Test chính là lưới an toàn để bắt những lỗi này trước khi chúng lên production.

Test CODE khác test DỮ LIỆU

Cần phân biệt rành mạch hai loại kiểm tra, vì chúng bắt hai loại lỗi khác nhau:

• Test code (bài này): kiểm tra logic biến đổi có đúng không. Cho đầu vào cố định đã biết, hàm tong_hop_giao_dich() có ra đúng kết quả kỳ vọng không? Đây là thứ chạy trong CI mỗi khi có commit, dữ liệu là dữ liệu mẫu do bạn dựng.
• Test dữ liệu (data quality): kiểm tra dữ liệu thật chảy qua pipeline có "sạch" không — cột khoá có bị NULL, giá trị có nằm trong khoảng hợp lệ, số dòng có sụt bất thường so với hôm qua. Đây là kiểm tra runtime trên dữ liệu production, làm bằng công cụ như Great Expectations, dbt tests, hay assertion trong pipeline. Xem sâu ở Data Quality.

Cùng một chữ "test" nhưng khác mục tiêu: một cái bảo vệ code không đổi hành vi ngoài ý muốn, cái kia bảo vệ dữ liệu vào đạt chuẩn. Bài này tập trung loại đầu, nhưng cả hai đều cần và bổ sung cho nhau.

pytest — nền tảng kiểm thử

pytest là framework test tiêu chuẩn của hệ sinh thái Python. Điểm mạnh: viết test chỉ cần một hàm bắt đầu bằng test_ và câu lệnh assert thuần Python — không cần class, không cần API rườm rà.

# tests/test_transform.py
from ncb.pipeline import tong_hop_giao_dich   # hàm cần test (minh hoạ)

def test_tong_theo_chi_nhanh():
    du_lieu = [
        {"chi_nhanh": "HN", "so_tien": 100, "loai": "debit"},
        {"chi_nhanh": "HN", "so_tien": 50,  "loai": "debit"},
        {"chi_nhanh": "HCM","so_tien": 30,  "loai": "credit"},
    ]
    ket_qua = tong_hop_giao_dich(du_lieu, loai="debit")
    assert ket_qua == {"HN": 150}   # HCM bị loại vì là credit

Chạy pytest -q là xong. pytest tự phát hiện file test_*.py, chạy mọi hàm test_*, và khi assert sai nó diễn giải rõ giá trị hai vế — không cần assertEqual như unittest cũ.

Fixtures — dựng dữ liệu / kết nối tạm

Fixture là cách pytest cung cấp "nguyên liệu" cho test: một DataFrame mẫu, một kết nối DB tạm, một file cấu hình. Fixture khai báo bằng decorator @pytest.fixture, và test nào cần chỉ việc nhận nó làm tham số. pytest lo việc dựng trước và dọn sau.

import pytest
import polars as pl

@pytest.fixture
def df_giao_dich():
    # dữ liệu mẫu tái sử dụng cho nhiều test (minh hoạ)
    return pl.DataFrame({
        "chi_nhanh": ["HN", "HN", "HCM"],
        "so_tien":   [100, 50, 30],
        "loai":      ["debit", "debit", "credit"],
    })

def test_loc_debit(df_giao_dich):
    out = df_giao_dich.filter(pl.col("loai") == "debit")
    assert out.height == 2

Fixture có thể có scope (function mặc định, module, session) để kiểm soát dựng lại bao lâu một lần — một kết nối DB nặng nên đặt scope="session". Fixture cũng có thể dọn dẹp bằng yield:

@pytest.fixture
def db_tam():
    conn = tao_ket_noi_sqlite_in_memory()   # minh hoạ
    yield conn                               # trả cho test dùng
    conn.close()                             # chạy sau khi test xong

parametrize — nhiều ca trong một hàm

Thay vì copy-paste một test cho từng ca, @pytest.mark.parametrize chạy cùng logic với nhiều bộ đầu vào/đầu ra. Mỗi bộ là một test độc lập, fail cái nào báo cái đó.

@pytest.mark.parametrize("loai, mong_doi", [
    ("debit",  {"HN": 150}),
    ("credit", {"HCM": 30}),
    ("unknown", {}),          # ca biên: không có giao dịch nào
])
def test_tong_hop_nhieu_loai(loai, mong_doi):
    du_lieu = [
        {"chi_nhanh": "HN", "so_tien": 100, "loai": "debit"},
        {"chi_nhanh": "HN", "so_tien": 50,  "loai": "debit"},
        {"chi_nhanh": "HCM","so_tien": 30,  "loai": "credit"},
    ]
    assert tong_hop_giao_dich(du_lieu, loai=loai) == mong_doi

Đây là công cụ đắt giá nhất cho code data: các ca biên (empty, một dòng, giá trị âm, NULL, currency lạ) thường là nơi pipeline vỡ, và parametrize giúp liệt kê chúng gọn gàng.

tmp_path, marker, conftest.py

• tmp_path — fixture built-in cho một thư mục tạm riêng cho mỗi test, tự xoá sau. Rất hợp khi test hàm đọc/ghi Parquet, CSV: ghi ra tmp_path, đọc lại, so sánh — không đụng file thật, không rác để lại.

def test_ghi_doc_parquet(df_giao_dich, tmp_path):
    p = tmp_path / "gd.parquet"
    df_giao_dich.write_parquet(p)
    lai = pl.read_parquet(p)
    assert lai.equals(df_giao_dich)

• Marker — nhãn gắn lên test để nhóm/lọc: @pytest.mark.slow cho test chậm, rồi chạy pytest -m "not slow" để bỏ qua khi cần nhanh. Khai báo marker trong cấu hình để tránh cảnh báo.
• conftest.py — file đặt fixture dùng chung cho cả thư mục test; pytest tự nạp, không cần import. Đặt df_giao_dich, db_tam ở đây thì mọi file test trong thư mục đều dùng được.

So sánh DataFrame Polars / pandas

Điểm bẫy hay gặp: không so sánh DataFrame bằng == — với pandas nó trả về DataFrame boolean chứ không phải một bool, dẫn tới lỗi "ambiguous truth value". Dùng đúng công cụ:

• Polars: df1.equals(df2), hoặc from polars.testing import assert_frame_equal để có thông báo lỗi chi tiết (lệch ở cột/dòng nào).
• pandas: from pandas.testing import assert_frame_equal — kiểm cả kiểu dữ liệu, cho phép dung sai số thực (rtol, atol), thứ tự cột.

from polars.testing import assert_frame_equal

def test_chuan_hoa_currency():
    ra = chuan_hoa(df_vao)                      # minh hoạ
    ky_vong = pl.DataFrame({"currency": ["VND", "USD"]})
    assert_frame_equal(ra, ky_vong, check_column_order=False)

Property-based testing với Hypothesis

pytest kiểm những ca bạn nghĩ ra. Nhưng lỗi thật thường nấp ở ca bạn không nghĩ tới. Hypothesis giải bài này bằng property-based testing: thay vì viết vài đầu vào cố định, bạn khai báo tính chất bất biến (invariant) phải luôn đúng, rồi Hypothesis sinh hàng trăm đầu vào ngẫu nhiên để cố phá vỡ nó. Khi tìm được ca phản ví dụ, nó tự "co nhỏ" (shrink) về ca đơn giản nhất còn fail — cực dễ debug.

Với logic dữ liệu, cách tư duy này rất hợp vì nhiều phép biến đổi có bất biến toán học rõ ràng. Ví dụ kinh điển: tổng sau khi gộp nhóm phải bằng tổng gốc (gộp nhóm không được làm mất hay đẻ thêm tiền).

from hypothesis import given, strategies as st

@given(st.lists(
    st.tuples(
        st.sampled_from(["HN", "HCM", "DN"]),        # chi nhánh
        st.integers(min_value=0, max_value=10_000),  # số tiền
    ),
    max_size=200,
))
def test_bat_bien_tong_khong_doi(ban_ghi):
    tong_goc = sum(tien for _, tien in ban_ghi)
    theo_nhom = tong_hop_theo_chi_nhanh(ban_ghi)     # minh hoạ
    assert sum(theo_nhom.values()) == tong_goc       # tổng phải bảo toàn

Các bất biến hay dùng cho pipeline dữ liệu: số dòng sau filter ≤ số dòng gốc; join một-nhiều không được vượt tích Descartes; sắp xếp rồi sắp lại cho kết quả ổn định (idempotent); round-trip ghi-rồi-đọc Parquet cho lại đúng dữ liệu. Một test Hypothesis thay được cho hàng chục test cố định và thường bắt đúng ca biên (danh sách rỗng, số 0, trùng khoá) mà con người hay quên.

Mocking I/O và API

Test tốt phải nhanh và tất định (deterministic) — không phụ thuộc mạng, không phụ thuộc database thật. Nếu hàm của bạn gọi API tỷ giá hay đọc từ core banking, test không được thật sự gọi ra ngoài: mạng có thể chậm, sập, hoặc trả dữ liệu khác nhau mỗi lần. Giải pháp là mock — thay phần I/O bằng một bản giả có hành vi biết trước.

• monkeypatch (fixture built-in của pytest) — thay tạm một hàm/thuộc tính/biến môi trường trong phạm vi test, tự khôi phục sau.

def test_quy_doi_ty_gia(monkeypatch):
    # thay hàm gọi API bằng bản giả trả tỷ giá cố định (minh hoạ)
    monkeypatch.setattr("ncb.fx.lay_ty_gia", lambda cur: 25_000)
    assert quy_doi_ve_vnd(2, "USD") == 50_000

• responses (hoặc respx cho async) — thư viện giả lập tầng HTTP: khai báo "khi gọi URL này thì trả JSON kia", để test code dùng requests/httpx mà không chạm mạng. Hợp khi muốn kiểm cả xử lý lỗi (giả HTTP 500, timeout) — thứ khó tái hiện với API thật.

Nguyên tắc: mock ở biên hệ thống (chỗ chạm mạng, đĩa, DB), giữ logic biến đổi chạy thật. Đừng mock quá sâu tới mức test chỉ còn kiểm chính cái mock của mình.

Kiểm TĨNH: bắt lỗi trước khi chạy

Test chạy code để tìm lỗi runtime. Kiểm tĩnh (static analysis) tìm lỗi mà không chạy code — nhanh hơn, rẻ hơn, bắt được lớp lỗi khác.

• mypy / pyright — bộ kiểm tra kiểu (type checker). Nếu code có type hints, chúng phát hiện gọi hàm sai kiểu, quên xử lý None, truy cập thuộc tính không tồn tại — trước khi chạy. Với pipeline dữ liệu, nơi một hàm nhận DataFrame trả dict, khai báo kiểu rõ giúp bắt sai lệch hợp đồng sớm. Nền tảng type hints và Pydantic xem Pydantic & typing.
• ruff — linter kiêm formatter viết bằng Rust, gộp vai trò flake8 + isort + black + nhiều plugin, chạy gần như tức thì trên codebase lớn. Bắt import thừa, biến không dùng, lỗi phong cách, và tự sửa (ruff check --fix, ruff format). Chi tiết thiết lập ở uv & ruff.

Ba tầng bổ sung nhau: ruff giữ code sạch và đúng phong cách, mypy đảm bảo kiểu nhất quán, pytest kiểm hành vi. Không cái nào thay được cái nào.

Coverage và chiến lược test cho pipeline

Coverage (độ bao phủ) đo bao nhiêu phần trăm dòng code được test chạm tới, qua pytest-cov: pytest --cov=ncb --cov-report=term-missing liệt kê cả những dòng chưa được test. Nhưng nhớ: coverage cao không đồng nghĩa test tốt — bạn có thể chạy qua một dòng mà không assert gì về nó. Coverage giúp tìm vùng trắng, không phải mục tiêu tự thân. Đừng chạy đua tới 100% một cách máy móc; tập trung vào logic biến đổi quan trọng.

Với một pipeline dữ liệu, chiến lược test chia hai tầng, theo hình kim tự tháp:

• Unit — test từng hàm biến đổi riêng lẻ với dữ liệu mẫu cố định. Nhiều, nhanh, tất định. Đây là phần lớn số test.
• Integration — chạy cả pipeline từ đầu tới cuối nhưng trên một mẫu nhỏ đại diện (vài chục dòng bao gồm ca biên), so kết quả với bảng kỳ vọng. Bắt lỗi ở chỗ nối giữa các bước mà unit test bỏ sót.
• Test dbt / SQL (nhắc ngắn) — nếu phần biến đổi nằm trong dbt hay SQL, dùng dbt test (unique, not_null, relationships, accepted_values) hoặc kiểm tra tự viết để chốt cùng loại bất biến ngay trong tầng SQL.

pre-commit và CI — cổng chất lượng

Có test tốt nhưng quên chạy thì vô nghĩa. Cần tự động hoá để chất lượng không phụ thuộc trí nhớ ai. Hai lớp gác cổng:

pre-commit — hook chạy trước mỗi commit trên máy lập trình viên, chặn commit nếu ruff/mypy fail. Bắt lỗi sớm nhất, ngay tại chỗ, trước khi kịp đẩy lên. Cấu hình bằng file .pre-commit-config.yaml (minh hoạ):

# .pre-commit-config.yaml (minh hoạ)
repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    rev: v0.6.0
    hooks:
      - id: ruff           # lint, tự sửa
        args: [--fix]
      - id: ruff-format    # format
  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v1.11.0
    hooks:
      - id: mypy

Chạy pre-commit install một lần, từ đó mỗi git commit tự kích hoạt các hook.

CI (Continuous Integration) — chạy lại toàn bộ ruff + mypy + pytest trên máy chủ mỗi khi có Pull Request. Đây là gác cổng không thể vòng qua: pre-commit có thể bị bỏ qua (--no-verify), nhưng CI thì không — cấu hình để test là điều kiện bắt buộc để merge (required check). PR nào có test fail thì không merge được, chấm hết.

Luồng chuẩn: pre-commit bắt lỗi rẻ ngay tại máy; CI là hàng rào cuối cùng trước khi code chạm nhánh chính. Kết quả: không commit nào phá pipeline lọt được lên production mà không bị chặn.

Use case thực tế

Bối cảnh (NCB). Đội dữ liệu duy trì một pipeline báo cáo tổng hợp giao dịch theo chi nhánh hằng ngày: đọc sao kê từ core banking, lọc theo loại giao dịch, gộp nhóm theo chi nhánh, tính tổng và trung bình, rồi nạp vào bảng phục vụ dashboard cho khối kinh doanh. Trước đây pipeline không có test: mỗi lần sửa logic (thêm loại giao dịch, đổi cách xử lý giao dịch huỷ) là một lần "cầu nguyện" — chỉ biết đúng/sai khi báo cáo sáng hôm sau ra số lạ và có người thắc mắc.

Sự cố điển hình. Một lần, lập trình viên đổi bộ lọc để "gộp thêm giao dịch hoàn tiền", vô tình khiến một join nhân đôi dòng với các giao dịch có nhiều bút toán. Tổng giao dịch của vài chi nhánh lớn phồng lên khoảng 15–20%. Pipeline vẫn chạy xanh, số vẫn ra, dashboard vẫn hiển thị — sai. Phải mất hai ngày và một cuộc họp căng thẳng mới truy ra nguyên nhân, sau khi báo cáo sai đã tới tay lãnh đạo.

Sau khi dựng cổng chất lượng. Đội thêm:

1. Unit test cho từng hàm transform (lọc, gộp, quy đổi) với dữ liệu mẫu cố định.
2. Một property test Hypothesis chốt bất biến "tổng sau gộp nhóm = tổng gốc" — chính bất biến mà sự cố nhân đôi dòng đã vi phạm.
3. Integration test chạy cả pipeline trên mẫu 50 dòng có sẵn ca giao dịch nhiều bút toán, so với bảng kỳ vọng.
4. pre-commit (ruff + mypy) và CI đặt pytest làm required check để merge.

Kết quả (ước lượng minh hoạ). Lần kế tiếp có người sửa logic gây nhân đôi dòng, property test đỏ ngay trong CI của PR — bug bị bắt trong vài phút, trước khi merge, thay vì hai ngày sau khi lên production. Ước tính đội chặn được phần lớn nhóm lỗi biến đổi loại "lặng lẽ làm lệch số" ngay ở tầng PR. Con số 15–20% và "hai ngày" là mô tả một sự cố cụ thể để thấy độ lớn; giá trị thật của bộ test là những sự cố tương tự về sau không còn xảy ra trên production — thứ khó đo nhưng dễ cảm nhận qua việc không còn "báo cáo sáng ra số lạ".

Ghi nhớ

• Code data cũng CẦN test vì lỗi pipeline âm thầm: không nổ, vẫn ra số trông hợp lý — pipeline sai → số liệu sai → quyết định sai. Test là lưới an toàn bắt lỗi trước production.
• Phân biệt test code (logic biến đổi, chạy trong CI trên dữ liệu mẫu) và test dữ liệu / data quality (kiểm dữ liệu thật lúc runtime — xem gov-03). Cả hai đều cần.
• pytest: hàm test_* + assert; fixtures dựng dữ liệu/kết nối tạm (có scope, dọn bằng yield); parametrize cho nhiều ca (nhất là ca biên); tmp_path cho file tạm; marker để nhóm/lọc; conftest.py cho fixture dùng chung.
• So sánh DataFrame KHÔNG dùng == — dùng polars.testing.assert_frame_equal / pandas.testing.assert_frame_equal (kiểm kiểu, dung sai, thứ tự cột).
• Hypothesis (property-based): khai báo bất biến (vd tổng sau gộp = tổng gốc), sinh dữ liệu ngẫu nhiên để phá và tự shrink phản ví dụ — rất hợp logic dữ liệu.
• Mock ở biên I/O: monkeypatch thay hàm/biến, responses/respx giả HTTP — để test nhanh, tất định, không phụ thuộc mạng.
• Kiểm TĨNH: mypy/pyright (kiểu — xem pymod-05) và ruff (lint/format — xem pymod-04) bắt lỗi trước khi chạy; bổ sung chứ không thay pytest.
• Coverage giúp tìm vùng chưa test, không phải mục tiêu tự thân; đừng đua 100% máy móc.
• Pipeline test theo kim tự tháp: nhiều unit (hàm transform) + integration end-to-end trên mẫu nhỏ; dbt/SQL dùng dbt test.
• pre-commit (ruff+mypy tại máy) + CI (ruff+mypy+pytest+coverage mỗi PR) làm cổng chất lượng; đặt test là required check — điều kiện bắt buộc để merge.
• Xem bản đồ cả series ở Python hiện đại 1 — Tổng quan.

Nguồn tham khảo

• pytest Documentation — fixtures, parametrize, tmp_path, marker, conftest.py, monkeypatch.
• Hypothesis Documentation — property-based testing, strategies, shrinking.
• Coverage.py Documentation và pytest-cov — đo độ bao phủ.
• mypy Documentation — kiểm tra kiểu tĩnh cho Python.
• Ruff Documentation — linter kiêm formatter (Astral).
• pre-commit Documentation — quản lý git hook cổng chất lượng.
• pandas.testing.assert_frame_equal và Polars Testing API (polars.testing.assert_frame_equal) — so sánh DataFrame trong test.