test_data_quality.py - master - kamball/lab8

test_data_quality.py
58 строк3 KB
kambal23123
first_commit
20 дек 2025, 13:59
20 дек 2025, 13:5967ff299
try:
    from great_expectations import get_context
    from great_expectations.core import ExpectationSuite
    GE_AVAILABLE = True
except ImportError:
    GE_AVAILABLE = False

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')
    
    df = pipeline.transformed_data
    
    # Проверяем ожидания качества данных
    print(" Проверяем качество данных...")
    
    # 1. Проверяем что колонки существуют
    expected_columns = [
        'id', 'name', 'age', 'salary', 'department', 'join_date',
        'experience_years', 'salary_category'
    ]
    assert list(df.columns) == expected_columns, f"Ожидались колонки {expected_columns}, получены {list(df.columns)}"
    print(" Все колонки присутствуют")
    
    # 2. Проверяем уникальность ID
    assert df['id'].is_unique, "ID должны быть уникальными"
    print(" ID уникальны")
    
    # 3. Проверяем что нет пустых значений
    assert df['id'].notna().all(), "ID не должны быть пустыми"
    assert df['name'].notna().all(), "Имена не должны быть пустыми"
    assert df['salary'].notna().all(), "Зарплаты не должны быть пустыми"
    print(" Нет пустых значений в важных полях")
    
    # 4. Проверяем диапазоны значений
    assert (df['age'] >= 18).all() and (df['age'] <= 70).all(), "Возраст должен быть от 18 до 70"
    assert (df['salary'] >= 10000).all() and (df['salary'] <= 200000).all(), "Зарплата должна быть от 10000 до 200000"
    print(" Диапазоны значений корректны")
    
    # 5. Проверяем допустимые значения департаментов
    valid_departments = ['IT', 'HR', 'Finance']
    assert df['department'].isin(valid_departments).all(), f"Департаменты должны быть из {valid_departments}"
    print(" Департаменты валидны")
    
    print("Все проверки качества данных пройдены!")
    
    return True


if __name__ == "__main__":
    test_data_quality()