Pandas & NumPy Academy · Lekcja

Testowanie etapów potoku za pomocą asercji

Dodaj na każdym etapie kontrole liczby wierszy, asercje dotyczące wartości null i zabezpieczenia sprawdzające oczekiwane kolumny, aby błędy były natychmiast widoczne.

Testowanie etapów potoku za pomocą asercji to bezpłatna lekcja Pandas & NumPy Academy na CoddyKit. To lekcja 3 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej Pandas & NumPy Academy, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs Pandas & NumPy Academy zawiera 4 lekcji w sumie.

Dlaczego testować kroki pipeline’u?

Pipeline danych, który działa bez błędów, nadal może tworzyć po cichu nieprawidłowe wyniki: usuwać wiersze z powodu niewłaściwego filtra, mnożyć wartości w kolumnie przez nieprawidłowy współczynnik albo duplikować wiersze podczas scalania, ponieważ klucz złączenia nie był unikatowy. Jedynym sposobem wykrywania takich cichych awarii jest osadzenie asercji sprawdzających oczekiwane właściwości danych na każdym etapie pipeline’u. Asercje zamieniają błędy logiczne w wyraźne i natychmiast widoczne błędy.

import pandas as pd
import numpy as np

# A transform that looks correct but has a bug:
def compute_revenue_buggy(df):
    # BUG: unit_price should be multiplied, not added
    df['revenue'] = df['quantity'] + df['unit_price']
    return df

# Without assertions, this runs silently with wrong numbers.

Kontrola liczby wierszy

Po każdym kroku filtrowania należy sprawdzić za pomocą asercji, czy wynikowa liczba wierszy mieści się w oczekiwanym zakresie. Zbyt mała liczba wierszy oznacza, że filtr był zbyt restrykcyjny, a zbyt duża — że złączenie zduplikowało rekordy. Oczekiwany zakres należy wyrazić jako ułamek danych wejściowych: na przykład krok usuwania wartości null nie powinien usuwać więcej niż 30% wierszy w poprawnych danych. Ten mechanizm ochronny wykrywa nieoczekiwane zmiany jakości danych w źródłach nadrzędnych.

def drop_nulls_guarded(df, required_cols, max_drop_fraction=0.3):
    before = len(df)
    result = df.dropna(subset=required_cols)
    after = len(result)
    drop_fraction = (before - after) / before
    assert drop_fraction <= max_drop_fraction, \
        f'dropna removed {drop_fraction:.1%} of rows (limit {max_drop_fraction:.1%})'
    return result

Kontrola istnienia kolumn

Po każdej funkcji transformacji należy sprawdzić za pomocą asercji, czy istnieją wszystkie oczekiwane kolumny wynikowe. Jeśli krok zmiany nazwy przypadkowo użyje niewłaściwego klucza, wynikowa kolumna będzie brakować, a błąd ujawni się dopiero znacznie później, gdy inny krok spróbuje jej użyć. Asercja umieszczona bezpośrednio po transformacji wykrywa problem u źródła, zamiast dopiero trzy kroki później za pomocą mylącego błędu KeyError.

def compute_revenue(df):
    df = df.assign(revenue=lambda d: d['quantity'] * d['unit_price'])

    # Guard: check that the new column exists and is non-null
    assert 'revenue' in df.columns, 'revenue column not created'
    assert df['revenue'].notna().all(), 'revenue has unexpected NaNs'
    return df

Asercje zakresu wartości

Po obliczeniu kolumny przychodu należy sprawdzić za pomocą asercji, czy nie zawiera ona wartości ujemnych. Po sparsowaniu dat należy sprawdzić, czy mieszczą się one w oczekiwanym zakresie lat. Po zakodowaniu kategorii należy sprawdzić, czy nie pojawiły się nieoczekiwane kody. Każda asercja jest kontraktem danych, który dokumentuje oczekiwane zachowanie i wcześnie wykrywa naruszenia. Należy zapisywać te kontrole jako asercje, a nie instrukcje print, aby podczas automatycznych uruchomień pipeline’u zgłaszały błędy.

def validate_computed_columns(df):
    assert (df['revenue'] >= 0).all(), \
        f'Negative revenue: {df[df["revenue"] < 0]["revenue"].head().tolist()}'
    assert (df['quantity'] > 0).all(), \
        'Non-positive quantity found'
    assert df['revenue'].between(0, 1_000_000).all(), \
        'Revenue out of plausible range'
    print('Value range checks passed.')

Ochrona przed duplikowaniem po scalaniu

