231 lines
11 KiB
Python
231 lines
11 KiB
Python
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__])
|
||
|