#!/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"])