| 1 | from __future__ import annotations
|
| 2 |
|
| 3 | import sys
|
| 4 | import unittest
|
| 5 | from pathlib import Path
|
| 6 |
|
| 7 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
| 8 |
|
| 9 | from oakbench.envwatch import (
|
| 10 | environment_suspect,
|
| 11 | parse_meminfo_available_bytes,
|
| 12 | parse_pmset_cpu_speed_limit,
|
| 13 | parse_vm_stat_free_bytes,
|
| 14 | sample_environment,
|
| 15 | )
|
| 16 |
|
| 17 |
|
| 18 | def _clean_sample() -> dict:
|
| 19 | return {
|
| 20 | "loadavg": [0.5, 0.4, 0.3],
|
| 21 | "mem_available_bytes": 8 * 1024**3,
|
| 22 | "cpu_speed_limit_pct": 100,
|
| 23 | "thermal_zone_max_c": None,
|
| 24 | "cpu_governor": None,
|
| 25 | }
|
| 26 |
|
| 27 |
|
| 28 | class EnvironmentSuspectTest(unittest.TestCase):
|
| 29 | def test_clean_window_has_no_reasons(self) -> None:
|
| 30 | self.assertEqual(environment_suspect(_clean_sample(), _clean_sample(), cpu_count=8), [])
|
| 31 |
|
| 32 | def test_high_load_at_either_boundary_is_flagged(self) -> None:
|
| 33 | loaded = dict(_clean_sample(), loadavg=[16.0, 8.0, 4.0])
|
| 34 | for before, after in ((loaded, _clean_sample()), (_clean_sample(), loaded)):
|
| 35 | reasons = environment_suspect(before, after, cpu_count=8)
|
| 36 | self.assertEqual(len(reasons), 1)
|
| 37 | self.assertIn("load_per_core", reasons[0])
|
| 38 |
|
| 39 | def test_load_exactly_at_threshold_is_not_flagged(self) -> None:
|
| 40 | boundary = dict(_clean_sample(), loadavg=[12.0, 0.0, 0.0]) # 12 / 8 == 1.5
|
| 41 | self.assertEqual(environment_suspect(boundary, _clean_sample(), cpu_count=8), [])
|
| 42 | above = dict(_clean_sample(), loadavg=[12.01, 0.0, 0.0])
|
| 43 | self.assertEqual(len(environment_suspect(above, _clean_sample(), cpu_count=8)), 1)
|
| 44 |
|
| 45 | def test_thermal_throttling_is_flagged(self) -> None:
|
| 46 | throttled = dict(_clean_sample(), cpu_speed_limit_pct=62)
|
| 47 | reasons = environment_suspect(_clean_sample(), throttled, cpu_count=8)
|
| 48 | self.assertEqual(len(reasons), 1)
|
| 49 | self.assertIn("CPU_Speed_Limit=62", reasons[0])
|
| 50 |
|
| 51 | def test_null_signals_are_unmeasured_not_suspect(self) -> None:
|
| 52 | nulls = {"loadavg": None, "cpu_speed_limit_pct": None, "cpu_governor": None}
|
| 53 | self.assertEqual(environment_suspect(nulls, nulls, cpu_count=8), [])
|
| 54 |
|
| 55 | def test_non_performance_governor_is_informational_suspicion(self) -> None:
|
| 56 | governed = dict(_clean_sample(), cpu_governor="powersave")
|
| 57 | reasons = environment_suspect(governed, governed, cpu_count=8)
|
| 58 | self.assertEqual(len(reasons), 1)
|
| 59 | self.assertIn("powersave", reasons[0])
|
| 60 | self.assertIn("informational", reasons[0])
|
| 61 |
|
| 62 | performance = dict(_clean_sample(), cpu_governor="performance")
|
| 63 | self.assertEqual(environment_suspect(performance, performance, cpu_count=8), [])
|
| 64 |
|
| 65 |
|
| 66 | class ParserTest(unittest.TestCase):
|
| 67 | def test_meminfo_available_parses_to_bytes(self) -> None:
|
| 68 | text = "MemTotal: 16000000 kB\nMemFree: 100 kB\nMemAvailable: 2048 kB\n"
|
| 69 | self.assertEqual(parse_meminfo_available_bytes(text), 2048 * 1024)
|
| 70 | self.assertIsNone(parse_meminfo_available_bytes("MemTotal: 16 kB\n"))
|
| 71 |
|
| 72 | def test_vm_stat_free_pages_parse_to_bytes(self) -> None:
|
| 73 | text = 'Mach Virtual Memory Statistics: (page size of 16384 bytes)\nPages free: 1000.\n'
|
| 74 | self.assertEqual(parse_vm_stat_free_bytes(text), 1000 * 16384)
|
| 75 | self.assertIsNone(parse_vm_stat_free_bytes("Pages free: 1000.\n"))
|
| 76 |
|
| 77 | def test_pmset_cpu_speed_limit_parse(self) -> None:
|
| 78 | self.assertEqual(parse_pmset_cpu_speed_limit("CPU_Speed_Limit \t= 100"), 100)
|
| 79 | self.assertEqual(parse_pmset_cpu_speed_limit("CPU_Speed_Limit = 51\n"), 51)
|
| 80 | self.assertIsNone(parse_pmset_cpu_speed_limit("No thermal warning level"))
|
| 81 |
|
| 82 |
|
| 83 | class SampleEnvironmentTest(unittest.TestCase):
|
| 84 | def test_sample_has_all_nullable_fields(self) -> None:
|
| 85 | sample = sample_environment()
|
| 86 | for key in ("loadavg", "mem_available_bytes", "cpu_speed_limit_pct", "thermal_zone_max_c", "cpu_governor"):
|
| 87 | self.assertIn(key, sample)
|
| 88 |
|
| 89 |
|
| 90 | if __name__ == "__main__":
|
| 91 | unittest.main()
|