#!/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'] for sc in static_configs: if 'targets' in sc: targets = sc['targets'] # targets может быть None если все строки закомментированы if targets is not None: # Проверяем, что все targets закомментированы for target in targets: assert target.startswith('#'), f"Alertmanager target should be commented: {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 rule_file.startswith('#'), f"Rule file should be commented: {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"])