test: expand test coverage for manufacturer profiles and ticket classification
Lint / Python (flake8) (push) Successful in 1m14s
Test / Python Tests (pytest) (push) Successful in 1m29s
Lint / Notify on failure (push) Has been skipped
Security / Python Security (bandit) (push) Failing after 11m40s

Add 42 new tests covering _get_manufacturer_profile (Seagate/WD/Samsung/Toshiba/
OOS/Ridata/unknown), _should_monitor_attribute, _get_attribute_thresholds,
_get_issue_type, _get_impact_level, and _categorize_issue.

Toshiba MG08 prefix matching and the raised High_Fly_Writes/Command_Timeout
thresholds introduced in the previous commit are now covered by tests.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-25 19:52:59 -04:00
parent 5ce5593f4a
commit 823ff18890
+201
View File
@@ -222,3 +222,204 @@ class TestIsPhysicalDisk:
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']