318 lines
9.3 KiB
Python
318 lines
9.3 KiB
Python
#!/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,
|
|
'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)
|