#!/usr/bin/env python3 """ Общие фикстуры для тестов инфраструктуры """ import pytest import asyncio import sys import os from unittest.mock import Mock, AsyncMock, patch from pathlib import Path # Добавляем путь к модулям мониторинга sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../../infra/monitoring')) # Настройка pytest-asyncio pytest_plugins = ('pytest_asyncio',) @pytest.fixture(scope="session") def event_loop(): """Создает event loop для асинхронных тестов""" loop = asyncio.get_event_loop_policy().new_event_loop() yield loop loop.close() @pytest.fixture def mock_metrics_data(): """Создает мок данных метрик для тестов""" return { '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 } @pytest.fixture def mock_system_info(): """Создает мок системной информации для тестов""" return { 'cpu_percent': 25.5, 'load_avg_1m': 1.2, 'load_avg_5m': 1.1, 'load_avg_15m': 1.0, 'cpu_count': 8, 'io_wait_percent': 2.5, 'ram_used': 8.0, 'ram_total': 16.0, 'ram_percent': 50.0, 'swap_used': 1.0, 'swap_total': 2.0, 'swap_percent': 50.0, 'disk_used': 100.0, 'disk_total': 500.0, 'disk_percent': 20.0, 'disk_free': 400.0, 'disk_read_speed': '1.0 MB/s', 'disk_write_speed': '512.0 KB/s', 'disk_io_percent': 15, 'system_uptime': '1д 0ч 0м', 'monitor_uptime': '1ч 0м', 'server_hostname': 'test-host', 'current_time': '2025-01-01 12:00:00' } @pytest.fixture def mock_psutil(): """Создает мок для psutil""" mock_psutil = Mock() # Мокаем CPU mock_psutil.cpu_percent.return_value = 25.5 mock_psutil.getloadavg.return_value = (1.2, 1.1, 1.0) mock_psutil.cpu_count.return_value = 8 # Мокаем память mock_memory = Mock() mock_memory.used = 8 * (1024**3) # 8 GB mock_memory.total = 16 * (1024**3) # 16 GB mock_psutil.virtual_memory.return_value = mock_memory mock_swap = Mock() mock_swap.used = 1 * (1024**3) # 1 GB mock_swap.total = 2 * (1024**3) # 2 GB mock_swap.percent = 50.0 mock_psutil.swap_memory.return_value = mock_swap # Мокаем диск mock_disk = Mock() mock_disk.used = 100 * (1024**3) # 100 GB mock_disk.total = 500 * (1024**3) # 500 GB mock_disk.free = 400 * (1024**3) # 400 GB mock_psutil.disk_usage.return_value = mock_disk # Мокаем disk I/O mock_disk_io = Mock() mock_disk_io.read_count = 1000 mock_disk_io.write_count = 500 mock_disk_io.read_bytes = 1024 * (1024**2) # 1 GB mock_disk_io.write_bytes = 512 * (1024**2) # 512 MB mock_psutil.disk_io_counters.return_value = mock_disk_io # Мокаем boot time import time mock_psutil.boot_time.return_value = time.time() - 86400 # 1 день назад return mock_psutil @pytest.fixture def mock_platform(): """Создает мок для platform""" mock_platform = Mock() mock_platform.system.return_value = 'Linux' return mock_platform @pytest.fixture def mock_subprocess(): """Создает мок для subprocess""" mock_subprocess = Mock() # Мокаем успешный результат diskutil mock_result = Mock() mock_result.returncode = 0 mock_result.stdout = """ Container Total Space: 500.0 GB Container Free Space: 400.0 GB """ mock_subprocess.run.return_value = mock_result return mock_subprocess @pytest.fixture def mock_os(): """Создает мок для os""" mock_os = Mock() mock_os.getenv.side_effect = lambda key, default=None: { 'THRESHOLD': '80.0', 'RECOVERY_THRESHOLD': '75.0' }.get(key, default) # Мокаем uname mock_uname = Mock() mock_uname.nodename = "test-host" mock_os.uname.return_value = mock_uname return mock_os @pytest.fixture def prometheus_config_sample(): """Создает пример конфигурации Prometheus для тестов""" return { 'global': { 'scrape_interval': '15s', 'evaluation_interval': '15s' }, 'rule_files': [ '# - "first_rules.yml"', '# - "second_rules.yml"' ], 'scrape_configs': [ { 'job_name': 'prometheus', 'static_configs': [ { 'targets': ['localhost:9090'] } ] }, { 'job_name': 'infrastructure', 'static_configs': [ { 'targets': ['host.docker.internal:9091'] } ], 'metrics_path': '/metrics', 'scrape_interval': '30s', 'scrape_timeout': '10s', 'honor_labels': True }, { 'job_name': 'telegram-helper-bot', 'static_configs': [ { 'targets': ['bots_telegram_bot:8080'], 'labels': { 'bot_name': 'telegram-helper-bot', 'environment': 'production', 'service': 'telegram-bot' } } ], 'metrics_path': '/metrics', 'scrape_interval': '15s', 'scrape_timeout': '10s', 'honor_labels': True } ], 'alerting': { 'alertmanagers': [ { 'static_configs': [ { 'targets': [ '# - alertmanager:9093' ] } ] } ] } } @pytest.fixture def mock_aiohttp(): """Создает мок для aiohttp""" mock_aiohttp = Mock() # Мокаем web.Application mock_app = Mock() mock_aiohttp.web.Application.return_value = mock_app # Мокаем web.Response mock_response = Mock() mock_response.status = 200 mock_response.content_type = 'text/plain' mock_response.text = 'Test response' mock_aiohttp.web.Response.return_value = mock_response return mock_aiohttp @pytest.fixture def mock_request(): """Создает мок для HTTP запроса""" request = Mock() request.method = 'GET' request.path = '/metrics' request.headers = {} return request @pytest.fixture def test_environment(): """Создает тестовое окружение""" return { 'os_type': 'ubuntu', 'threshold': 80.0, 'recovery_threshold': 75.0, 'host': '127.0.0.1', 'port': 9091 } # Маркеры для категоризации тестов def pytest_configure(config): """Настройка маркеров pytest""" config.addinivalue_line( "markers", "asyncio: mark test as async" ) config.addinivalue_line( "markers", "slow: mark test as slow" ) config.addinivalue_line( "markers", "integration: mark test as integration test" ) config.addinivalue_line( "markers", "unit: mark test as unit test" ) config.addinivalue_line( "markers", "prometheus: mark test as prometheus related" ) config.addinivalue_line( "markers", "metrics: mark test as metrics related" ) # Автоматическая маркировка тестов def pytest_collection_modifyitems(config, items): """Автоматически маркирует тесты по их расположению""" for item in items: # Маркируем асинхронные тесты if "async" in item.name or "Async" in item.name: item.add_marker(pytest.mark.asyncio) # Маркируем интеграционные тесты if "integration" in item.name.lower() or "Integration" in str(item.cls): item.add_marker(pytest.mark.integration) # Маркируем unit тесты if "unit" in item.name.lower() or "Unit" in str(item.cls): item.add_marker(pytest.mark.unit) # Маркируем медленные тесты if "slow" in item.name.lower() or "Slow" in str(item.cls): item.add_marker(pytest.mark.slow) # Маркируем тесты Prometheus if "prometheus" in item.name.lower() or "Prometheus" in str(item.cls): item.add_marker(pytest.mark.prometheus) # Маркируем тесты метрик if "metrics" in item.name.lower() or "Metrics" in str(item.cls): item.add_marker(pytest.mark.metrics)