#!/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"])