Typowym błędem danych jest złączenie wiele-do-wielu, które przypadkowo zwielokrotnia liczbę wierszy. Po każdym pd.merge() należy sprawdzić za pomocą asercji, czy liczba wierszy jest zgodna z oczekiwaniami — zazwyczaj nie powinna przekraczać liczby wierszy w lewym DataFrame w przypadku lewego złączenia. Należy również sprawdzić, czy kolumna klucza jest unikatowa, jeśli powinna taka być, aby natychmiast wykryć przypadkowe złączenia kartezjańskie.

def safe_merge(left, right, on, how='left'):
    before = len(left)
    result = left.merge(right, on=on, how=how)
    after = len(result)

    if how == 'left':
        assert after == before, \
            f'Left join increased rows from {before} to {after} — check key uniqueness in right df'
    return result

Asercja braku wartości null po kluczowych krokach

Niektóre kolumny nigdy nie powinny zawierać wartości null na żadnym etapie pipeline’u. Należy sprawdzać za pomocą asercji df['key_col'].notna().all() po każdym kroku, który może przypadkowo wprowadzić wartości null — na przykład po scaleniu, które nie dopasuje niektórych wierszy (co spowoduje wartości NaN w scalonych kolumnach), albo po wywołaniu map(), które zwróci NaN dla niezamapowanych wartości. Asercje te powinny być dwiema ostatnimi liniami każdej funkcji transformacji operującej na tych kolumnach.

NOT_NULL_AFTER_TRANSFORM = ['order_id', 'revenue', 'category']

def post_transform_checks(df):
    for col in NOT_NULL_AFTER_TRANSFORM:
        null_count = df[col].isna().sum()
        assert null_count == 0, \
            f'{col}: {null_count} unexpected NaN values after transform'
    print('Null checks passed.')
    return df

Tworzenie zestawu testów dla kroków pipeline’u

Należy pisać testy jednostkowe dla każdej funkcji pipeline’u, używając niewielkich, ręcznie przygotowanych DataFrame’ów, które izolują testowaną logikę. Każdy test powinien: przygotować minimalne dane wejściowe, wywołać funkcję i sprawdzić właściwości wyniku za pomocą asercji. W przypadku prostych pipeline’ów wystarczy wbudowana w Pythona instrukcja assert; w większych projektach należy używać pytest do automatycznego uruchamiania wszystkich testów przed wdrożeniem.

def test_compute_revenue():
    test_df = pd.DataFrame({
        'quantity': [2, 3],
        'unit_price': [10.0, 5.0]
    })
    result = compute_revenue(test_df)

    assert 'revenue' in result.columns
    assert result['revenue'].tolist() == [20.0, 15.0]
    assert result['revenue'].dtype == float
    print('test_compute_revenue PASSED')

test_compute_revenue()

Testowanie przypadków brzegowych

Dobre testy obejmują nie tylko poprawną ścieżkę, lecz także przypadki brzegowe: dane wejściowe zawierające wyłącznie wartości null, pusty DataFrame, DataFrame z jednym wierszem oraz kolumny z wartościami skrajnymi. Pusty DataFrame powinien zwrócić pusty DataFrame, a nie błąd. DataFrame z jednym wierszem powinien dać poprawny wynik. Należy jawnie testować te przypadki, aby mieć pewność, że pipeline poradzi sobie z nimi poprawnie w środowisku produkcyjnym, gdy pojawią się nietypowe dane.

def test_empty_df():
    empty = pd.DataFrame({'quantity': [], 'unit_price': []})
    result = compute_revenue(empty)
    assert len(result) == 0, 'Empty input should produce empty output'
    assert 'revenue' in result.columns, 'Revenue column should still be created'
    print('test_empty_df PASSED')

def test_single_row():
    single = pd.DataFrame({'quantity': [1], 'unit_price': [99.0]})
    result = compute_revenue(single)
    assert result['revenue'].iloc[0] == 99.0
    print('test_single_row PASSED')

test_empty_df()
test_single_row()

Test integracyjny: pełny pipeline na przykładowych danych

Oprócz testów jednostkowych poszczególnych funkcji należy napisać test integracyjny, który uruchamia cały pipeline na małej, reprezentatywnej próbce rzeczywistych danych. Należy sprawdzić za pomocą asercji, czy wynik ma oczekiwaną liczbę kolumn, kolumna klucza jest unikatowa, a łączny przychód mieści się w realistycznym zakresie. To sprawdzenie od początku do końca wykrywa błędy w interakcji między krokami, których testy jednostkowe nie ujawniają.

