test_performance.py - master - 1lyxxaa/testing_lab_8

test_performance.py
73 строки3 KB
1lyxxaa
first_commit
26 дек 2025, 17:25
26 дек 2025, 17:25b281194
import time
import pandas as pd
import sys
import os

# Добавляем родительскую директорию в путь для импорта
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
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("Тесты производительности пройдены!")
    
    # НЕ возвращаем значение, pytest ожидает None или assertion
    # return execution_time  # УБИРАЕМ return для pytest!

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("Тесты использования памяти пройдены!")
    
    # НЕ возвращаем значение, pytest ожидает None или assertion
    # return memory_usage  # УБИРАЕМ return для pytest!

# Для прямого запуска (не через pytest)
if __name__ == "__main__":
    print("Запуск тестов производительности...")
    
    try:
        # Тест производительности
        test_performance()
    except AssertionError as e:
        print(f"❌ Тест производительности провален: {e}")
        exit(1)
    
    try:
        # Тест использования памяти
        test_memory_usage()
    except AssertionError as e:
        print(f"❌ Тест использования памяти провален: {e}")
        exit(1)
    
    print("\n✅ Все тесты производительности пройдены!")