#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Тест AJAX обновления логов Автор: Сергей Антропов Сайт: https://devops.org.ru """ import asyncio import aiohttp import json import time from datetime import datetime import os import sys # Добавляем корневую директорию в путь sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) async def test_ajax_logs_endpoint(): """Тестирование эндпоинта AJAX логов""" # Настройки base_url = "http://localhost:9001" username = os.getenv("LOGBOARD_USER", "admin") password = os.getenv("LOGBOARD_PASS", "admin") print(f"🧪 Тестирование AJAX обновления логов") print(f"📡 URL: {base_url}") print(f"👤 Пользователь: {username}") print("=" * 50) async with aiohttp.ClientSession() as session: try: # 1. Получаем токен авторизации print("🔐 Получение токена авторизации...") auth_data = { "username": username, "password": password } async with session.post(f"{base_url}/api/auth/login", json=auth_data) as response: if response.status != 200: print(f"❌ Ошибка авторизации: {response.status}") return False auth_response = await response.json() token = auth_response.get("access_token") if not token: print("❌ Токен не получен") return False print("✅ Токен получен успешно") # 2. Получаем список контейнеров print("\n📋 Получение списка контейнеров...") headers = {"Authorization": f"Bearer {token}"} async with session.get(f"{base_url}/api/services", headers=headers) as response: if response.status != 200: print(f"❌ Ошибка получения контейнеров: {response.status}") return False containers = await response.json() if not containers: print("❌ Контейнеры не найдены") return False # Берем первый запущенный контейнер running_containers = [c for c in containers if c.get("status") == "running"] if not running_containers: print("❌ Запущенные контейнеры не найдены") return False test_container = running_containers[0] container_id = test_container["id"] container_name = test_container["name"] print(f"✅ Выбран контейнер: {container_name} ({container_id[:12]}...)") # 3. Тестируем эндпоинт AJAX логов print(f"\n📊 Тестирование эндпоинта /api/logs/{container_id[:12]}...") # Первый запрос print("📤 Первый запрос (получение последних логов)...") url = f"{base_url}/api/logs/{container_id}" params = {"tail": 10} async with session.get(url, headers=headers, params=params) as response: if response.status != 200: print(f"❌ Ошибка получения логов: {response.status}") return False data = await response.json() print(f"✅ Получено {data.get('total_lines', 0)} строк логов") print(f"📅 Временная метка: {data.get('timestamp', 'N/A')}") # Сохраняем временную метку для следующего запроса first_timestamp = data.get('timestamp') if data.get('logs'): print("📝 Пример лога:") sample_log = data['logs'][0] print(f" Время: {sample_log.get('timestamp', 'N/A')}") print(f" Сообщение: {sample_log.get('message', 'N/A')[:100]}...") # 4. Ждем немного и делаем второй запрос print(f"\n⏳ Ожидание 3 секунды...") await asyncio.sleep(3) print("📤 Второй запрос (получение логов без since)...") params = {"tail": 10} async with session.get(url, headers=headers, params=params) as response: if response.status != 200: print(f"❌ Ошибка получения логов: {response.status}") return False data = await response.json() print(f"✅ Получено {data.get('total_lines', 0)} строк логов") print(f"📅 Временная метка: {data.get('timestamp', 'N/A')}") if data.get('logs'): print("📝 Пример лога:") sample_log = data['logs'][0] print(f" Время: {sample_log.get('timestamp', 'N/A')}") print(f" Сообщение: {sample_log.get('message', 'N/A')[:100]}...") # 5. Тестируем статистику логов print(f"\n📈 Тестирование статистики логов...") stats_url = f"{base_url}/api/logs/stats/{container_id}" async with session.get(stats_url, headers=headers) as response: if response.status != 200: print(f"❌ Ошибка получения статистики: {response.status}") return False stats = await response.json() print("✅ Статистика логов:") print(f" DEBUG: {stats.get('debug', 0)}") print(f" INFO: {stats.get('info', 0)}") print(f" WARN: {stats.get('warn', 0)}") print(f" ERROR: {stats.get('error', 0)}") print("\n🎉 Все тесты прошли успешно!") return True except Exception as e: print(f"❌ Ошибка тестирования: {e}") return False async def test_ajax_performance(): """Тестирование производительности AJAX запросов""" print(f"\n🚀 Тестирование производительности AJAX запросов") print("=" * 50) # Настройки base_url = "http://localhost:9001" username = os.getenv("LOGBOARD_USER", "admin") password = os.getenv("LOGBOARD_PASS", "admin") async with aiohttp.ClientSession() as session: try: # Получаем токен auth_data = {"username": username, "password": password} async with session.post(f"{base_url}/api/auth/login", json=auth_data) as response: auth_response = await response.json() token = auth_response.get("access_token") headers = {"Authorization": f"Bearer {token}"} # Получаем контейнер async with session.get(f"{base_url}/api/services", headers=headers) as response: containers = await response.json() running_containers = [c for c in containers if c.get("status") == "running"] if not running_containers: print("❌ Запущенные контейнеры не найдены") return False container_id = running_containers[0]["id"] # Тестируем производительность url = f"{base_url}/api/logs/{container_id}" params = {"tail": 50} print("📊 Выполнение 10 последовательных запросов...") start_time = time.time() for i in range(10): request_start = time.time() async with session.get(url, headers=headers, params=params) as response: await response.json() request_time = (time.time() - request_start) * 1000 print(f" Запрос {i+1}: {request_time:.2f}ms") # Небольшая пауза между запросами await asyncio.sleep(0.1) total_time = time.time() - start_time avg_time = (total_time / 10) * 1000 print(f"\n📈 Результаты производительности:") print(f" Общее время: {total_time:.2f}с") print(f" Среднее время запроса: {avg_time:.2f}ms") print(f" Запросов в секунду: {10/total_time:.2f}") if avg_time < 100: print("✅ Отличная производительность!") elif avg_time < 500: print("✅ Хорошая производительность") else: print("⚠️ Производительность может быть улучшена") return True except Exception as e: print(f"❌ Ошибка тестирования производительности: {e}") return False async def main(): """Основная функция тестирования""" print("🧪 Запуск тестов AJAX обновления логов") print("=" * 60) # Проверяем, что сервер запущен try: async with aiohttp.ClientSession() as session: async with session.get("http://localhost:9001/healthz") as response: if response.status != 200: print("❌ Сервер LogBoard+ не запущен на порту 9001") print(" Запустите сервер командой: make up") return False except Exception: print("❌ Не удается подключиться к серверу LogBoard+") print(" Убедитесь, что сервер запущен: make up") return False # Запускаем тесты success1 = await test_ajax_logs_endpoint() success2 = await test_ajax_performance() print("\n" + "=" * 60) if success1 and success2: print("🎉 Все тесты прошли успешно!") print("✅ AJAX обновление логов работает корректно") return True else: print("❌ Некоторые тесты не прошли") print("🔧 Проверьте логи сервера и настройки") return False if __name__ == "__main__": result = asyncio.run(main()) sys.exit(0 if result else 1)