def integration_test(config):
    raw = extract(config)
    clean = transform(raw, config)

    assert set(config['required_cols']).issubset(set(clean.columns))
    assert clean['order_id'].is_unique
    assert (clean['revenue'] >= 0).all()
    assert len(clean) > 0

    print(f'Integration test PASSED. Output: {clean.shape}')

integration_test(CONFIG)

Ciągłe testowanie za pomocą pytest

W miarę rozrastania się pipeline’u należy zebrać wszystkie testy w katalogu tests/ i uruchamiać je z wiersza poleceń za pomocą pytest. Plik conftest.py może tworzyć współdzielone fixture, takie jak przykładowe DataFrame’y. Należy zintegrować pytest z pipeline’em CI/CD, aby każda zmiana w kodzie automatycznie uruchamiała wszystkie testy przed wdrożeniem. Nieudany test blokuje wdrożenie i zapobiega przedostaniu się wadliwego kodu do środowiska produkcyjnego.

# tests/test_transform.py — example structure
# import pytest, pandas as pd
# from pipeline.transform import compute_revenue, drop_nulls

# @pytest.fixture
# def sample_df():
#     return pd.DataFrame({'quantity': [2, 3], 'unit_price': [10.0, 5.0]})

# def test_revenue(sample_df):
#     result = compute_revenue(sample_df)
#     assert result['revenue'].tolist() == [20.0, 15.0]

print('Run: pytest tests/ -v  to execute all pipeline tests')

Asercje jako żywa dokumentacja

Każda asercja w pipeline’ie jest jednocześnie mechanizmem ochronnym i fragmentem dokumentacji: w formie odczytywalnej maszynowo określa, jak muszą wyglądać dane w danym miejscu. Gdy do projektu dołącza nowy analityk i czyta kod pipeline’u, asercje informują go o kontraktach danych bez potrzeby korzystania z osobnego dokumentu specyfikacji. Do każdej asercji należy podchodzić tak jak do komentarza — komunikat powinien być wystarczająco opisowy, aby można było zrozumieć wymuszaną regułę biznesową.

# These assertions document business rules as code:
assert (df['order_date'] >= pd.Timestamp('2020-01-01')).all(), \
    'Company was founded 2020-01-01; no orders can predate this'
assert df['region'].isin({'North', 'South', 'East', 'West', 'Central'}).all(), \
    'Only 5 sales regions are valid; new regions require config update'
print('Business rules verified as assertions.')

Szybkie sprawdzenie

Sprawdź swoją wiedzę na temat analizy danych zdobytą w tej lekcji.

Podsumowanie lekcji

W tej lekcji nauczyli się Państwo: osadzać w pipeline’ie asercje sprawdzające liczbę wierszy, istnienie kolumn, zakresy wartości i wartości null, pisać testy jednostkowe dla poszczególnych funkcji transformacji z uwzględnieniem przypadków brzegowych oraz uruchamiać testy integracyjne i traktować asercje jako żywą dokumentację kontraktów danych. Następnie zajmiemy się planowaniem i rejestrowaniem uruchomień pipeline’u w celu automatycznego wykonywania ich codziennie.

Często zadawane pytania

Czy lekcja „Testowanie etapów potoku za pomocą asercji” jest bezpłatna?

Tak — pełny tekst „Testowanie etapów potoku za pomocą asercji” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu Pandas & NumPy Academy, przejdź na CoddyKit PRO. Kurs Pandas & NumPy Academy zawiera 4 lekcji w sumie.

Co nauczysz się w „Testowanie etapów potoku za pomocą asercji”?

Dodaj na każdym etapie kontrole liczby wierszy, asercje dotyczące wartości null i zabezpieczenia sprawdzające oczekiwane kolumny, aby błędy były natychmiast widoczne. Ćwiczysz Pandas & NumPy Academy z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć Pandas & NumPy Academy?

Nie wymagamy żadnego doświadczenia. Pandas & NumPy Academy w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 3 z 4.

Ile czasu zajmuje lekcja „Testowanie etapów potoku za pomocą asercji”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji Pandas & NumPy Academy?

Tak. Każda lekcja Pandas & NumPy Academy zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

Strukturyzowanie etapów transformacji jako funkcji
Parametryzowanie potoków za pomocą słowników konfiguracji
Testowanie etapów potoku za pomocą asercji
Harmonogramowanie i rejestrowanie uruchomień potoku

← Powrót do Pandas & NumPy Academy