some fix
This commit is contained in:
@@ -3,314 +3,351 @@
|
||||
Тесты для конфигурации Prometheus
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
import sys
|
||||
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'
|
||||
|
||||
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:
|
||||
|
||||
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}"
|
||||
|
||||
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"
|
||||
|
||||
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" 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"
|
||||
|
||||
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 (
|
||||
"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']
|
||||
|
||||
scrape_configs = prometheus_config["scrape_configs"]
|
||||
|
||||
# Ищем job для prometheus
|
||||
prometheus_job = None
|
||||
for job in scrape_configs:
|
||||
if job.get('job_name') == 'prometheus':
|
||||
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 (
|
||||
"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"
|
||||
|
||||
|
||||
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']
|
||||
|
||||
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':
|
||||
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 "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"
|
||||
|
||||
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']
|
||||
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"
|
||||
|
||||
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', {})
|
||||
labels = static_configs[0].get("labels", {})
|
||||
expected_labels = {
|
||||
'bot_name': 'telegram-helper-bot',
|
||||
'environment': 'production',
|
||||
'service': 'telegram-bot'
|
||||
"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 "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"
|
||||
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']
|
||||
if "targets" in sc:
|
||||
targets = sc["targets"]
|
||||
# targets может быть None если все строки закомментированы
|
||||
if targets is not None:
|
||||
assert isinstance(targets, list), "targets should be a list"
|
||||
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}"
|
||||
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}"
|
||||
|
||||
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']
|
||||
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}"
|
||||
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}"
|
||||
|
||||
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']
|
||||
|
||||
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}"
|
||||
|
||||
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"
|
||||
|
||||
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 "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']
|
||||
|
||||
scrape_configs = prometheus_config["scrape_configs"]
|
||||
|
||||
# Проверяем, что порты корректно настроены
|
||||
for job in scrape_configs:
|
||||
static_configs = job['static_configs']
|
||||
static_configs = job["static_configs"]
|
||||
for static_config in static_configs:
|
||||
targets = static_config['targets']
|
||||
targets = static_config["targets"]
|
||||
for target in targets:
|
||||
if ':' in target:
|
||||
host, port = target.split(':', 1)
|
||||
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"
|
||||
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']
|
||||
|
||||
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']
|
||||
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"
|
||||
|
||||
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']
|
||||
|
||||
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"
|
||||
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': [
|
||||
"global": {"scrape_interval": "15s", "evaluation_interval": "15s"},
|
||||
"scrape_configs": [
|
||||
{
|
||||
'job_name': 'test',
|
||||
'static_configs': [
|
||||
{
|
||||
'targets': ['localhost:9090']
|
||||
}
|
||||
]
|
||||
"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 "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 "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 "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': []
|
||||
}
|
||||
|
||||
config_without_global = {"scrape_configs": []}
|
||||
|
||||
# Конфигурация без scrape_configs
|
||||
config_without_scrape = {
|
||||
'global': {
|
||||
'scrape_interval': '15s'
|
||||
}
|
||||
}
|
||||
|
||||
config_without_scrape = {"global": {"scrape_interval": "15s"}}
|
||||
|
||||
# Конфигурация с пустыми scrape_configs
|
||||
config_empty_scrape = {
|
||||
'global': {
|
||||
'scrape_interval': '15s'
|
||||
},
|
||||
'scrape_configs': []
|
||||
"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
|
||||
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__":
|
||||
|
||||
@@ -3,28 +3,36 @@
|
||||
Тест конфигурации pytest
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import os
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def test_pytest_config_loaded():
|
||||
"""Проверяем, что конфигурация pytest загружена"""
|
||||
# Проверяем, что мы находимся в корневой директории проекта
|
||||
assert os.path.exists('pytest.ini'), "pytest.ini должен существовать в корне проекта"
|
||||
|
||||
assert os.path.exists(
|
||||
"pytest.ini"
|
||||
), "pytest.ini должен существовать в корне проекта"
|
||||
|
||||
# Проверяем, что директория tests существует
|
||||
assert os.path.exists('tests'), "Директория tests должна существовать"
|
||||
assert os.path.exists('tests/infra'), "Директория tests/infra должна существовать"
|
||||
assert os.path.exists('tests/bot'), "Директория tests/bot должна существовать"
|
||||
assert os.path.exists("tests"), "Директория tests должна существовать"
|
||||
assert os.path.exists("tests/infra"), "Директория tests/infra должна существовать"
|
||||
assert os.path.exists("tests/bot"), "Директория tests/bot должна существовать"
|
||||
|
||||
|
||||
def test_test_structure():
|
||||
"""Проверяем структуру тестов"""
|
||||
# Проверяем наличие __init__.py файлов
|
||||
assert os.path.exists('tests/__init__.py'), "tests/__init__.py должен существовать"
|
||||
assert os.path.exists('tests/infra/__init__.py'), "tests/infra/__init__.py должен существовать"
|
||||
assert os.path.exists('tests/bot/__init__.py'), "tests/bot/__init__.py должен существовать"
|
||||
|
||||
assert os.path.exists("tests/__init__.py"), "tests/__init__.py должен существовать"
|
||||
assert os.path.exists(
|
||||
"tests/infra/__init__.py"
|
||||
), "tests/infra/__init__.py должен существовать"
|
||||
assert os.path.exists(
|
||||
"tests/bot/__init__.py"
|
||||
), "tests/bot/__init__.py должен существовать"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
|
||||
Reference in New Issue
Block a user