#!/usr/bin/env python3 """ CI Server Script - Запускает серверы для тестирования в неблокирующем режиме """ import logging import os import signal import subprocess import sys import threading import time from pathlib import Path from typing import Any, Dict, Optional # Добавляем корневую папку в путь sys.path.insert(0, str(Path(__file__).parent.parent)) # Создаем собственный логгер без дублирования def create_ci_logger(): """Создает логгер для CI без дублирования""" logger = logging.getLogger("ci-server") logger.setLevel(logging.INFO) # Убираем существующие обработчики logger.handlers.clear() # Создаем форматтер formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") # Создаем обработчик handler = logging.StreamHandler() handler.setFormatter(formatter) logger.addHandler(handler) # Отключаем пропагацию к root logger logger.propagate = False return logger logger = create_ci_logger() class CIServerManager: """Менеджер CI серверов""" def __init__(self) -> None: self.backend_process: Optional[subprocess.Popen] = None self.frontend_process: Optional[subprocess.Popen] = None self.backend_pid_file = Path("backend.pid") self.frontend_pid_file = Path("frontend.pid") # Настройки по умолчанию self.backend_host = os.getenv("BACKEND_HOST", "0.0.0.0") self.backend_port = int(os.getenv("BACKEND_PORT", "8000")) self.frontend_port = int(os.getenv("FRONTEND_PORT", "3000")) # Флаги состояния self.backend_ready = False self.frontend_ready = False # Обработчики сигналов для корректного завершения signal.signal(signal.SIGINT, self._signal_handler) signal.signal(signal.SIGTERM, self._signal_handler) def _signal_handler(self, signum: int, frame: Any) -> None: """Обработчик сигналов для корректного завершения""" logger.info(f"Получен сигнал {signum}, завершаем работу...") self.cleanup() sys.exit(0) def start_backend_server(self) -> bool: """Запускает backend сервер""" try: logger.info(f"🚀 Запускаем backend сервер на {self.backend_host}:{self.backend_port}") # Запускаем сервер в фоне self.backend_process = subprocess.Popen( [sys.executable, "dev.py", "--host", self.backend_host, "--port", str(self.backend_port)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, universal_newlines=True, ) # Сохраняем PID self.backend_pid_file.write_text(str(self.backend_process.pid)) logger.info(f"✅ Backend сервер запущен с PID: {self.backend_process.pid}") # Запускаем мониторинг в отдельном потоке threading.Thread(target=self._monitor_backend, daemon=True).start() return True except Exception as e: logger.error(f"❌ Ошибка запуска backend сервера: {e}") return False def start_frontend_server(self) -> bool: """Запускает frontend сервер""" try: logger.info(f"🚀 Запускаем frontend сервер на порту {self.frontend_port}") # Переходим в папку panel panel_dir = Path("panel") if not panel_dir.exists(): logger.error("❌ Папка panel не найдена") return False # Запускаем npm run dev в фоне self.frontend_process = subprocess.Popen( ["npm", "run", "dev"], cwd=panel_dir, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, universal_newlines=True, ) # Сохраняем PID self.frontend_pid_file.write_text(str(self.frontend_process.pid)) logger.info(f"✅ Frontend сервер запущен с PID: {self.frontend_process.pid}") # Запускаем мониторинг в отдельном потоке threading.Thread(target=self._monitor_frontend, daemon=True).start() return True except Exception as e: logger.error(f"❌ Ошибка запуска frontend сервера: {e}") return False def _monitor_backend(self) -> None: """Мониторит backend сервер""" try: while self.backend_process and self.backend_process.poll() is None: time.sleep(1) # Проверяем доступность сервера if not self.backend_ready: try: import requests response = requests.get(f"http://{self.backend_host}:{self.backend_port}/", timeout=5) if response.status_code == 200: self.backend_ready = True logger.info("✅ Backend сервер готов к работе!") else: logger.debug(f"Backend отвечает с кодом: {response.status_code}") except Exception as e: logger.debug(f"Backend еще не готов: {e}") except Exception as e: logger.error(f"❌ Ошибка мониторинга backend: {e}") def _monitor_frontend(self) -> None: """Мониторит frontend сервер""" try: while self.frontend_process and self.frontend_process.poll() is None: time.sleep(1) # Проверяем доступность сервера if not self.frontend_ready: try: import requests response = requests.get(f"http://localhost:{self.frontend_port}/", timeout=5) if response.status_code == 200: self.frontend_ready = True logger.info("✅ Frontend сервер готов к работе!") else: logger.debug(f"Frontend отвечает с кодом: {response.status_code}") except Exception as e: logger.debug(f"Frontend еще не готов: {e}") except Exception as e: logger.error(f"❌ Ошибка мониторинга frontend: {e}") def wait_for_servers(self, timeout: int = 180) -> bool: # Увеличил таймаут """Ждет пока серверы будут готовы""" logger.info(f"⏳ Ждем готовности серверов (таймаут: {timeout}с)...") start_time = time.time() while time.time() - start_time < timeout: logger.debug(f"Backend готов: {self.backend_ready}, Frontend готов: {self.frontend_ready}") if self.backend_ready and self.frontend_ready: logger.info("🎉 Все серверы готовы к работе!") return True time.sleep(3) # Увеличил интервал проверки logger.error("⏰ Таймаут ожидания готовности серверов") logger.error(f"Backend готов: {self.backend_ready}, Frontend готов: {self.frontend_ready}") return False def cleanup(self) -> None: """Очищает ресурсы и завершает процессы""" logger.info("🧹 Очищаем ресурсы...") # Завершаем процессы if self.backend_process: try: self.backend_process.terminate() self.backend_process.wait(timeout=10) except subprocess.TimeoutExpired: self.backend_process.kill() except Exception as e: logger.error(f"Ошибка завершения backend: {e}") if self.frontend_process: try: self.frontend_process.terminate() self.frontend_process.wait(timeout=10) except subprocess.TimeoutExpired: self.frontend_process.kill() except Exception as e: logger.error(f"Ошибка завершения frontend: {e}") # Удаляем PID файлы for pid_file in [self.backend_pid_file, self.frontend_pid_file]: if pid_file.exists(): try: pid_file.unlink() except Exception as e: logger.error(f"Ошибка удаления {pid_file}: {e}") # Убиваем все связанные процессы try: subprocess.run(["pkill", "-f", "python dev.py"], check=False) subprocess.run(["pkill", "-f", "npm run dev"], check=False) subprocess.run(["pkill", "-f", "vite"], check=False) except Exception as e: logger.error(f"Ошибка принудительного завершения: {e}") logger.info("✅ Очистка завершена") def run_tests_in_ci(): """Запускаем тесты в CI режиме""" logger.info("🧪 Запускаем тесты в CI режиме...") # Создаем папку для результатов тестов os.makedirs("test-results", exist_ok=True) # Сначала проверяем здоровье серверов logger.info("🏥 Проверяем здоровье серверов...") try: health_result = subprocess.run( ["uv", "run", "pytest", "tests/test_server_health.py", "-v"], capture_output=False, text=True, timeout=120, # 2 минуты на проверку здоровья ) if health_result.returncode != 0: logger.warning("⚠️ Тест здоровья серверов не прошел, но продолжаем...") else: logger.info("✅ Серверы здоровы!") except Exception as e: logger.warning(f"⚠️ Ошибка при проверке здоровья серверов: {e}, продолжаем...") test_commands = [ (["uv", "run", "pytest", "tests/", "-m", "not e2e", "-v", "--tb=short"], "Unit тесты"), (["uv", "run", "pytest", "tests/", "-m", "integration", "-v", "--tb=short"], "Integration тесты"), (["uv", "run", "pytest", "tests/", "-m", "e2e", "-v", "--tb=short"], "E2E тесты"), (["uv", "run", "pytest", "tests/", "-m", "browser", "-v", "--tb=short", "--timeout=60"], "Browser тесты"), ] for cmd, test_type in test_commands: logger.info(f"🚀 Запускаем {test_type}...") max_retries = 3 # Увеличиваем количество попыток for attempt in range(1, max_retries + 1): logger.info(f"📝 Попытка {attempt}/{max_retries} для {test_type}") try: # Запускаем тесты с выводом в реальном времени result = subprocess.run( cmd, capture_output=False, # Потоковый вывод text=True, timeout=600, # 10 минут на тесты ) if result.returncode == 0: logger.info(f"✅ {test_type} прошли успешно!") break else: if attempt == max_retries: if test_type == "Browser тесты": logger.warning( f"⚠️ {test_type} не прошли после {max_retries} попыток (ожидаемо) - продолжаем..." ) else: logger.error(f"❌ {test_type} не прошли после {max_retries} попыток") return False else: logger.warning( f"⚠️ {test_type} не прошли, повторяем через 10 секунд... (попытка {attempt}/{max_retries})" ) time.sleep(10) except subprocess.TimeoutExpired: logger.error(f"⏰ Таймаут для {test_type} (10 минут)") if attempt == max_retries: return False else: logger.warning(f"⚠️ Повторяем {test_type} через 10 секунд... (попытка {attempt}/{max_retries})") time.sleep(10) except Exception as e: logger.error(f"❌ Ошибка при запуске {test_type}: {e}") if attempt == max_retries: return False else: logger.warning(f"⚠️ Повторяем {test_type} через 10 секунд... (попытка {attempt}/{max_retries})") time.sleep(10) logger.info("🎉 Все тесты завершены!") return True def initialize_test_database(): """Инициализирует тестовую базу данных""" try: logger.info("🗄️ Инициализируем тестовую базу данных...") # Создаем файл базы если его нет db_file = Path("database.db") if not db_file.exists(): db_file.touch() logger.info("✅ Создан файл базы данных") # Импортируем и создаем таблицы from sqlalchemy import inspect from auth.orm import Author, AuthorBookmark, AuthorFollower, AuthorRating from orm.base import Base from orm.community import Community, CommunityAuthor, CommunityFollower from orm.draft import Draft from orm.invite import Invite from orm.notification import Notification from orm.reaction import Reaction from orm.shout import Shout from orm.topic import Topic from services.db import engine logger.info("✅ Engine импортирован успешно") logger.info("Creating all tables...") Base.metadata.create_all(engine) # Проверяем что таблицы созданы inspector = inspect(engine) tables = inspector.get_table_names() logger.info(f"✅ Созданы таблицы: {tables}") # Проверяем критически важные таблицы critical_tables = ["community_author", "community", "author"] missing_tables = [table for table in critical_tables if table not in tables] if missing_tables: logger.error(f"❌ Отсутствуют критически важные таблицы: {missing_tables}") return False else: logger.info("✅ Все критически важные таблицы созданы") return True except Exception as e: logger.error(f"❌ Ошибка инициализации базы данных: {e}") import traceback traceback.print_exc() return False def main(): """Основная функция""" logger.info("🚀 Запуск CI Server Manager") # Создаем менеджер manager = CIServerManager() try: # Инициализируем базу данных if not initialize_test_database(): logger.error("❌ Не удалось инициализировать базу данных") return 1 # Запускаем серверы if not manager.start_backend_server(): logger.error("❌ Не удалось запустить backend сервер") return 1 if not manager.start_frontend_server(): logger.error("❌ Не удалось запустить frontend сервер") return 1 # Ждем готовности if not manager.wait_for_servers(): logger.error("❌ Серверы не готовы в течение таймаута") return 1 logger.info("🎯 Серверы запущены и готовы к тестированию") # В CI режиме запускаем тесты автоматически ci_mode = os.getenv("CI_MODE", "false").lower() logger.info(f"🔧 Проверяем CI режим: CI_MODE={ci_mode}") if ci_mode in ["true", "1", "yes"]: logger.info("🔧 CI режим: запускаем тесты автоматически...") return run_tests_in_ci() else: logger.info("💡 Локальный режим: для запуска тестов нажмите Ctrl+C") # Держим скрипт запущенным try: while True: time.sleep(1) # Проверяем что процессы еще живы if manager.backend_process and manager.backend_process.poll() is not None: logger.error("❌ Backend сервер завершился неожиданно") break if manager.frontend_process and manager.frontend_process.poll() is not None: logger.error("❌ Frontend сервер завершился неожиданно") break except KeyboardInterrupt: logger.info("👋 Получен сигнал прерывания") return 0 except Exception as e: logger.error(f"❌ Критическая ошибка: {e}") return 1 finally: manager.cleanup() if __name__ == "__main__": sys.exit(main())