85757ca717
- Добавлен CORSMiddleware в app.py для решения проблем с CORS - Создан test_api.py для тестирования API endpoints - Создан test_websocket.py для тестирования WebSocket соединений - Создан test_browser.html для тестирования в браузере - Все тесты показывают, что API и WebSocket работают корректно Диагностика показала: ✅ API endpoints работают (health, auth, containers, logs) ✅ WebSocket соединения работают и передают логи ✅ Проблема может быть в браузере или localStorage Для тестирования в браузере: 1. Откройте http://localhost:9001/test_browser.html 2. Выполните тесты по порядку 3. Проверьте консоль браузера на ошибки Автор: Сергей Антропов Сайт: https://devops.org.ru
88 lines
3.3 KiB
Python
88 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
Тестовый скрипт для проверки WebSocket соединений LogBoard+
|
|
Автор: Сергей Антропов
|
|
Сайт: https://devops.org.ru
|
|
"""
|
|
|
|
import asyncio
|
|
import websockets
|
|
import json
|
|
import requests
|
|
|
|
async def test_websocket():
|
|
"""Тестирование WebSocket соединений"""
|
|
|
|
base_url = "http://localhost:9001"
|
|
|
|
# 1. Получаем токен
|
|
print("🔐 Получение токена...")
|
|
login_data = {"username": "admin", "password": "admin"}
|
|
try:
|
|
response = requests.post(f"{base_url}/api/auth/login", json=login_data)
|
|
if response.status_code != 200:
|
|
print(f" ❌ Ошибка входа: {response.status_code}")
|
|
return
|
|
|
|
token = response.json()["access_token"]
|
|
print(" ✅ Токен получен")
|
|
except Exception as e:
|
|
print(f" ❌ Ошибка входа: {e}")
|
|
return
|
|
|
|
# 2. Получаем список контейнеров
|
|
print("\n🔍 Получение контейнеров...")
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
try:
|
|
response = requests.get(f"{base_url}/api/containers/services", headers=headers)
|
|
if response.status_code != 200:
|
|
print(f" ❌ Ошибка получения контейнеров: {response.status_code}")
|
|
return
|
|
|
|
containers = response.json()
|
|
print(f" ✅ Получено контейнеров: {len(containers)}")
|
|
|
|
if not containers:
|
|
print(" ❌ Нет контейнеров для тестирования")
|
|
return
|
|
|
|
first_container = containers[0]
|
|
container_id = first_container.get('id')
|
|
container_name = first_container.get('name')
|
|
print(f" 📦 Тестируем контейнер: {container_name} ({container_id})")
|
|
|
|
except Exception as e:
|
|
print(f" ❌ Ошибка получения контейнеров: {e}")
|
|
return
|
|
|
|
# 3. Тестируем WebSocket соединение
|
|
print(f"\n🔌 Тестирование WebSocket для {container_name}...")
|
|
|
|
# Формируем WebSocket URL
|
|
ws_url = f"ws://localhost:9001/api/websocket/logs/{container_id}?tail=10&token={token}&service={container_name}"
|
|
print(f" 🔗 WebSocket URL: {ws_url}")
|
|
|
|
try:
|
|
async with websockets.connect(ws_url) as websocket:
|
|
print(" ✅ WebSocket соединение установлено")
|
|
|
|
# Ждем сообщения
|
|
print(" 📡 Ожидание сообщений...")
|
|
message_count = 0
|
|
max_messages = 5
|
|
|
|
async for message in websocket:
|
|
message_count += 1
|
|
print(f" 📨 Сообщение {message_count}: {message[:100]}...")
|
|
|
|
if message_count >= max_messages:
|
|
print(f" ✅ Получено {message_count} сообщений")
|
|
break
|
|
|
|
except Exception as e:
|
|
print(f" ❌ Ошибка WebSocket: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(test_websocket())
|