Add host temperature telemetry and history

agent/tests/test_temperatures.py

@@ -0,0 +1,96 @@
+import errno
+import importlib.util
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+spec = importlib.util.spec_from_file_location("ec_agent_temperatures", Path(__file__).resolve().parents[1] / "ec_agent.py")
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+
+
+class TemperatureTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.hwmon = self.root / "class"
+        self.hwmon.mkdir()
+        self.patch = patch.object(module, "HWMON_ROOT", self.hwmon)
+        self.patch.start()
+        self.addCleanup(self.patch.stop)
+        self.agent = object.__new__(module.Agent)
+
+    def chip(self, number, device, readings):
+        physical = self.root / (device + str(number)) / "hwmon" / ("hwmon" + str(number))
+        physical.mkdir(parents=True)
+        (self.hwmon / physical.name).symlink_to(physical, target_is_directory=True)
+        (physical / "name").write_text(device)
+        for channel, label, value in readings:
+            (physical / (channel + "_input")).write_text(str(value))
+            if label is not None:
+                (physical / (channel + "_label")).write_text(label)
+        return physical
+
+    def test_ec2_sensors_and_units_without_guessing_unknown_devices(self):
+        self.chip(0, "nvme", [("temp1", "Composite", 50900), ("temp2", "Sensor 1", 50900)])
+        self.chip(1, "pch_cometlake", [("temp1", None, 62000)])
+        self.chip(2, "coretemp", [("temp1", "Package id 0", 65000), ("temp2", "Core 0", 63000)])
+        self.chip(3, "iwlwifi_1", [("temp1", None, "unavailable")])
+        samples = self.agent.temperatures()
+        self.assertEqual([s["celsius"] for s in samples], [50.9, 50.9, 62.0, 65.0, 63.0])
+        self.assertEqual([s["kind"] for s in samples], ["nvme", "nvme", "chipset", "cpu-package", "cpu-core"])
+        self.assertEqual(len({s["id"] for s in samples}), len(samples))
+
+    def test_fault_disabled_bad_value_and_enodata_do_not_lose_other_sensors(self):
+        chip = self.chip(0, "coretemp", [("temp" + str(i), "Core " + str(i), value)
+                         for i, value in enumerate([65000, 60000, "bad", 500000, 61000, 63000, 0], 1)])
+        (chip / "temp1_fault").write_text("1")
+        (chip / "temp2_enable").write_text("0")
+        original = Path.read_text
+        def read(path, *args, **kwargs):
+            if path.name == "temp5_input":
+                raise OSError(errno.ENODATA, "No data available")
+            return original(path, *args, **kwargs)
+        with patch.object(Path, "read_text", read):
+            self.assertEqual([s["celsius"] for s in self.agent.temperatures()], [63.0, 0.0])
+
+    def test_amd_keeps_physical_die_readings_and_excludes_control_offsets(self):
+        self.chip(0, "k10temp", [("temp1", "Tctl", 85000), ("temp2", "Tdie", 65000), ("temp3", "Tccd1", 64000)])
+        self.assertEqual([s["label"] for s in self.agent.temperatures()], ["Tdie", "Tccd1"])
+
+    def test_sensor_identity_survives_hwmon_renumbering_and_separates_drives(self):
+        physical = self.chip(1, "nvme", [("temp1", "Composite", 50000)])
+        self.chip(2, "nvme", [("temp1", "Composite", 51000)])
+        before = {s["celsius"]: s["id"] for s in self.agent.temperatures()}
+        renamed = physical.with_name("hwmon9")
+        physical.rename(renamed)
+        (self.hwmon / "hwmon1").unlink()
+        (self.hwmon / "hwmon9").symlink_to(renamed, target_is_directory=True)
+        self.assertEqual(before, {s["celsius"]: s["id"] for s in self.agent.temperatures()})
+        self.assertEqual(len(set(before.values())), 2)
+
+    def test_missing_hwmon_and_disappearing_devices_are_optional(self):
+        self.assertEqual(self.agent.temperatures(), [])
+        (self.hwmon / "hwmon0").symlink_to(self.root / "missing")
+        self.assertEqual(self.agent.temperatures(), [])
+        with patch.object(Path, "glob", side_effect=PermissionError()):
+            self.assertEqual(self.agent.temperatures(), [])
+
+    def test_direct_hwmon_child_keeps_identity_when_renumbered(self):
+        physical = self.chip(1, "nvme", [("temp1", "Composite", 50000)])
+        direct = physical.parent.parent / "hwmon8"
+        physical.rename(direct)
+        (self.hwmon / "hwmon1").unlink()
+        (self.hwmon / "hwmon8").symlink_to(direct, target_is_directory=True)
+        before = self.agent.temperatures()[0]["id"]
+        renamed = direct.with_name("hwmon9")
+        direct.rename(renamed)
+        (self.hwmon / "hwmon8").unlink()
+        (self.hwmon / "hwmon9").symlink_to(renamed, target_is_directory=True)
+        self.assertEqual(self.agent.temperatures()[0]["id"], before)
+
+    def test_inventory_size_is_bounded(self):
+        self.chip(0, "coretemp", [("temp" + str(i), "Core " + str(i), 65000) for i in range(1, 200)])
+        self.assertEqual(len(self.agent.temperatures()), 128)