Andrey
dd8b1c02a4
- Upgraded Python version in Dockerfile from 3.9 to 3.11.9 for enhanced performance and security. - Adjusted paths in Dockerfile to reflect the new Python version. - Modified test commands in Makefile to activate the virtual environment before running tests, ensuring proper dependency management.
318 lines
15 KiB
Python
318 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Тесты для конфигурации Prometheus
|
||
"""
|
||
|
||
import pytest
|
||
import yaml
|
||
import sys
|
||
import os
|
||
from pathlib import Path
|
||
|
||
|
||
|
||
class TestPrometheusConfig:
|
||
"""Тесты для конфигурации Prometheus"""
|
||
|
||
@pytest.fixture
|
||
def prometheus_config_path(self):
|
||
"""Путь к файлу конфигурации Prometheus"""
|
||
return Path(__file__).parent.parent.parent / 'infra' / 'prometheus' / 'prometheus.yml'
|
||
|
||
@pytest.fixture
|
||
def prometheus_config(self, prometheus_config_path):
|
||
"""Загруженная конфигурация Prometheus"""
|
||
if not prometheus_config_path.exists():
|
||
pytest.skip(f"Prometheus config file not found: {prometheus_config_path}")
|
||
|
||
with open(prometheus_config_path, 'r', encoding='utf-8') as f:
|
||
return yaml.safe_load(f)
|
||
|
||
def test_config_file_exists(self, prometheus_config_path):
|
||
"""Тест существования файла конфигурации"""
|
||
assert prometheus_config_path.exists(), f"Prometheus config file not found: {prometheus_config_path}"
|
||
|
||
def test_config_is_valid_yaml(self, prometheus_config):
|
||
"""Тест валидности YAML конфигурации"""
|
||
assert isinstance(prometheus_config, dict), "Config should be a valid YAML dictionary"
|
||
|
||
def test_global_section(self, prometheus_config):
|
||
"""Тест глобальной секции конфигурации"""
|
||
assert 'global' in prometheus_config, "Config should have global section"
|
||
|
||
global_config = prometheus_config['global']
|
||
assert 'scrape_interval' in global_config, "Global section should have scrape_interval"
|
||
assert 'evaluation_interval' in global_config, "Global section should have evaluation_interval"
|
||
|
||
# Проверяем значения интервалов
|
||
assert global_config['scrape_interval'] == '15s', "Default scrape_interval should be 15s"
|
||
assert global_config['evaluation_interval'] == '15s', "Default evaluation_interval should be 15s"
|
||
|
||
def test_scrape_configs_section(self, prometheus_config):
|
||
"""Тест секции scrape_configs"""
|
||
assert 'scrape_configs' in prometheus_config, "Config should have scrape_configs section"
|
||
|
||
scrape_configs = prometheus_config['scrape_configs']
|
||
assert isinstance(scrape_configs, list), "scrape_configs should be a list"
|
||
assert len(scrape_configs) >= 1, "Should have at least one scrape config"
|
||
|
||
def test_prometheus_job(self, prometheus_config):
|
||
"""Тест job для самого Prometheus"""
|
||
scrape_configs = prometheus_config['scrape_configs']
|
||
|
||
# Ищем job для prometheus
|
||
prometheus_job = None
|
||
for job in scrape_configs:
|
||
if job.get('job_name') == 'prometheus':
|
||
prometheus_job = job
|
||
break
|
||
|
||
assert prometheus_job is not None, "Should have prometheus job"
|
||
assert 'static_configs' in prometheus_job, "Prometheus job should have static_configs"
|
||
|
||
static_configs = prometheus_job['static_configs']
|
||
assert isinstance(static_configs, list), "static_configs should be a list"
|
||
assert len(static_configs) > 0, "Should have at least one static config"
|
||
|
||
# Проверяем targets
|
||
targets = static_configs[0].get('targets', [])
|
||
assert 'localhost:9090' in targets, "Prometheus should scrape localhost:9090"
|
||
|
||
|
||
def test_telegram_bot_job(self, prometheus_config):
|
||
"""Тест job для telegram-helper-bot"""
|
||
scrape_configs = prometheus_config['scrape_configs']
|
||
|
||
# Ищем job для telegram-helper-bot
|
||
bot_job = None
|
||
for job in scrape_configs:
|
||
if job.get('job_name') == 'telegram-helper-bot':
|
||
bot_job = job
|
||
break
|
||
|
||
assert bot_job is not None, "Should have telegram-helper-bot job"
|
||
|
||
# Проверяем основные параметры
|
||
assert 'static_configs' in bot_job, "Bot job should have static_configs"
|
||
assert 'metrics_path' in bot_job, "Bot job should have metrics_path"
|
||
assert 'scrape_interval' in bot_job, "Bot job should have scrape_interval"
|
||
assert 'scrape_timeout' in bot_job, "Bot job should have scrape_timeout"
|
||
assert 'honor_labels' in bot_job, "Bot job should have honor_labels"
|
||
|
||
# Проверяем значения
|
||
assert bot_job['metrics_path'] == '/metrics', "Metrics path should be /metrics"
|
||
assert bot_job['scrape_interval'] == '15s', "Scrape interval should be 15s"
|
||
assert bot_job['scrape_timeout'] == '10s', "Scrape timeout should be 10s"
|
||
assert bot_job['honor_labels'] is True, "honor_labels should be True"
|
||
|
||
# Проверяем static_configs
|
||
static_configs = bot_job['static_configs']
|
||
assert len(static_configs) > 0, "Should have at least one static config"
|
||
|
||
# Проверяем targets
|
||
targets = static_configs[0].get('targets', [])
|
||
assert 'bots_telegram_bot:8080' in targets, "Should scrape bots_telegram_bot:8080"
|
||
|
||
# Проверяем labels
|
||
labels = static_configs[0].get('labels', {})
|
||
expected_labels = {
|
||
'bot_name': 'telegram-helper-bot',
|
||
'environment': 'production',
|
||
'service': 'telegram-bot'
|
||
}
|
||
|
||
for key, value in expected_labels.items():
|
||
assert key in labels, f"Should have label {key}"
|
||
assert labels[key] == value, f"Label {key} should be {value}"
|
||
|
||
def test_alerting_section(self, prometheus_config):
|
||
"""Тест секции alerting"""
|
||
assert 'alerting' in prometheus_config, "Config should have alerting section"
|
||
|
||
alerting_config = prometheus_config['alerting']
|
||
assert 'alertmanagers' in alerting_config, "Alerting section should have alertmanagers"
|
||
|
||
alertmanagers = alerting_config['alertmanagers']
|
||
assert isinstance(alertmanagers, list), "alertmanagers should be a list"
|
||
|
||
# Проверяем, что alertmanager настроен правильно
|
||
if len(alertmanagers) > 0:
|
||
for am in alertmanagers:
|
||
if 'static_configs' in am:
|
||
static_configs = am['static_configs']
|
||
assert isinstance(static_configs, list), "static_configs should be a list"
|
||
for sc in static_configs:
|
||
if 'targets' in sc:
|
||
targets = sc['targets']
|
||
# targets может быть None если все строки закомментированы
|
||
if targets is not None:
|
||
assert isinstance(targets, list), "targets should be a list"
|
||
# Проверяем, что targets не пустые и имеют правильный формат
|
||
for target in targets:
|
||
assert isinstance(target, str), f"Target should be a string: {target}"
|
||
# Если target не закомментирован, проверяем формат
|
||
if not target.startswith('#'):
|
||
assert ':' in target, f"Target should have port: {target}"
|
||
|
||
def test_rule_files_section(self, prometheus_config):
|
||
"""Тест секции rule_files"""
|
||
assert 'rule_files' in prometheus_config, "Config should have rule_files section"
|
||
|
||
rule_files = prometheus_config['rule_files']
|
||
# rule_files может быть None если все строки закомментированы
|
||
if rule_files is not None:
|
||
assert isinstance(rule_files, list), "rule_files should be a list"
|
||
|
||
# Проверяем, что rule files имеют правильный формат
|
||
for rule_file in rule_files:
|
||
assert isinstance(rule_file, str), f"Rule file should be a string: {rule_file}"
|
||
# Если rule file не закомментирован, проверяем, что это валидный путь
|
||
if not rule_file.startswith('#'):
|
||
assert rule_file.endswith('.yml') or rule_file.endswith('.yaml'), \
|
||
f"Rule file should have .yml or .yaml extension: {rule_file}"
|
||
|
||
def test_config_structure_consistency(self, prometheus_config):
|
||
"""Тест консистентности структуры конфигурации"""
|
||
# Проверяем, что все job'ы имеют одинаковую структуру
|
||
scrape_configs = prometheus_config['scrape_configs']
|
||
|
||
required_fields = ['job_name', 'static_configs']
|
||
optional_fields = ['metrics_path', 'scrape_interval', 'scrape_timeout', 'honor_labels']
|
||
|
||
for job in scrape_configs:
|
||
# Проверяем обязательные поля
|
||
for field in required_fields:
|
||
assert field in job, f"Job {job.get('job_name', 'unknown')} missing required field: {field}"
|
||
|
||
# Проверяем, что static_configs содержит targets
|
||
static_configs = job['static_configs']
|
||
assert isinstance(static_configs, list), f"Job {job.get('job_name', 'unknown')} static_configs should be list"
|
||
|
||
for static_config in static_configs:
|
||
assert 'targets' in static_config, f"Static config should have targets"
|
||
targets = static_config['targets']
|
||
assert isinstance(targets, list), "Targets should be a list"
|
||
assert len(targets) > 0, "Targets should not be empty"
|
||
|
||
def test_port_configurations(self, prometheus_config):
|
||
"""Тест конфигурации портов"""
|
||
scrape_configs = prometheus_config['scrape_configs']
|
||
|
||
# Проверяем, что порты корректно настроены
|
||
for job in scrape_configs:
|
||
static_configs = job['static_configs']
|
||
for static_config in static_configs:
|
||
targets = static_config['targets']
|
||
for target in targets:
|
||
if ':' in target:
|
||
host, port = target.split(':', 1)
|
||
# Проверяем, что порт это число
|
||
try:
|
||
port_num = int(port)
|
||
assert 1 <= port_num <= 65535, f"Port {port_num} out of range"
|
||
except ValueError:
|
||
# Это может быть Docker service name без порта
|
||
pass
|
||
|
||
def test_environment_labels(self, prometheus_config):
|
||
"""Тест labels окружения"""
|
||
scrape_configs = prometheus_config['scrape_configs']
|
||
|
||
# Проверяем, что production окружение правильно помечено
|
||
for job in scrape_configs:
|
||
if job.get('job_name') == 'telegram-helper-bot':
|
||
static_configs = job['static_configs']
|
||
for static_config in static_configs:
|
||
labels = static_config.get('labels', {})
|
||
if 'environment' in labels:
|
||
assert labels['environment'] == 'production', "Environment should be production"
|
||
|
||
def test_metrics_path_consistency(self, prometheus_config):
|
||
"""Тест консистентности paths для метрик"""
|
||
scrape_configs = prometheus_config['scrape_configs']
|
||
|
||
# Проверяем, что все job'ы используют /metrics
|
||
for job in scrape_configs:
|
||
if 'metrics_path' in job:
|
||
assert job['metrics_path'] == '/metrics', f"Job {job.get('job_name', 'unknown')} should use /metrics path"
|
||
|
||
|
||
class TestPrometheusConfigValidation:
|
||
"""Тесты валидации конфигурации Prometheus"""
|
||
|
||
@pytest.fixture
|
||
def sample_valid_config(self):
|
||
"""Пример валидной конфигурации"""
|
||
return {
|
||
'global': {
|
||
'scrape_interval': '15s',
|
||
'evaluation_interval': '15s'
|
||
},
|
||
'scrape_configs': [
|
||
{
|
||
'job_name': 'test',
|
||
'static_configs': [
|
||
{
|
||
'targets': ['localhost:9090']
|
||
}
|
||
]
|
||
}
|
||
]
|
||
}
|
||
|
||
def test_minimal_valid_config(self, sample_valid_config):
|
||
"""Тест минимальной валидной конфигурации"""
|
||
# Проверяем, что конфигурация содержит все необходимые поля
|
||
assert 'global' in sample_valid_config
|
||
assert 'scrape_configs' in sample_valid_config
|
||
|
||
global_config = sample_valid_config['global']
|
||
assert 'scrape_interval' in global_config
|
||
assert 'evaluation_interval' in global_config
|
||
|
||
scrape_configs = sample_valid_config['scrape_configs']
|
||
assert len(scrape_configs) > 0
|
||
|
||
for job in scrape_configs:
|
||
assert 'job_name' in job
|
||
assert 'static_configs' in job
|
||
|
||
static_configs = job['static_configs']
|
||
assert len(static_configs) > 0
|
||
|
||
for static_config in static_configs:
|
||
assert 'targets' in static_config
|
||
targets = static_config['targets']
|
||
assert len(targets) > 0
|
||
|
||
def test_config_without_required_fields(self):
|
||
"""Тест конфигурации без обязательных полей"""
|
||
# Конфигурация без global секции
|
||
config_without_global = {
|
||
'scrape_configs': []
|
||
}
|
||
|
||
# Конфигурация без scrape_configs
|
||
config_without_scrape = {
|
||
'global': {
|
||
'scrape_interval': '15s'
|
||
}
|
||
}
|
||
|
||
# Конфигурация с пустыми scrape_configs
|
||
config_empty_scrape = {
|
||
'global': {
|
||
'scrape_interval': '15s'
|
||
},
|
||
'scrape_configs': []
|
||
}
|
||
|
||
# Все эти конфигурации должны быть невалидными
|
||
assert 'global' not in config_without_global
|
||
assert 'scrape_configs' not in config_without_scrape
|
||
assert len(config_empty_scrape['scrape_configs']) == 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v"])
|