355 lines
15 KiB
Python
355 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Тесты для конфигурации Prometheus
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
import yaml
|
||
|
||
|
||
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"])
|