Блог

Каков наилучший способ тестирования цепочки обработки фреймов данных pandas? Я удалил файл сценария и тестовый файл ниже, чтобы вы могли понять, что я имею в виду.

Я запутался в передовой практике, моя единственная руководящая интуиция — сделать тесты так, чтобы они могли выполняться в любом порядке, ограничить количество раз, когда csv загружается с диска, а также убедиться, что каждая точка в цепочке не изменяет приспособление. Каждый шаг процесса зависит от предыдущих шагов, поэтому модульное тестирование каждого узла похоже на тестирование накопления обработки до этой точки в конвейере. Пока я выполняю миссию, но, похоже, происходит много дублирования кода, потому что я постепенно создаю конвейер в каждом тесте.

Каков способ тестирования такого скрипта на Python?

Это отключенный файл обработки данных:

 #main_script.py

def calc_allocation_methodology(df_row):
    print('calculating allocation methodoloyg')
    return 'simple'

def flag_data_for_the_allocation_methodology(df):       
    allocation_methodology = df.apply(calc_allocation_methodology, axis=1)
    df.assign(allocation_methodology=allocation_methodology)
    print('flagging each row for the allocation methodoloyg')
    return df

def convert_repeating_values_to_nan(df):
    'keep one value and nan the rest of the values'
    print('convert repeating values to nan')
    return df

def melt_and_drop_accounting_columns(df):
    print('melt and drop accounting columns')
    print(f'colums remaining: {df.shape[0]}')
    return df
    
def melt_and_drop_engineering_columns(df):
    print('melt and drop engineering columns')
    print(f'colums remaining: {df.shape[0]}')
    return df
    
    
def process_csv_to_tiny_format(df):
    print('process the entire pipeline')
    return (df
        .pipe(flag_data_for_the_allocation_methodology)
        .pipe(convert_repeating_values_to_nan)
        .pipe(melt_and_drop_accounting_columns)
        .pipe(melt_and_drop_engineering_columns)
        )
 

Это тестовый файл, который был удален

 

#test_main.py


from pytest import fixture
import main_script as main
import pandas as pd

@fixture(scope='session')
def df_from_csv()
    return pd.load_csv('database_dump.csv')

@fixture
def df_copy(df_from_csv):
    df = df_from_csv.copy()
    return df

    
    
def test_expected_flag_data_for_the_allocation_methodology(df_copy)
    df = df_copy
    node_to_test = df.pipe(main.flag_data_for_the_allocation_methodology)
    assert True

def test_convert_repeating_values_to_nan(df_copy)
    df = df_copy
    node_to_test = df.pipe(main.flag_data_for_the_allocation_methodology).pipe(main.convert_repeating_values_to_nan)
    assert True     
        
def test_melt_and_drop_accounting_columns(df_copy)
    df = df_copy
    node_to_test = (df
        .pipe(main.flag_data_for_the_allocation_methodology)
        .pipe(main.convert_repeating_values_to_nan)
        .pipe(main.melt_and_drop_accounting_columns))
    assert True         
    
def test_melt_and_drop_engineering_columns(df_copy)
    df = df_copy
    node_to_test = (df
        .pipe(main.flag_data_for_the_allocation_methodology)
        .pipe(main.convert_repeating_values_to_nan)
        .pipe(main.melt_and_drop_accounting_columns)
        .pipe(main.melt_and_drop_engineering_columns))
    assert True     

def test_process_csv_to_tiny_format(df_from_csv):
    df = df_from_csv.copy()
    tiny_data = main.process_csv_to_tiny_format(df)
    assert True