test_performance.py - master - burning_hell/lab8_etl_testing

test_performance.py
50 строк2 KB
burning-hel
all tests completed
28 ноя 2025, 21:02
28 ноя 2025, 21:0299d081b
import time
import pandas as pd
from etl_pipeline import ETLPipeline

def test_performance():
    """Тестируем производительность ETL процесса"""
    
    pipeline = ETLPipeline()
    
    # Измеряем время выполнения
    start_time = time.time()
    
    # Запускаем ETL
    result = pipeline.run_pipeline('data/raw_data.csv')
    
    end_time = time.time()
    execution_time = end_time - start_time
    
    print(f"⏱️  Время выполнения ETL: {execution_time:.2f} секунд")
    
    # Проверяем что выполняется достаточно быстро
    # Для 7 записей должно быть меньше 5 секунд
    assert execution_time < 5, f"ETL выполняется слишком долго: {execution_time} секунд"
    
    # Проверяем объем данных
    assert len(result) == 7, f"Ожидалось 7 записей, получено {len(result)}"
    
    print("✅ Тесты производительности пройдены!")
    
    return execution_time

def test_memory_usage():
    """Тестируем использование памяти"""
    pipeline = ETLPipeline()
    pipeline.run_pipeline('data/raw_data.csv')
    
    # Проверяем размер данных в памяти
    memory_usage = pipeline.transformed_data.memory_usage(deep=True).sum()
    print(f"💾 Использование памяти: {memory_usage} байт")
    
    # Для 7 записей должно быть меньше 10KB
    assert memory_usage < 10000, f"Слишком большое использование памяти: {memory_usage} байт"
    
    print("✅ Тесты использования памяти пройдены!")
    
    return memory_usage

if __name__ == "__main__":
    test_performance()
    test_memory_usage()