jaredandClaude Opus 4.8
48d4474c20
Lint / Python (flake8) (push) Successful in 1m13s
Security / Python Security (bandit) (push) Successful in 37s
Test / Python Tests (pytest) (push) Successful in 41s
Lint / Notify on failure (push) Has been skipped
Lint / Python (flake8) (pull_request) Successful in 30s
Security / Python Security (bandit) (pull_request) Successful in 1m14s
Test / Python Tests (pytest) (pull_request) Successful in 43s
Lint / Notify on failure (pull_request) Has been skipped
_check_cpu_usage took a single 1-second psutil sample per run, so a momentary spike raised a CPU ticket -- and because the ticket API reopens a closed ticket on a matching alert, the same CPU ticket flapped open/closed on every hourly run. Ticketing is now gated on the 15-minute load average normalized per core (a true "sustained" signal): a transient spike barely moves the 15-min load, so it no longer raises or reopens a ticket. The instantaneous percentage is still reported (dry-run summary + description) and the measured sustained load is shown in the ticket description; the title stays value-free so it doesn't churn each run. Falls back to the instantaneous sample only if os.getloadavg() is unavailable. Adds 5 regression tests. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
608 lines
27 KiB
Python
608 lines
27 KiB
Python
"""Tests for SystemHealthMonitor pure methods — no external processes or filesystem."""
|
|
import sys
|
|
import os
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
|
|
|
from unittest.mock import patch, MagicMock # noqa: E402
|
|
from hwmonDaemon import SystemHealthMonitor # noqa: E402
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def monitor():
|
|
"""Create a minimal monitor instance with all external side-effects patched out."""
|
|
with patch.object(SystemHealthMonitor, 'load_env_config'), \
|
|
patch.object(SystemHealthMonitor, '_check_tool_availability', return_value={}), \
|
|
patch('os.makedirs'):
|
|
return SystemHealthMonitor(dry_run=True)
|
|
|
|
|
|
# ── _format_bytes_human ──────────────────────────────────────────────────────
|
|
|
|
class TestFormatBytesHuman:
|
|
def test_bytes(self, monitor):
|
|
assert monitor._format_bytes_human(512) == '512.0 B'
|
|
|
|
def test_kilobytes(self, monitor):
|
|
assert monitor._format_bytes_human(1024) == '1.0 KB'
|
|
|
|
def test_megabytes(self, monitor):
|
|
assert monitor._format_bytes_human(1024 ** 2) == '1.0 MB'
|
|
|
|
def test_gigabytes(self, monitor):
|
|
assert monitor._format_bytes_human(1024 ** 3) == '1.0 GB'
|
|
|
|
def test_terabytes(self, monitor):
|
|
assert monitor._format_bytes_human(1024 ** 4) == '1.0 TB'
|
|
|
|
def test_fractional(self, monitor):
|
|
assert monitor._format_bytes_human(1536) == '1.5 KB'
|
|
|
|
def test_zero(self, monitor):
|
|
assert monitor._format_bytes_human(0) == '0.0 B'
|
|
|
|
|
|
# ── _parse_size ───────────────────────────────────────────────────────────────
|
|
|
|
class TestParseSize:
|
|
def test_gigabytes(self, monitor):
|
|
result = monitor._parse_size('15.7G')
|
|
assert abs(result - 15.7 * 1024**3) < 1
|
|
|
|
def test_terabytes(self, monitor):
|
|
result = monitor._parse_size('21.8T')
|
|
assert abs(result - 21.8 * 1024**4) < 1
|
|
|
|
def test_megabytes(self, monitor):
|
|
result = monitor._parse_size('512M')
|
|
assert result == 512 * 1024**2
|
|
|
|
def test_kilobytes(self, monitor):
|
|
result = monitor._parse_size('100K')
|
|
assert result == 100 * 1024
|
|
|
|
def test_bytes(self, monitor):
|
|
result = monitor._parse_size('100B')
|
|
assert result == 100
|
|
|
|
def test_invalid_returns_zero(self, monitor):
|
|
assert monitor._parse_size('notasize') == 0.0
|
|
|
|
def test_non_string_returns_zero(self, monitor):
|
|
assert monitor._parse_size(None) == 0.0
|
|
assert monitor._parse_size(42) == 0.0
|
|
|
|
|
|
# ── _parse_smart_value ────────────────────────────────────────────────────────
|
|
|
|
class TestParseSmartValue:
|
|
def test_plain_integer(self, monitor):
|
|
assert monitor._parse_smart_value('42') == 42
|
|
|
|
def test_temperature_with_celsius(self, monitor):
|
|
assert monitor._parse_smart_value('38 °C') == 38
|
|
|
|
def test_time_format(self, monitor):
|
|
assert monitor._parse_smart_value('15589h+17m+33.939s') == 15589
|
|
|
|
def test_hex_value(self, monitor):
|
|
assert monitor._parse_smart_value('0x0a') == 10
|
|
|
|
def test_invalid_returns_zero(self, monitor):
|
|
assert monitor._parse_smart_value('not_a_number') == 0
|
|
|
|
|
|
# ── _detect_manufacturer ──────────────────────────────────────────────────────
|
|
|
|
class TestDetectManufacturer:
|
|
def test_western_digital(self, monitor):
|
|
assert monitor._detect_manufacturer('WDC WD40EFRX') == 'Western Digital'
|
|
|
|
def test_hgst(self, monitor):
|
|
assert monitor._detect_manufacturer('HGST HUH728080ALE604') == 'Western Digital'
|
|
|
|
def test_seagate(self, monitor):
|
|
assert monitor._detect_manufacturer('ST4000DM004') == 'Seagate'
|
|
|
|
def test_samsung(self, monitor):
|
|
assert monitor._detect_manufacturer('Samsung SSD 870 EVO') == 'Samsung'
|
|
|
|
def test_intel(self, monitor):
|
|
assert monitor._detect_manufacturer('INTEL SSDSC2KB480G8') == 'Intel'
|
|
|
|
def test_micron(self, monitor):
|
|
assert monitor._detect_manufacturer('Crucial CT500MX500SSD1') == 'Micron'
|
|
|
|
def test_toshiba(self, monitor):
|
|
assert monitor._detect_manufacturer('TOSHIBA MG06ACA10TE') == 'Toshiba'
|
|
|
|
def test_unknown(self, monitor):
|
|
assert monitor._detect_manufacturer('GENERICDRIVE XYZ') == 'Unknown'
|
|
|
|
def test_empty_model(self, monitor):
|
|
assert monitor._detect_manufacturer('') == 'Unknown'
|
|
|
|
def test_none_model(self, monitor):
|
|
assert monitor._detect_manufacturer(None) == 'Unknown'
|
|
|
|
|
|
# ── _check_thermal_health ─────────────────────────────────────────────────────
|
|
|
|
class TestCheckThermalHealth:
|
|
def test_hdd_ok_temperature(self, monitor):
|
|
issues = monitor._check_thermal_health('sda', 45, 'HDD')
|
|
assert issues == []
|
|
|
|
def test_hdd_info_temperature(self, monitor):
|
|
issues = monitor._check_thermal_health('sda', 62, 'HDD')
|
|
assert len(issues) == 1
|
|
assert 'INFO' in issues[0]
|
|
|
|
def test_hdd_warning_temperature(self, monitor):
|
|
issues = monitor._check_thermal_health('sda', 66, 'HDD')
|
|
assert len(issues) == 1
|
|
assert 'WARNING' in issues[0]
|
|
|
|
def test_hdd_critical_temperature(self, monitor):
|
|
issues = monitor._check_thermal_health('sda', 76, 'HDD')
|
|
assert len(issues) == 1
|
|
assert 'CRITICAL' in issues[0]
|
|
|
|
def test_ssd_has_higher_warning_threshold(self, monitor):
|
|
# HDD warning=65°C, SSD warning=70°C; at 67°C:
|
|
# HDD → WARNING, SSD → INFO (above optimal_max=65 but below warning=70)
|
|
issues_hdd = monitor._check_thermal_health('sda', 67, 'HDD')
|
|
issues_ssd = monitor._check_thermal_health('sda', 67, 'SSD')
|
|
assert any('WARNING' in i for i in issues_hdd)
|
|
assert not any('WARNING' in i for i in issues_ssd)
|
|
assert any('INFO' in i for i in issues_ssd)
|
|
|
|
def test_none_temperature_returns_empty(self, monitor):
|
|
issues = monitor._check_thermal_health('sda', None, 'HDD')
|
|
assert issues == []
|
|
|
|
|
|
# ── _is_excluded_mount ────────────────────────────────────────────────────────
|
|
|
|
class TestIsExcludedMount:
|
|
def test_exact_excluded_mount(self, monitor):
|
|
assert monitor._is_excluded_mount('/media') is True
|
|
|
|
def test_pattern_excluded(self, monitor):
|
|
assert monitor._is_excluded_mount('/media/external') is True
|
|
|
|
def test_downloads_excluded(self, monitor):
|
|
assert monitor._is_excluded_mount('/mnt/data/downloads') is True
|
|
|
|
def test_normal_mount_not_excluded(self, monitor):
|
|
assert monitor._is_excluded_mount('/') is False
|
|
assert monitor._is_excluded_mount('/var') is False
|
|
assert monitor._is_excluded_mount('/mnt/ceph') is False
|
|
|
|
|
|
# ── _is_new_drive ─────────────────────────────────────────────────────────────
|
|
|
|
class TestIsNewDrive:
|
|
def test_brand_new_drive(self, monitor):
|
|
assert monitor._is_new_drive(0) is True
|
|
|
|
def test_one_hour_drive(self, monitor):
|
|
assert monitor._is_new_drive(1) is True
|
|
|
|
def test_under_threshold(self, monitor):
|
|
assert monitor._is_new_drive(719) is True
|
|
|
|
def test_at_threshold_is_not_new(self, monitor):
|
|
assert monitor._is_new_drive(720) is False
|
|
|
|
def test_old_drive(self, monitor):
|
|
assert monitor._is_new_drive(50000) is False
|
|
|
|
|
|
# ── _is_physical_disk ────────────────────────────────────────────────────────
|
|
|
|
class TestIsPhysicalDisk:
|
|
def test_real_sata_disk(self, monitor):
|
|
# /dev/sda should pass (no exclusion pattern matches)
|
|
# Note: _is_physical_disk also checks os.path.exists and reads sysfs,
|
|
# but the exclusion logic runs first and can return False early.
|
|
# We test the exclusion cases which are pure.
|
|
assert monitor._is_physical_disk('/dev/mapper/data') is False
|
|
|
|
def test_device_mapper_excluded(self, monitor):
|
|
assert monitor._is_physical_disk('/dev/dm-0') is False
|
|
|
|
def test_loop_device_excluded(self, monitor):
|
|
assert monitor._is_physical_disk('/dev/loop0') is False
|
|
|
|
def test_partition_excluded(self, monitor):
|
|
assert monitor._is_physical_disk('/dev/sda1') is False
|
|
|
|
def test_rbd_excluded(self, monitor):
|
|
assert monitor._is_physical_disk('/dev/rbd0') is False
|
|
|
|
|
|
# ── _get_manufacturer_profile ────────────────────────────────────────────────
|
|
|
|
class TestGetManufacturerProfile:
|
|
def test_seagate_model_matched(self, monitor):
|
|
profile = monitor._get_manufacturer_profile('ST4000DM004')
|
|
assert 'High_Fly_Writes' in profile['attributes']
|
|
|
|
def test_seagate_command_timeout_disabled(self, monitor):
|
|
profile = monitor._get_manufacturer_profile('ST4000DM004')
|
|
assert profile['attributes']['Command_Timeout']['monitor'] is False
|
|
|
|
def test_wd_model_matched(self, monitor):
|
|
profile = monitor._get_manufacturer_profile('WDC WD40EFRX')
|
|
assert profile['attributes']['Command_Timeout']['monitor'] is False
|
|
|
|
def test_samsung_model_matched(self, monitor):
|
|
profile = monitor._get_manufacturer_profile('Samsung SSD 870 EVO')
|
|
assert 'Program_Fail_Cnt_Total' in profile['attributes']
|
|
assert profile['attributes']['Program_Fail_Cnt_Total']['monitor'] is False
|
|
|
|
def test_samsung_erase_fail_chip_disabled(self, monitor):
|
|
profile = monitor._get_manufacturer_profile('Samsung SSD 870 EVO')
|
|
assert profile['attributes']['Erase_Fail_Count_Chip']['monitor'] is False
|
|
|
|
def test_toshiba_mg08_model_matched_by_prefix(self, monitor):
|
|
# MG08 prefix — model string without "TOSHIBA" word
|
|
profile = monitor._get_manufacturer_profile('MG08ACP16TE')
|
|
assert profile['attributes']['Command_Timeout']['monitor'] is True
|
|
assert profile['attributes']['Command_Timeout']['warning_threshold'] == 1000
|
|
|
|
def test_toshiba_command_timeout_raised_threshold(self, monitor):
|
|
profile = monitor._get_manufacturer_profile('TOSHIBA MG06ACA10TE')
|
|
assert profile['attributes']['Command_Timeout']['critical_threshold'] == 5000
|
|
|
|
def test_oos_model_matched(self, monitor):
|
|
profile = monitor._get_manufacturer_profile('OOS14000G')
|
|
assert profile['attributes']['Command_Timeout']['monitor'] is False
|
|
assert profile['attributes']['Seek_Error_Rate']['monitor'] is False
|
|
|
|
def test_ridata_firmware_match(self, monitor):
|
|
profile = monitor._get_manufacturer_profile('SSD 512GB', firmware='HT3618B7')
|
|
assert profile['attributes']['Erase_Fail_Count_Chip']['monitor'] is False
|
|
|
|
def test_unknown_model_returns_generic(self, monitor):
|
|
profile = monitor._get_manufacturer_profile('GENERICDRIVE XYZ')
|
|
assert profile is monitor.MANUFACTURER_SMART_PROFILES['Generic']
|
|
|
|
|
|
# ── _should_monitor_attribute ────────────────────────────────────────────────
|
|
|
|
class TestShouldMonitorAttribute:
|
|
def test_disabled_attribute_returns_false(self, monitor):
|
|
seagate = monitor._get_manufacturer_profile('ST4000DM004')
|
|
assert monitor._should_monitor_attribute('Command_Timeout', seagate) is False
|
|
|
|
def test_enabled_attribute_returns_true(self, monitor):
|
|
seagate = monitor._get_manufacturer_profile('ST4000DM004')
|
|
assert monitor._should_monitor_attribute('High_Fly_Writes', seagate) is True
|
|
|
|
def test_unknown_attribute_defaults_to_true(self, monitor):
|
|
seagate = monitor._get_manufacturer_profile('ST4000DM004')
|
|
assert monitor._should_monitor_attribute('Some_Unknown_Attr', seagate) is True
|
|
|
|
def test_none_profile_defaults_to_true(self, monitor):
|
|
assert monitor._should_monitor_attribute('Anything', None) is True
|
|
|
|
def test_samsung_program_fail_total_disabled(self, monitor):
|
|
samsung = monitor._get_manufacturer_profile('Samsung SSD 870 EVO')
|
|
assert monitor._should_monitor_attribute('Program_Fail_Cnt_Total', samsung) is False
|
|
|
|
|
|
# ── _get_attribute_thresholds ────────────────────────────────────────────────
|
|
|
|
class TestGetAttributeThresholds:
|
|
def test_seagate_high_fly_writes_thresholds(self, monitor):
|
|
seagate = monitor._get_manufacturer_profile('ST4000DM004')
|
|
t = monitor._get_attribute_thresholds('High_Fly_Writes', seagate)
|
|
assert t['warning'] == 100
|
|
assert t['critical'] == 500
|
|
|
|
def test_toshiba_command_timeout_thresholds(self, monitor):
|
|
toshiba = monitor._get_manufacturer_profile('MG08ACP16TE')
|
|
t = monitor._get_attribute_thresholds('Command_Timeout', toshiba)
|
|
assert t['warning'] == 1000
|
|
assert t['critical'] == 5000
|
|
|
|
def test_base_threshold_reallocated_sector(self, monitor):
|
|
generic = monitor.MANUFACTURER_SMART_PROFILES['Generic']
|
|
t = monitor._get_attribute_thresholds('Reallocated_Sector_Ct', generic)
|
|
assert t['warning'] == 5
|
|
assert t['critical'] == 10
|
|
|
|
def test_base_threshold_high_fly_writes_raised(self, monitor):
|
|
# Default (non-Seagate) High_Fly_Writes threshold is now 100/500
|
|
generic = monitor.MANUFACTURER_SMART_PROFILES['Generic']
|
|
t = monitor._get_attribute_thresholds('High_Fly_Writes', generic)
|
|
assert t['warning'] == 100
|
|
assert t['critical'] == 500
|
|
|
|
def test_unknown_attribute_returns_none(self, monitor):
|
|
generic = monitor.MANUFACTURER_SMART_PROFILES['Generic']
|
|
assert monitor._get_attribute_thresholds('Made_Up_Attribute', generic) is None
|
|
|
|
def test_behavior_defaults_to_countup(self, monitor):
|
|
generic = monitor.MANUFACTURER_SMART_PROFILES['Generic']
|
|
t = monitor._get_attribute_thresholds('Reallocated_Sector_Ct', generic)
|
|
assert t['behavior'] == 'countup'
|
|
|
|
|
|
# ── _get_issue_type ───────────────────────────────────────────────────────────
|
|
|
|
class TestGetIssueType:
|
|
def test_smart_issue(self, monitor):
|
|
assert monitor._get_issue_type('SMART attribute warning on /dev/sda') == 'SMART Health Issue'
|
|
|
|
def test_drive_issue(self, monitor):
|
|
assert monitor._get_issue_type('Drive /dev/sdb has reallocated sectors') == 'Storage Issue'
|
|
|
|
def test_ceph_issue(self, monitor):
|
|
assert monitor._get_issue_type('Ceph cluster is HEALTH_WARN') == 'Ceph Cluster Issue'
|
|
|
|
def test_ecc_issue(self, monitor):
|
|
assert monitor._get_issue_type('ECC memory errors detected') == 'Memory Issue'
|
|
|
|
def test_cpu_issue(self, monitor):
|
|
assert monitor._get_issue_type('CPU usage at 95%') == 'Performance Issue'
|
|
|
|
def test_network_issue(self, monitor):
|
|
assert monitor._get_issue_type('Network interface eth0 down') == 'Network Issue'
|
|
|
|
def test_lxc_issue(self, monitor):
|
|
assert monitor._get_issue_type('LXC container storage usage at 90%') == 'Container Storage Issue'
|
|
|
|
def test_unknown_defaults_to_hardware(self, monitor):
|
|
assert monitor._get_issue_type('Something completely unknown') == 'Hardware Issue'
|
|
|
|
|
|
# ── _get_impact_level ─────────────────────────────────────────────────────────
|
|
|
|
class TestGetImpactLevel:
|
|
def test_critical_issue(self, monitor):
|
|
level = monitor._get_impact_level('CRITICAL: drive failure imminent')
|
|
assert '[CRIT]' in level
|
|
|
|
def test_unhealthy_is_critical(self, monitor):
|
|
level = monitor._get_impact_level('Ceph is UNHEALTHY')
|
|
assert '[CRIT]' in level
|
|
|
|
def test_warning_issue(self, monitor):
|
|
level = monitor._get_impact_level('WARNING: temperature elevated')
|
|
assert '[WARN]' in level
|
|
|
|
def test_storage_usage_is_warn_not_crit(self, monitor):
|
|
# "STORAGE USAGE" keyword takes priority over "CRITICAL" substring check
|
|
level = monitor._get_impact_level('CRITICAL storage usage at 95%')
|
|
assert '[WARN]' in level
|
|
|
|
def test_cpu_usage_is_warn(self, monitor):
|
|
level = monitor._get_impact_level('CPU usage at 80% threshold exceeded')
|
|
assert '[WARN]' in level
|
|
|
|
def test_low_priority(self, monitor):
|
|
level = monitor._get_impact_level('Informational: drive age notification')
|
|
assert '[LOW]' in level
|
|
|
|
def test_health_err_is_critical(self, monitor):
|
|
level = monitor._get_impact_level('Ceph status: HEALTH_ERR')
|
|
assert '[CRIT]' in level
|
|
|
|
def test_down_is_warning(self, monitor):
|
|
level = monitor._get_impact_level('OSD.3 is DOWN')
|
|
assert '[WARN]' in level
|
|
|
|
|
|
# ── _categorize_issue ─────────────────────────────────────────────────────────
|
|
|
|
class TestCategorizeIssue:
|
|
def test_smart_critical_is_hardware_issue(self, monitor):
|
|
cat, ttype, _, _ = monitor._categorize_issue('SMART critical error on /dev/sda')
|
|
assert cat == monitor.TICKET_CATEGORIES['HARDWARE']
|
|
assert ttype == monitor.TICKET_TYPES['ISSUE']
|
|
|
|
def test_smart_warning_is_hardware_problem(self, monitor):
|
|
cat, ttype, _, _ = monitor._categorize_issue('SMART warning: High_Fly_Writes elevated')
|
|
assert cat == monitor.TICKET_CATEGORIES['HARDWARE']
|
|
assert ttype == monitor.TICKET_TYPES['PROBLEM']
|
|
|
|
def test_lxc_critical_is_software_issue(self, monitor):
|
|
cat, ttype, _, _ = monitor._categorize_issue('LXC container storage critical')
|
|
assert cat == monitor.TICKET_CATEGORIES['SOFTWARE']
|
|
assert ttype == monitor.TICKET_TYPES['ISSUE']
|
|
|
|
def test_temperature_is_hardware(self, monitor):
|
|
cat, _, _, _ = monitor._categorize_issue('temperature warning on /dev/sdb')
|
|
assert cat == monitor.TICKET_CATEGORIES['HARDWARE']
|
|
|
|
def test_nvme_is_hardware(self, monitor):
|
|
cat, _, _, _ = monitor._categorize_issue('NVMe drive warning on /dev/nvme0')
|
|
assert cat == monitor.TICKET_CATEGORIES['HARDWARE']
|
|
|
|
|
|
# ── _check_system_drive_indicators (dmesg scan) ───────────────────────────────
|
|
|
|
def _dmesg(monitor, lines):
|
|
"""Run _check_system_drive_indicators with a faked `dmesg` output."""
|
|
fake = MagicMock(returncode=0, stdout='\n'.join(lines), stderr='')
|
|
with patch('subprocess.run', return_value=fake):
|
|
return monitor._check_system_drive_indicators()
|
|
|
|
|
|
class TestSystemDriveIndicators:
|
|
def test_rbd_buffer_io_errors_are_ignored(self, monitor):
|
|
"""Ceph RBD Buffer I/O errors are storage-connectivity, not local hardware —
|
|
they must not raise a CRITICAL drive alert (regression for the false
|
|
'[hardware] CRITICAL: Buffer I/O errors' ticket)."""
|
|
lines = [
|
|
f'[Wed Jul 15 03:2{i}:51 2026] Buffer I/O error on dev rbd2, '
|
|
'logical block 8999, lost sync page write'
|
|
for i in range(75)
|
|
]
|
|
res = _dmesg(monitor, lines)
|
|
assert res['status'] == 'OK'
|
|
assert res['issues'] == []
|
|
|
|
def test_local_disk_buffer_io_errors_still_counted(self, monitor):
|
|
"""Genuine local-disk Buffer I/O errors must still be reported."""
|
|
lines = [
|
|
f'[Wed Jul 15 03:30:0{i} 2026] Buffer I/O error on dev sda1, '
|
|
'logical block 42, lost async page write'
|
|
for i in range(5)
|
|
]
|
|
res = _dmesg(monitor, lines)
|
|
assert res['status'] == 'CRITICAL'
|
|
assert any('Buffer I/O errors' in i and '5 occurrences' in i for i in res['issues'])
|
|
|
|
def test_rbd_noise_does_not_inflate_real_errors(self, monitor):
|
|
"""RBD errors mixed with real disk errors: only the real ones count."""
|
|
lines = [f'Buffer I/O error on dev rbd4, logical block {i}, lost sync page write'
|
|
for i in range(75)]
|
|
lines += [
|
|
'Buffer I/O error on dev sda1, logical block 1, lost async page write',
|
|
'Buffer I/O error on dev sda1, logical block 2, lost async page write',
|
|
]
|
|
res = _dmesg(monitor, lines)
|
|
# 2 real occurrences → WARNING (>=2), not CRITICAL (>=5)
|
|
assert res['status'] == 'WARNING'
|
|
assert any('Buffer I/O errors' in i and '2 occurrences' in i for i in res['issues'])
|
|
|
|
def test_device_mapper_and_loop_ignored(self, monitor):
|
|
"""Device-mapper and loop devices are also non-physical and ignored."""
|
|
lines = [
|
|
'Buffer I/O error on dev dm-3, logical block 5, lost sync page write',
|
|
'Buffer I/O error on dev loop0, logical block 6, lost sync page write',
|
|
]
|
|
res = _dmesg(monitor, lines)
|
|
assert res['status'] == 'OK'
|
|
assert res['issues'] == []
|
|
|
|
|
|
# ── _get_attribute_thresholds (drive age is not ticketed) ─────────────────────
|
|
|
|
class TestAttributeThresholds:
|
|
def test_power_on_hours_has_no_threshold(self, monitor):
|
|
"""Drive age must not raise a ticket: we run to hard-failure (Ceph redundancy +
|
|
PBS backups), so Power_On_Hours has no warning/critical threshold. Regression for
|
|
the '[hardware] Drive ... has SMART issues: Warning Power_On_Hours' age tickets."""
|
|
assert monitor._get_attribute_thresholds('Power_On_Hours', {}) is None
|
|
|
|
def test_real_failure_predictors_still_alert(self, monitor):
|
|
"""Genuine failure-predictor attributes must keep their thresholds."""
|
|
for attr in ('Reallocated_Sector_Ct', 'Current_Pending_Sector',
|
|
'Offline_Uncorrectable', 'Reported_Uncorrect'):
|
|
th = monitor._get_attribute_thresholds(attr, {})
|
|
assert th is not None and 'warning' in th and 'critical' in th
|
|
|
|
def test_power_on_hours_still_recorded_for_new_drive_logic(self, monitor):
|
|
"""Age is still *tracked* (just not ticketed) — new-drive detection must work."""
|
|
assert monitor._is_new_drive(100) is True # ~4 days
|
|
assert monitor._is_new_drive(64860) is False # ~7.4 years
|
|
|
|
|
|
# ── _format_lxc_dry_run (LXC storage shown in --dry-run summary; issue #23) ────
|
|
|
|
class TestLxcDryRunSummary:
|
|
def test_includes_container_usage_and_flags_over_threshold(self, monitor):
|
|
"""Regression for issue #23: dry-run summary must surface LXC storage."""
|
|
lxc = {
|
|
'status': 'WARNING',
|
|
'issues': ['LXC 105 high storage usage: 84.0% on /'],
|
|
'containers': [
|
|
{'vmid': '105', 'filesystems': [{'mountpoint': '/', 'usage_percent': 84.0}]},
|
|
],
|
|
}
|
|
text = '\n'.join(monitor._format_lxc_dry_run(lxc))
|
|
assert 'LXC Storage:' in text
|
|
assert 'CT105 /' in text and '84.0% used' in text
|
|
assert '⚠️' in text # flagged: over the 80% warning threshold
|
|
assert 'Issues: 1 found' in text
|
|
|
|
def test_healthy_container_not_flagged(self, monitor):
|
|
lxc = {'status': 'OK', 'issues': [],
|
|
'containers': [{'vmid': '119', 'filesystems': [{'mountpoint': '/', 'usage_percent': 26.0}]}]}
|
|
text = '\n'.join(monitor._format_lxc_dry_run(lxc))
|
|
assert 'CT119 /' in text and '26.0% used' in text
|
|
assert '⚠️' not in text
|
|
|
|
def test_empty_when_no_containers(self, monitor):
|
|
assert monitor._format_lxc_dry_run({'status': 'OK', 'containers': [], 'issues': []}) == []
|
|
|
|
def test_error_status_is_reported(self, monitor):
|
|
lines = monitor._format_lxc_dry_run({'status': 'ERROR', 'containers': [], 'issues': ['boom']})
|
|
assert any('check error' in ln for ln in lines)
|
|
|
|
|
|
# ── _check_cpu_usage (ticket on sustained load, not transient spikes) ──────────
|
|
|
|
def _cpu(monitor, instantaneous, load15, cpu_count=4):
|
|
"""Run _check_cpu_usage with mocked psutil + load average."""
|
|
with patch('psutil.cpu_percent', return_value=instantaneous), \
|
|
patch('psutil.cpu_count', return_value=cpu_count), \
|
|
patch('os.getloadavg', return_value=(load15, load15, load15)):
|
|
return monitor._check_cpu_usage()
|
|
|
|
|
|
def _report_with_cpu(cpu_health):
|
|
"""Minimal health_report skeleton (all sections empty) with cpu_health overridden,
|
|
sufficient for _detect_issues which direct-indexes several sections for logging."""
|
|
return {
|
|
'drives_health': {'drives': [], 'overall_status': 'HEALTHY'},
|
|
'memory_health': {'status': 'OK'},
|
|
'cpu_health': cpu_health,
|
|
'network_health': {'management_network': {}, 'ceph_network': {}},
|
|
'ceph_health': {},
|
|
'lxc_health': {'status': 'OK', 'issues': []},
|
|
'system_health': {'status': 'OK', 'issues': []},
|
|
'pbs_health': {},
|
|
}
|
|
|
|
|
|
class TestCpuSustainedGating:
|
|
def test_transient_spike_does_not_warn(self, monitor):
|
|
"""Instantaneous 100% but low 15-min load (e.g. 0.4/core = 10%) -> OK, no ticket."""
|
|
h = _cpu(monitor, instantaneous=100.0, load15=1.6, cpu_count=4) # 1.6/4 = 40%/core
|
|
assert h['status'] == 'OK'
|
|
assert h['sustained_high'] is False
|
|
assert h['cpu_usage_percent'] == 100.0 # still reported for context
|
|
assert h['load15_per_core_percent'] == 40.0
|
|
|
|
def test_sustained_high_load_warns(self, monitor):
|
|
"""15-min load 3.9/4 cores = 97.5%/core >= 95% threshold -> WARNING."""
|
|
h = _cpu(monitor, instantaneous=20.0, load15=3.9, cpu_count=4)
|
|
assert h['status'] == 'WARNING'
|
|
assert h['sustained_high'] is True
|
|
assert h['load15_per_core_percent'] == 97.5
|
|
|
|
def test_detect_issues_ignores_transient_spike(self, monitor):
|
|
"""A high instantaneous sample with sustained_high False must not create a CPU issue."""
|
|
report = _report_with_cpu({'cpu_usage_percent': 99.0, 'sustained_high': False,
|
|
'load15_per_core_percent': 30.0, 'status': 'OK'})
|
|
issues = monitor._detect_issues(report)
|
|
assert not any('CPU' in i for i in issues)
|
|
|
|
def test_detect_issues_flags_sustained(self, monitor):
|
|
"""sustained_high True must create exactly one stable, value-free CPU issue."""
|
|
report = _report_with_cpu({'cpu_usage_percent': 96.0, 'sustained_high': True,
|
|
'load15_per_core_percent': 98.0, 'status': 'WARNING'})
|
|
issues = monitor._detect_issues(report)
|
|
cpu_issues = [i for i in issues if 'CPU' in i]
|
|
assert len(cpu_issues) == 1
|
|
# Title-stable: no measured value baked into the message (only the threshold constant)
|
|
assert '98' not in cpu_issues[0]
|
|
assert 'sustained' in cpu_issues[0].lower()
|
|
|
|
def test_falls_back_to_instantaneous_without_loadavg(self, monitor):
|
|
"""If os.getloadavg is unavailable, don't go silent — use the instantaneous sample."""
|
|
with patch('psutil.cpu_percent', return_value=97.0), \
|
|
patch('psutil.cpu_count', return_value=4), \
|
|
patch('os.getloadavg', side_effect=OSError):
|
|
h = monitor._check_cpu_usage()
|
|
assert h['status'] == 'WARNING'
|
|
assert h['load15_per_core_percent'] is None
|