import pytest import time from unittest.mock import Mock, patch import sys import os # Добавляем путь к модулю для импорта sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'infra', 'monitoring')) from metrics_collector import MetricsCollector class TestAlertDelays: """Тесты для механизма задержки алертов""" def setup_method(self): """Настройка перед каждым тестом""" # Мокаем переменные окружения with patch.dict(os.environ, { 'CPU_ALERT_DELAY': '5', # 5 секунд для быстрого тестирования 'RAM_ALERT_DELAY': '7', # 7 секунд для быстрого тестирования 'DISK_ALERT_DELAY': '10' # 10 секунд для быстрого тестирования }): self.collector = MetricsCollector() def test_alert_delays_initialization(self): """Тест инициализации задержек алертов""" assert self.collector.alert_delays['cpu'] == 5 assert self.collector.alert_delays['ram'] == 7 assert self.collector.alert_delays['disk'] == 10 # Проверяем, что время начала превышения инициализировано как None assert self.collector.alert_start_times['cpu'] is None assert self.collector.alert_start_times['ram'] is None assert self.collector.alert_start_times['disk'] is None def test_cpu_alert_delay_logic(self): """Тест логики задержки алерта CPU""" # Симулируем превышение порога CPU system_info = { 'cpu_percent': 85.0, # Выше порога 80% 'ram_percent': 70.0, # Нормально 'disk_percent': 75.0, # Нормально 'load_avg_1m': 2.5, 'ram_used': 8.0, 'ram_total': 16.0, 'disk_free': 25.0 } # Первая проверка - должно начать отсчет задержки alerts, recoveries = self.collector.check_alerts(system_info) assert len(alerts) == 0 # Алерт еще не отправлен assert self.collector.alert_start_times['cpu'] is not None # Время начала установлено # Проверяем, что состояние алерта не изменилось assert not self.collector.alert_states['cpu'] # Симулируем время, прошедшее с начала превышения # Устанавливаем время начала в прошлое (больше задержки) self.collector.alert_start_times['cpu'] = time.time() - 6 # 6 секунд назад # Теперь алерт должен сработать alerts, recoveries = self.collector.check_alerts(system_info) assert len(alerts) == 1 # Алерт отправлен assert alerts[0][0] == 'cpu' # Тип алерта assert alerts[0][1] == 85.0 # Значение CPU assert self.collector.alert_states['cpu'] # Состояние алерта установлено def test_alert_reset_on_recovery(self): """Тест сброса алерта при восстановлении""" # Сначала превышаем порог и ждем задержку system_info_high = { 'cpu_percent': 85.0, 'ram_percent': 70.0, 'disk_percent': 75.0, 'load_avg_1m': 2.5, 'ram_used': 8.0, 'ram_total': 16.0, 'disk_free': 25.0 } # Устанавливаем время начала превышения в прошлое self.collector.alert_start_times['cpu'] = time.time() - 6 # Проверяем - алерт должен сработать alerts, recoveries = self.collector.check_alerts(system_info_high) assert len(alerts) == 1 # Алерт отправлен assert self.collector.alert_states['cpu'] # Состояние установлено # Теперь симулируем восстановление system_info_low = { 'cpu_percent': 70.0, # Ниже порога восстановления 75% 'ram_percent': 70.0, 'disk_percent': 75.0, 'load_avg_1m': 1.2, 'ram_used': 8.0, 'ram_total': 16.0, 'disk_free': 25.0 } alerts, recoveries = self.collector.check_alerts(system_info_low) assert len(recoveries) == 1 # Сообщение о восстановлении assert recoveries[0][0] == 'cpu' # Тип восстановления assert not self.collector.alert_states['cpu'] # Состояние сброшено assert self.collector.alert_start_times['cpu'] is None # Время сброшено def test_multiple_metrics_alert(self): """Тест алертов по нескольким метрикам одновременно""" system_info = { 'cpu_percent': 85.0, # Выше порога 'ram_percent': 85.0, # Выше порога 'disk_percent': 75.0, # Нормально 'load_avg_1m': 2.5, 'ram_used': 13.0, 'ram_total': 16.0, 'disk_free': 25.0 } # Устанавливаем время начала превышения для CPU и RAM в прошлое self.collector.alert_start_times['cpu'] = time.time() - 6 # Больше CPU_ALERT_DELAY (5 сек) self.collector.alert_start_times['ram'] = time.time() - 8 # Больше RAM_ALERT_DELAY (7 сек) alerts, recoveries = self.collector.check_alerts(system_info) assert len(alerts) == 2 # Два алерта: CPU и RAM # Проверяем типы алертов alert_types = [alert[0] for alert in alerts] assert 'cpu' in alert_types assert 'ram' in alert_types # Проверяем состояния assert self.collector.alert_states['cpu'] assert self.collector.alert_states['ram'] assert not self.collector.alert_states['disk'] def test_alert_delay_customization(self): """Тест настройки пользовательских задержек""" # Тестируем с другими значениями задержек with patch.dict(os.environ, { 'CPU_ALERT_DELAY': '2', 'RAM_ALERT_DELAY': '3', 'DISK_ALERT_DELAY': '4' }): collector = MetricsCollector() assert collector.alert_delays['cpu'] == 2 assert collector.alert_delays['ram'] == 3 assert collector.alert_delays['disk'] == 4 def test_no_false_alerts(self): """Тест отсутствия ложных алертов при кратковременных пиках""" system_info = { 'cpu_percent': 85.0, 'ram_percent': 70.0, 'disk_percent': 75.0, 'load_avg_1m': 2.5, 'ram_used': 8.0, 'ram_total': 16.0, 'disk_free': 25.0 } # Проверяем сразу после превышения порога alerts, recoveries = self.collector.check_alerts(system_info) assert len(alerts) == 0 # Алерт не должен сработать сразу # Проверяем, что время начала установлено assert self.collector.alert_start_times['cpu'] is not None # Проверяем через короткое время (до истечения задержки) # Устанавливаем время начала в прошлое, но меньше задержки self.collector.alert_start_times['cpu'] = time.time() - 2 # 2 секунды назад alerts, recoveries = self.collector.check_alerts(system_info) assert len(alerts) == 0 # Алерт все еще не должен сработать def test_alert_state_persistence(self): """Тест сохранения состояния алерта между проверками""" system_info = { 'cpu_percent': 85.0, 'ram_percent': 70.0, 'disk_percent': 75.0, 'load_avg_1m': 2.5, 'ram_used': 8.0, 'ram_total': 16.0, 'disk_free': 25.0 } # Первая проверка - начинаем отсчет alerts, recoveries = self.collector.check_alerts(system_info) assert len(alerts) == 0 initial_time = self.collector.alert_start_times['cpu'] assert initial_time is not None # Проверяем еще раз - время начала должно сохраниться alerts, recoveries = self.collector.check_alerts(system_info) assert len(alerts) == 0 assert self.collector.alert_start_times['cpu'] == initial_time # Время не изменилось def test_disk_alert_delay(self): """Тест задержки алерта для диска""" system_info = { 'cpu_percent': 70.0, 'ram_percent': 70.0, 'disk_percent': 85.0, # Выше порога 'load_avg_1m': 1.2, 'ram_used': 8.0, 'ram_total': 16.0, 'disk_free': 15.0 } # Первая проверка alerts, recoveries = self.collector.check_alerts(system_info) assert len(alerts) == 0 assert self.collector.alert_start_times['disk'] is not None # Устанавливаем время начала превышения в прошлое, но меньше задержки self.collector.alert_start_times['disk'] = time.time() - 5 # 5 секунд назад (меньше DISK_ALERT_DELAY) alerts, recoveries = self.collector.check_alerts(system_info) assert len(alerts) == 0 # Алерт не должен сработать # Устанавливаем время начала превышения в прошлое, больше задержки self.collector.alert_start_times['disk'] = time.time() - 11 # 11 секунд назад (больше DISK_ALERT_DELAY) alerts, recoveries = self.collector.check_alerts(system_info) assert len(alerts) == 1 # Алерт должен сработать assert alerts[0][0] == 'disk' if __name__ == '__main__': pytest.main([__file__])