test_data_quality.py - master - Slor/lab8

test_data_quality.py
50 строк2 KB
Slor
upload files
06 дек 2025, 11:53
06 дек 2025, 11:53424a1b3
import great_expectations as ge
import pandas as pd
from etl_pipeline import ETLPipeline

def test_data_quality():
    """Тестируем качество данных с помощью Great Expectations"""
    
    # Запускаем ETL процесс
    pipeline = ETLPipeline()
    pipeline.run_pipeline('data/raw_data.csv')
    
    # Создаем Great Expectations dataset
    df = ge.from_pandas(pipeline.transformed_data)
    
    # Проверяем ожидания качества данных
    print("🧪 Проверяем качество данных...")
    
    # 1. Проверяем что колонки существуют
    expected_columns = [
        'id', 'name', 'age', 'salary', 'department', 'join_date',
        'experience_years', 'salary_category'
    ]
    assert df.expect_table_columns_to_match_ordered_list(expected_columns).success
    
    # 2. Проверяем уникальность ID
    assert df.expect_column_values_to_be_unique('id').success
    
    # 3. Проверяем что нет пустых значений
    assert df.expect_column_values_to_not_be_null('id').success
    assert df.expect_column_values_to_not_be_null('name').success
    assert df.expect_column_values_to_not_be_null('salary').success
    
    # 4. Проверяем диапазоны значений
    assert df.expect_column_values_to_be_between('age', 18, 70).success
    assert df.expect_column_values_to_be_between('salary', 10000, 200000).success
    
    # 5. Проверяем допустимые значения департаментов
    valid_departments = ['IT', 'HR', 'Finance']
    assert df.expect_column_values_to_be_in_set('department', valid_departments).success
    
    print("✅ Все проверки качества данных пройдены!")
    
    # Показываем результаты
    results = df.validate()
    print(f"Успешных проверок: {results['success']}")
    
    return results

if __name__ == "__main__":
    test_data_quality()