swf-monitor/tests/test_worker_disk_space.py

0001 """Advance capacity warning and lost visibility are different from recovery."""
0002 import importlib.util
0003 from pathlib import Path
0004 import sys
0005 import unittest
0006 from types import SimpleNamespace
0007
0008 from monitor_app.worker_disk_space import summarize
0009 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'alarms'))
0010 from swf_alarms.alarms import worker_disk_space as alarm
0011
0012
0013 class WorkerDiskSpaceTests(unittest.TestCase):
0014     def metrics(self, free=38000, stamp=1000):
0015         return (f'diskSample={stamp} ceFreeMB={free} ceTotalMB=200000 '
0016                 'ceIFree=900000 ceITotal=1000000')
0017
0018     def detect(self, record):
0019         record = dict(record, age_seconds=0)
0020         return list(alarm.detect(SimpleNamespace(_get=lambda _, params=None: record), {}))
0021
0022     def test_warns_before_one_gib_abort(self):
0023         record = summarize([('UM_GREX_PanDA_1', 1, self.metrics())], 1000)
0024         found = self.detect(record)
0025         self.assertEqual(len(found), 1)
0026         self.assertIn('low', found[0].subject)
0027         self.assertGreater(found[0].extra_data['free_bytes'], 30 * 2**30)
0028
0029     def test_worst_fresh_measurement_is_used(self):
0030         result = summarize([('UM_GREX_PanDA_1', 1, self.metrics(80000)),
0031                             ('UM_GREX_PanDA_1', 2, self.metrics(1000))], 1000)
0032         self.assertEqual(result['volumes'][0]['pandaid'], 2)
0033         self.assertIn('critical', self.detect(result)[0].subject)
0034
0035     def test_missing_and_stale_measurements_do_not_clear(self):
0036         for metrics in ('', self.metrics(stamp=1)):
0037             result = summarize([('UM_GREX_PanDA_1', 1, metrics)], 2000)
0038             self.assertIn('unavailable', self.detect(result)[0].subject)
0039
0040     def test_healthy_space_clears(self):
0041         self.assertEqual(self.detect(summarize([('UM_GREX_PanDA_1', 1,
0042                                                 self.metrics(80000))], 1000)), [])
0043
0044     def test_worker_volume_does_not_substitute_for_missing_shared_spool(self):
0045         metrics = self.metrics(80000).replace('ce', 'pilot')
0046         record = summarize([('UM_GREX_PanDA_1', 1, metrics)], 1000)
0047         self.assertEqual(record['reporting_jobs'], 1)
0048         self.assertIn('unavailable', self.detect(record)[0].subject)
0049
0050     def test_new_shared_spool_measurement_replaces_old_low_space(self):
0051         rows = [('UM_GREX_PanDA_1', 1, self.metrics(1000, stamp=950)),
0052                 ('UM_GREX_PanDA_1', 2, self.metrics(80000, stamp=1000))]
0053         for order in (rows, rows[::-1]):
0054             result = summarize(order, 1000)
0055             self.assertEqual(result['volumes'][0]['pandaid'], 2)
0056             self.assertEqual(self.detect(result), [])
0057
0058     def test_distinct_worker_filesystems_keep_worst_fresh_sample(self):
0059         rows = [('UM_GREX_PanDA_1', 1, self.metrics(1000, stamp=950).replace('ce', 'scratch')),
0060                 ('UM_GREX_PanDA_1', 2, self.metrics(80000, stamp=1000).replace('ce', 'scratch'))]
0061         self.assertEqual(summarize(rows, 1000)['volumes'][0]['pandaid'], 1)
0062
0063     def test_inode_pressure_warns_even_with_free_bytes(self):
0064         metrics = self.metrics(80000).replace('ceIFree=900000', 'ceIFree=10')
0065         self.assertEqual(len(self.detect(summarize([('UM_GREX_PanDA_1', 1, metrics)], 1000))), 1)
0066
0067
0068 if __name__ == '__main__':
0069     unittest.main()