# test_runner.py
#!/usr/bin/env python3
"""
Расширенный скрипт для запуска тестов с различными опциями и индикаторами выполнения.
Этот скрипт предоставляет гибкий интерфейс для запуска тестов pytest с различными
параметрами, включая параллельное выполнение, измерение покрытия кода и фильтрацию
тестов по ключевым словам.
"""
import subprocess
import sys
import argparse
import os
from pathlib import Path
import time
from datetime import datetime
[документация]
def get_project_root():
"""Возвращает корневую директорию проекта.
:return: Объект Path, представляющий корневую директорию проекта
:rtype: pathlib.Path
"""
return Path(__file__).parent
[документация]
def run_tests_with_spinner(command, cwd=None):
"""Запускает тесты с визуальным индикатором выполнения.
Выполняет команду pytest с отображением спиннера во время выполнения
и подробной информацией о времени выполнения.
:param command: Список аргументов командной строки для выполнения
:type command: list[str]
:param cwd: Рабочая директория для выполнения команды (по умолчанию None)
:type cwd: str or pathlib.Path, optional
:return: True если команда выполнена успешно, False в случае ошибки
:rtype: bool
"""
print(f"🕐 [{datetime.now().strftime('%H:%M:%S')}] Выполняю: {' '.join(command)}")
start_time = time.time()
try:
# Запускаем процесс
process = subprocess.Popen(
command,
cwd=cwd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace"
)
# Простой спиннер
spinner = ['|', '/', '-', '\\']
spinner_index = 0
print("⏳ Выполнение: ", end="", flush=True)
# Ждем завершения процесса с визуальным индикатором
while process.poll() is None:
print(f"\r⏳ Выполнение: {spinner[spinner_index % len(spinner)]}", end="", flush=True)
spinner_index += 1
time.sleep(0.2)
print(f"\r🕐 [{datetime.now().strftime('%H:%M:%S')}] Завершено!{' ' * 20}")
# Получаем результат
stdout, stderr = process.communicate()
end_time = time.time()
duration = end_time - start_time
print(f"🕐 Общее время: {duration:.2f} секунд")
if process.returncode == 0:
print("✅ Успешно")
return True
else:
print("❌ Ошибка:")
if stdout:
print("STDOUT:", stdout)
if stderr:
print("STDERR:", stderr)
return False
except Exception as e:
end_time = time.time()
duration = end_time - start_time
print(f"\r🕐 [{datetime.now().strftime('%H:%M:%S')}] Прервано через {duration:.2f} секунд")
print(f"❌ Ошибка выполнения: {e}")
return False
[документация]
def run_tests(test_pattern=None, verbose=True, coverage=False, parallel=False):
"""Запускает тесты с заданными параметрами.
Конфигурирует и выполняет команду pytest с указанными параметрами,
включая фильтрацию по ключевым словам, измерение покрытия и параллельное
выполнение.
:param test_pattern: Паттерн для фильтрации тестов по ключевым словам (по умолчанию None)
:type test_pattern: str, optional
:param verbose: Флаг подробного вывода (по умолчанию True)
:type verbose: bool
:param coverage: Флаг включения измерения покрытия кода (по умолчанию False)
:type coverage: bool
:param parallel: Флаг включения параллельного выполнения тестов (по умолчанию False)
:type parallel: bool
:return: True если тесты выполнены успешно, False в случае ошибки
:rtype: bool
"""
project_root = get_project_root()
# Базовая команда
command = [sys.executable, "-m", "pytest"]
# Добавляем директорию с тестами
command.append("test_suite/")
# Добавляем паттерн тестов, если указан
if test_pattern:
command.extend(["-k", test_pattern])
# Добавляем флаги
if verbose:
command.append("-v")
if coverage:
try:
import pytest_cov
command.extend(["--cov=src", "--cov-report=html", "--cov-report=term"])
except ImportError:
print("⚠️ pytest-cov не установлен. Установите его для coverage: pip install pytest-cov")
if parallel:
try:
import pytest_xdist
command.append("-n auto")
except ImportError:
print("⚠️ pytest-xdist не установлен. Установите его для параллельного запуска: pip install pytest-xdist")
command.extend(["--tb=short"])
# Используем функцию с индикатором вместо обычного subprocess.run
return run_tests_with_spinner(command, cwd=project_root)
[документация]
def check_pytest():
"""Проверяет наличие pytest и выводит информацию о нем.
Выполняет проверку доступности pytest в текущем окружении и
отображает версию установленного пакета.
:return: True если pytest доступен, False в случае ошибки
:rtype: bool
"""
print("🔍 Проверяю наличие pytest...")
try:
result = subprocess.run(
[sys.executable, "-m", "pytest", "--version"],
capture_output=True,
text=True,
timeout=10
)
if result.returncode == 0:
print(f"✅ pytest найден: {result.stdout.strip()}")
return True
else:
print("❌ Ошибка при проверке pytest")
return False
except subprocess.TimeoutExpired:
print("❌ Таймаут при проверке pytest")
return False
except Exception as e:
print(f"❌ Ошибка при проверке pytest: {e}")
return False
[документация]
def main():
"""Основная точка входа в скрипт.
Парсит аргументы командной строки, проверяет окружение и
запускает тесты с указанными параметрами.
:return: True если все тесты пройдены успешно, False в случае ошибок
:rtype: bool
"""
parser = argparse.ArgumentParser(description="Запуск тестов проекта")
parser.add_argument("-k", "--keyword", help="Запустить тесты по ключевому слову")
parser.add_argument("-q", "--quiet", action="store_true", help="Минимальный вывод")
parser.add_argument("--cov", action="store_true", help="Запустить с coverage")
parser.add_argument("-n", "--parallel", action="store_true", help="Параллельный запуск тестов")
parser.add_argument("--slow", action="store_true", help="Запустить медленные тесты")
args = parser.parse_args()
print("🚀 Запуск тестов проекта")
print("=" * 60)
print(f"📁 Рабочая директория: {get_project_root()}")
print(f"🕐 Начало: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
# Проверяем наличие pytest
if not check_pytest():
print("❌ pytest не найден. Установите его: pip install pytest")
return False
# Проверяем наличие директории с тестами
test_dir = get_project_root() / "test_suite"
if not test_dir.exists():
print(f"❌ Директория с тестами не найдена: {test_dir}")
return False
test_files = list(test_dir.glob('test_*.py'))
print(f"📋 Найдено тестов: {len(test_files)} файлов")
# Запускаем тесты
print(f"\n📋 Запуск тестов с параметрами:")
print(f" Ключевое слово: {args.keyword or 'все тесты'}")
print(f" Coverage: {'включен' if args.cov else 'выключен'}")
print(f" Параллельный запуск: {'включен' if args.parallel else 'выключен'}")
print(f" Подробный вывод: {'выключен' if args.quiet else 'включен'}")
success = run_tests(
test_pattern=args.keyword,
verbose=not args.quiet,
coverage=args.cov,
parallel=args.parallel
)
end_time = datetime.now()
print("=" * 60)
if success:
print(f"🎉 Все тесты успешно пройдены! Завершено: {end_time.strftime('%Y-%m-%d %H:%M:%S')}")
return True
else:
print(f"💥 Некоторые тесты провалены! Завершено: {end_time.strftime('%Y-%m-%d %H:%M:%S')}")
return False
if __name__ == "__main__":
try:
success = main()
sys.exit(0 if success else 1)
except KeyboardInterrupt:
print("\n⚠️ Прервано пользователем")
sys.exit(1)
except Exception as e:
print(f"\n💥 Критическая ошибка: {e}")
sys.exit(1)