File indexing completed on 2026-11-09 10:47:38
0001 """Host-disk alarms distinguish early pressure, recovery and lost visibility."""
0002 from datetime import datetime, timedelta, timezone
0003 import importlib.util
0004 from pathlib import Path
0005 from types import SimpleNamespace
0006 import sys
0007 import unittest
0008 from unittest.mock import MagicMock, patch
0009
0010 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'alarms'))
0011 from swf_alarms.alarms import host_disk_space as alarm
0012
0013
0014 class HostDiskSpaceTests(unittest.TestCase):
0015 def detect(self, volumes, age=0):
0016 stamp = datetime.now(timezone.utc) - timedelta(minutes=age)
0017 cursor = MagicMock()
0018 cursor.__enter__.return_value = cursor
0019 cursor.fetchall.return_value = [dict(key='host_report:osgsub01', built_at=stamp,
0020 value={'collected_at': stamp.isoformat(), 'health': {'volumes': volumes}})]
0021 conn = SimpleNamespace(cursor=lambda: cursor)
0022 return list(alarm.detect(SimpleNamespace(db_conn=conn), {'hosts': 'osgsub01'}))
0023
0024 def test_pressure_warns_before_full_and_aliases_are_not_repeated(self):
0025 volume = dict(device=2, used_percent=90, free_bytes=26 * 2**30)
0026 found = self.detect({'/var': volume, '/data': volume})
0027 self.assertEqual(len(found), 1)
0028 self.assertIn('/var 90.0% full, 26.0 GiB free', found[0].subject)
0029 self.assertEqual(len(found[0].extra_data['volumes']), 1)
0030
0031 def test_healthy_space_clears_and_small_root_is_not_false_alarm(self):
0032 self.assertEqual(self.detect({'/': dict(used_percent=71, free_bytes=3 * 2**30)}), [])
0033
0034 def test_missing_or_stale_reading_stays_an_incident(self):
0035 low = {'/var': dict(used_percent=97, free_bytes=8 * 2**30)}
0036 for found in (self.detect(low), self.detect(low, age=20), self.detect({})):
0037 self.assertEqual(found[0].dedupe_key, 'host-disk:osgsub01')
0038 self.assertIn('monitoring unavailable', self.detect(low, age=20)[0].subject)
0039
0040 def test_reporter_records_unprivileged_free_space(self):
0041 path = Path(__file__).resolve().parents[1] / 'scripts/harvester-reporter.py'
0042 spec = importlib.util.spec_from_file_location('reporter_disk', path)
0043 reporter = importlib.util.module_from_spec(spec)
0044 spec.loader.exec_module(reporter)
0045 health = reporter.disk_health()
0046 self.assertGreater(health['volumes']['/']['total_bytes'], 0)
0047 self.assertIn('free_bytes', health['volumes']['/'])
0048 actual = reporter.os.statvfs
0049 def without_data(path):
0050 if path == '/data':
0051 raise FileNotFoundError(path)
0052 return actual(path)
0053 with patch.object(reporter.os, 'statvfs', side_effect=without_data):
0054 self.assertNotIn('/data', reporter.disk_health()['volumes'])
0055
0056
0057 if __name__ == '__main__':
0058 unittest.main()
