test_performance.py - master - ioplock/lab8

test_performance.py
53 строки2 KB
Ioplock
Init
20 дек 2025, 10:55
20 дек 2025, 10:5560a1b1e
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()