310 lines
13 KiB
Python
310 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Тесты для PrometheusServer
|
||
"""
|
||
|
||
import pytest
|
||
import asyncio
|
||
import sys
|
||
import os
|
||
from unittest.mock import Mock, AsyncMock, patch, MagicMock
|
||
from aiohttp import web
|
||
from aiohttp.test_utils import TestClient
|
||
|
||
# Добавляем путь к модулям мониторинга
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../../infra/monitoring'))
|
||
|
||
from prometheus_server import PrometheusServer
|
||
|
||
|
||
class TestPrometheusServer:
|
||
"""Тесты для класса PrometheusServer"""
|
||
|
||
@pytest.fixture
|
||
def prometheus_server(self):
|
||
"""Создает экземпляр PrometheusServer для тестов"""
|
||
return PrometheusServer(host='127.0.0.1', port=9091)
|
||
|
||
@pytest.fixture
|
||
def mock_metrics_collector(self):
|
||
"""Создает мок MetricsCollector"""
|
||
mock_collector = Mock()
|
||
mock_collector.os_type = "ubuntu"
|
||
mock_collector.get_metrics_data.return_value = {
|
||
'cpu_usage_percent': 25.5,
|
||
'ram_usage_percent': 60.2,
|
||
'disk_usage_percent': 45.8,
|
||
'load_average_1m': 1.2,
|
||
'load_average_5m': 1.1,
|
||
'load_average_15m': 1.0,
|
||
'swap_usage_percent': 10.5,
|
||
'disk_io_percent': 15.3,
|
||
'system_uptime_seconds': 86400.0,
|
||
'monitor_uptime_seconds': 3600.0
|
||
}
|
||
return mock_collector
|
||
|
||
def test_init(self, prometheus_server):
|
||
"""Тест инициализации PrometheusServer"""
|
||
assert prometheus_server.host == '127.0.0.1'
|
||
assert prometheus_server.port == 9091
|
||
assert prometheus_server.metrics_collector is not None
|
||
assert isinstance(prometheus_server.app, web.Application)
|
||
|
||
def test_setup_routes(self, prometheus_server):
|
||
"""Тест настройки маршрутов"""
|
||
routes = list(prometheus_server.app.router.routes())
|
||
# aiohttp создает по 2 маршрута для каждого эндпоинта (GET и HEAD)
|
||
assert len(routes) == 6
|
||
|
||
# Проверяем наличие всех маршрутов
|
||
route_paths = [route.resource.canonical for route in routes]
|
||
assert '/' in route_paths
|
||
assert '/metrics' in route_paths
|
||
assert '/health' in route_paths
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_root_handler(self, prometheus_server):
|
||
"""Тест главного обработчика"""
|
||
request = Mock()
|
||
response = await prometheus_server.root_handler(request)
|
||
|
||
assert isinstance(response, web.Response)
|
||
assert response.status == 200
|
||
assert response.content_type == 'text/plain'
|
||
assert 'Prometheus Metrics Server' in response.text
|
||
assert '/metrics' in response.text
|
||
assert '/health' in response.text
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_health_handler(self, prometheus_server):
|
||
"""Тест health check обработчика"""
|
||
request = Mock()
|
||
response = await prometheus_server.health_handler(request)
|
||
|
||
assert isinstance(response, web.Response)
|
||
assert response.status == 200
|
||
assert response.content_type == 'text/plain'
|
||
assert response.text == 'OK'
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_metrics_handler_success(self, prometheus_server, mock_metrics_collector):
|
||
"""Тест обработчика метрик при успешном получении данных"""
|
||
# Заменяем metrics_collector на мок
|
||
prometheus_server.metrics_collector = mock_metrics_collector
|
||
|
||
request = Mock()
|
||
response = await prometheus_server.metrics_handler(request)
|
||
|
||
assert isinstance(response, web.Response)
|
||
assert response.status == 200
|
||
assert response.content_type == 'text/plain'
|
||
|
||
# Проверяем, что метрики содержат ожидаемые данные
|
||
metrics_text = response.text
|
||
assert '# HELP system_info System information' in metrics_text
|
||
assert '# TYPE system_info gauge' in metrics_text
|
||
assert 'system_info{os="ubuntu"}' in metrics_text
|
||
assert '# HELP cpu_usage_percent CPU usage percentage' in metrics_text
|
||
assert 'cpu_usage_percent 25.5' in metrics_text
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_metrics_handler_error(self, prometheus_server, mock_metrics_collector):
|
||
"""Тест обработчика метрик при ошибке"""
|
||
# Настраиваем мок для вызова исключения
|
||
mock_metrics_collector.get_metrics_data.side_effect = Exception("Test error")
|
||
prometheus_server.metrics_collector = mock_metrics_collector
|
||
|
||
request = Mock()
|
||
response = await prometheus_server.metrics_handler(request)
|
||
|
||
assert isinstance(response, web.Response)
|
||
assert response.status == 500
|
||
assert response.content_type == 'text/plain'
|
||
assert 'Error: Test error' in response.text
|
||
|
||
def test_format_prometheus_metrics(self, prometheus_server, mock_metrics_collector):
|
||
"""Тест форматирования метрик в Prometheus формат"""
|
||
prometheus_server.metrics_collector = mock_metrics_collector
|
||
|
||
metrics_data = mock_metrics_collector.get_metrics_data()
|
||
formatted_metrics = prometheus_server._format_prometheus_metrics(metrics_data)
|
||
|
||
# Проверяем структуру метрик
|
||
lines = formatted_metrics.split('\n')
|
||
|
||
# Проверяем наличие системной информации
|
||
assert any('system_info' in line for line in lines)
|
||
assert any('os="ubuntu"' in line for line in lines)
|
||
|
||
# Проверяем наличие CPU метрик
|
||
assert any('cpu_usage_percent' in line for line in lines)
|
||
assert any('25.5' in line for line in lines)
|
||
|
||
# Проверяем наличие RAM метрик
|
||
assert any('ram_usage_percent' in line for line in lines)
|
||
assert any('60.2' in line for line in lines)
|
||
|
||
# Проверяем наличие disk метрик
|
||
assert any('disk_usage_percent' in line for line in lines)
|
||
assert any('45.8' in line for line in lines)
|
||
|
||
# Проверяем наличие load average метрик
|
||
assert any('load_average_1m' in line for line in lines)
|
||
assert any('1.2' in line for line in lines)
|
||
|
||
def test_format_prometheus_metrics_empty_data(self, prometheus_server):
|
||
"""Тест форматирования метрик с пустыми данными"""
|
||
empty_metrics = {}
|
||
formatted_metrics = prometheus_server._format_prometheus_metrics(empty_metrics)
|
||
|
||
# Должна быть только системная информация
|
||
lines = formatted_metrics.split('\n')
|
||
assert len(lines) == 3 # system_info help, type, value
|
||
assert any('system_info' in line for line in lines)
|
||
|
||
def test_format_prometheus_metrics_partial_data(self, prometheus_server, mock_metrics_collector):
|
||
"""Тест форматирования метрик с частичными данными"""
|
||
prometheus_server.metrics_collector = mock_metrics_collector
|
||
|
||
# Только CPU метрики
|
||
partial_metrics = {
|
||
'cpu_usage_percent': 50.0,
|
||
'load_average_1m': 2.5
|
||
}
|
||
|
||
formatted_metrics = prometheus_server._format_prometheus_metrics(partial_metrics)
|
||
lines = formatted_metrics.split('\n')
|
||
|
||
# Проверяем, что есть системная информация + CPU + load average
|
||
assert any('system_info' in line for line in lines)
|
||
assert any('cpu_usage_percent' in line for line in lines)
|
||
assert any('load_average_1m' in line for line in lines)
|
||
assert any('50.0' in line for line in lines)
|
||
assert any('2.5' in line for line in lines)
|
||
|
||
# Проверяем, что нет RAM метрик
|
||
assert not any('ram_usage_percent' in line for line in lines)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_start_and_stop(self, prometheus_server):
|
||
"""Тест запуска и остановки сервера"""
|
||
# Мокаем web.AppRunner и TCPSite
|
||
with patch('prometheus_server.web.AppRunner') as mock_runner_class, \
|
||
patch('prometheus_server.web.TCPSite') as mock_site_class:
|
||
|
||
mock_runner = Mock()
|
||
mock_runner.setup = AsyncMock()
|
||
mock_runner.cleanup = AsyncMock()
|
||
mock_runner_class.return_value = mock_runner
|
||
|
||
mock_site = Mock()
|
||
mock_site.start = AsyncMock()
|
||
mock_site_class.return_value = mock_site
|
||
|
||
# Запускаем сервер
|
||
runner = await prometheus_server.start()
|
||
|
||
# Проверяем, что методы были вызваны
|
||
mock_runner.setup.assert_called_once()
|
||
mock_site.start.assert_called_once()
|
||
assert runner == mock_runner
|
||
|
||
# Останавливаем сервер
|
||
await prometheus_server.stop(runner)
|
||
mock_runner.cleanup.assert_called_once()
|
||
|
||
def test_different_os_types(self):
|
||
"""Тест работы с разными типами ОС"""
|
||
# Тестируем macOS
|
||
with patch('platform.system', return_value='Darwin'):
|
||
server_macos = PrometheusServer()
|
||
assert server_macos.metrics_collector.os_type == "macos"
|
||
|
||
# Тестируем Linux
|
||
with patch('platform.system', return_value='Linux'):
|
||
server_linux = PrometheusServer()
|
||
assert server_linux.metrics_collector.os_type == "ubuntu"
|
||
|
||
# Тестируем неизвестную ОС
|
||
with patch('platform.system', return_value='Windows'):
|
||
server_unknown = PrometheusServer()
|
||
assert server_unknown.metrics_collector.os_type == "unknown"
|
||
|
||
def test_custom_host_port(self):
|
||
"""Тест создания сервера с пользовательскими параметрами"""
|
||
server = PrometheusServer(host='192.168.1.100', port=9092)
|
||
assert server.host == '192.168.1.100'
|
||
assert server.port == 9092
|
||
|
||
def test_metrics_collector_integration(self, prometheus_server):
|
||
"""Тест интеграции с MetricsCollector"""
|
||
# Проверяем, что metrics_collector имеет необходимые методы
|
||
collector = prometheus_server.metrics_collector
|
||
assert hasattr(collector, 'get_metrics_data')
|
||
assert hasattr(collector, 'os_type')
|
||
|
||
# Проверяем, что можем получить данные
|
||
metrics_data = collector.get_metrics_data()
|
||
assert isinstance(metrics_data, dict)
|
||
|
||
|
||
class TestPrometheusServerIntegration:
|
||
"""Интеграционные тесты для PrometheusServer"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_server_creation_integration(self):
|
||
"""Интеграционный тест создания сервера"""
|
||
server = PrometheusServer(host='127.0.0.1', port=0)
|
||
|
||
# Проверяем, что сервер создался
|
||
assert server is not None
|
||
assert server.host == '127.0.0.1'
|
||
assert server.port == 0
|
||
|
||
# Проверяем, что приложение создалось
|
||
assert server.app is not None
|
||
|
||
# Проверяем, что маршруты настроены
|
||
routes = list(server.app.router.routes())
|
||
assert len(routes) > 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_metrics_collector_integration(self):
|
||
"""Интеграционный тест с MetricsCollector"""
|
||
server = PrometheusServer(host='127.0.0.1', port=0)
|
||
|
||
# Проверяем, что можем получить метрики
|
||
metrics_data = server.metrics_collector.get_metrics_data()
|
||
assert isinstance(metrics_data, dict)
|
||
|
||
# Проверяем, что можем отформатировать метрики
|
||
prometheus_metrics = server._format_prometheus_metrics(metrics_data)
|
||
assert isinstance(prometheus_metrics, str)
|
||
assert len(prometheus_metrics) > 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_endpoint_handlers_integration(self):
|
||
"""Интеграционный тест обработчиков эндпоинтов"""
|
||
server = PrometheusServer(host='127.0.0.1', port=0)
|
||
|
||
# Тестируем корневой обработчик
|
||
request = Mock()
|
||
response = await server.root_handler(request)
|
||
assert response.status == 200
|
||
assert 'Prometheus Metrics Server' in response.text
|
||
|
||
# Тестируем health обработчик
|
||
response = await server.health_handler(request)
|
||
assert response.status == 200
|
||
assert response.text == 'OK'
|
||
|
||
# Тестируем metrics обработчик
|
||
response = await server.metrics_handler(request)
|
||
assert response.status == 200
|
||
assert '# HELP system_info' in response.text
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v"])
|