feat: add manual disk discovery and weekly capacity cache

tests/test_storage_cache.py

@@ -6,7 +6,7 @@ from storage_cache import StorageCache
 
 
 class StorageTests(unittest.TestCase):
-    def test_daily_reads_and_restart_cache(self):
+    def test_weekly_reads_and_restart_cache(self):
         with tempfile.TemporaryDirectory() as directory:
             now=[1000];reads=[]
             def reader(path):
@@ -18,7 +18,7 @@ class StorageTests(unittest.TestCase):
             self.assertEqual(len(reads),1)
             restored=StorageCache(p,reader,lambda:now[0]);restored.sample()
             self.assertEqual(len(reads),1)
-            now[0]=87400;restored.sample()
+            now[0]=605800;restored.sample()
             self.assertEqual(len(reads),2)
 
     def test_failure_does_not_fabricate_zero_or_retry_each_second(self):

tests/test_storage_config.py

@@ -0,0 +1,35 @@
+import importlib.util
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "app"))
+stem = "pigway_ubuntu_agent_api" if (ROOT / "app/pigway_ubuntu_agent_api.py").exists() else "pigway_pi_control_api"
+spec = importlib.util.spec_from_file_location("storage_api_under_test", ROOT / "app" / (stem + ".py"))
+api = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(api)
+
+
+class StorageConfigTests(unittest.TestCase):
+    def test_config_round_trip_and_invalid_edits_are_atomic(self):
+        with tempfile.TemporaryDirectory() as directory:
+            config = Path(directory) / "agent.conf"
+            config.write_text("[device]\nname = auto\n")
+            identifier = "a" * 32
+            with patch.object(api, "CFG", config):
+                api.update_ini({"storage_names": {identifier: "媒体盘 100%"}, "storage_visible": {identifier: "false"}})
+                self.assertEqual(api.config_json()["storage_names"][identifier], "媒体盘 100%")
+                self.assertEqual(api.config_json()["storage_visible"][identifier], "false")
+                original = config.read_text()
+                for updates in [
+                    {"storage_visible": {identifier: "yes"}},
+                    {"storage_names": {"invalid-id": "Bad"}},
+                    {"storage_names": {identifier: "x" * 65}},
+                    {"storage_names": {identifier: "bad\tname"}},
+                ]:
+                    with self.assertRaises(ValueError):
+                        api.update_ini(updates)
+                    self.assertEqual(config.read_text(), original)

tests/test_storage_inventory.py

@@ -0,0 +1,97 @@
+import configparser
+import json
+import sys
+import tempfile
+import unittest
+from collections import namedtuple
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "app"))
+from storage_cache import StorageInventory
+
+Usage = namedtuple("Usage", "total used free")
+
+
+class StorageInventoryTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.mounts = Path(self.tmp.name) / "mountinfo"
+        self.mounts.write_text("1 0 8:1 / / rw - ext4 /dev/sda1 rw\n2 0 8:2 / /media/My\\040Disk rw - ext4 /dev/sda2 rw\n3 0 8:2 /folder /bind rw - ext4 /dev/sda2 rw\n")
+        self.data = {"blockdevices": [
+            {"name": "/dev/sda", "type": "disk", "serial": "serial", "children": [
+                {"name": "/dev/sda1", "type": "part", "uuid": "root", "maj:min": "8:1", "fstype": "ext4", "size": 1000},
+                {"name": "/dev/sda2", "type": "part", "uuid": "data", "maj:min": "8:2", "fstype": "ext4", "size": 2000},
+            ]},
+            {"name": "/dev/sdb", "type": "disk", "size": 3000},
+            {"name": "/dev/loop0", "type": "loop", "size": 4000},
+            {"name": "/dev/zram0", "type": "disk", "fstype": "swap"},
+        ]}
+        self.now = 100
+        self.reader = StorageInventory(clock=lambda: self.now, runner=lambda: json.dumps(self.data),
+                                       mounts_path=self.mounts, usage=lambda p: Usage(1000, 0, 1000),
+                                       cache_path=Path(self.tmp.name)/"inventory.json")
+        self.reader.scan()
+
+    def test_weekly_usage_cache_survives_service_restart(self):
+        capacity=Path(self.tmp.name)/'usage.json'
+        self.reader.capacity_path=capacity
+        self.reader.sample()
+        def forbidden(*args):raise AssertionError('Not due; should not read capacity or discover disks')
+        restarted=StorageInventory(clock=lambda:self.now+100,runner=forbidden,mounts_path=self.mounts,
+                                   usage=forbidden,cache_path=self.reader.cache_path,capacity_path=capacity)
+        self.assertEqual(len(restarted.sample()),3)
+        self.assertEqual(restarted.observed_at,self.now)
+        self.assertEqual(restarted.next_update_at,self.now+604800)
+
+    def test_mounts_raw_disk_and_pseudo_filters(self):
+        disks = self.reader.sample()
+        self.assertEqual(len(disks), 3)
+        self.assertEqual(disks[0]["mountpoint"], "/")
+        self.assertEqual(next(d for d in disks if d.get("uuid") == "data")["mountpoint"], "/media/My Disk")
+        self.assertEqual(disks[0]["used_percent"], 0)
+        raw = next(d for d in disks if d["device"] == "/dev/sdb")
+        self.assertEqual(raw["state"], "unmounted")
+        self.assertIsNone(raw["used_bytes"])
+
+    def test_uuid_survives_renumber_and_configuration(self):
+        disk = self.reader.sample()[0]
+        self.data["blockdevices"][0]["children"][0]["name"] = "/dev/sdz1"
+        self.now += 61
+        self.reader.scan()
+        config = configparser.ConfigParser()
+        config.read_string(f'[storage_names]\n{disk["id"]} = 数据盘 100%\n[storage_visible]\n{disk["id"]} = false\n')
+        result = next(d for d in self.reader.sample(config) if d["id"] == disk["id"])
+        self.assertEqual(result["name"], "数据盘 100%")
+        self.assertFalse(result["visible"])
+
+    def test_removal_and_discovery_error_are_distinct(self):
+        disk = self.reader.sample()[0]
+        config = configparser.ConfigParser()
+        config.read_dict({"storage_names": {disk["id"]: "System"}})
+        self.reader.runner = lambda: "invalid json"
+        self.now += 61
+        with self.assertRaises(ValueError):self.reader.scan()
+        self.assertEqual(self.reader.sample(config)[0]["state"], "mounted")
+        self.reader.runner = lambda: '{"blockdevices":[]}'
+        self.now += 61
+        self.reader.scan()
+        self.assertEqual(self.reader.sample(config)[0]["state"], "missing")
+        self.assertFalse(self.reader.error)
+
+    def test_hotplug_requires_manual_scan_and_inventory_survives_restart(self):
+        self.assertEqual(len(self.reader.sample()), 3)
+        self.data["blockdevices"].append({"name": "/dev/sdc", "type": "disk", "uuid": "new", "size": 9000})
+        self.now += 29
+        self.assertEqual(len(self.reader.sample()), 3)
+        self.now += 100000
+        self.assertEqual(len(self.reader.sample()), 3)
+        self.reader.scan()
+        self.assertEqual(len(self.reader.sample()), 4)
+        def forbidden():raise AssertionError("automatic enumeration is forbidden")
+        restored=StorageInventory(clock=lambda:self.now,runner=forbidden,mounts_path=self.mounts,
+                                  usage=lambda p:Usage(1000,0,1000),cache_path=self.reader.cache_path)
+        self.assertEqual(len(restored.sample()),4)
+
+
+if __name__ == "__main__":
+    unittest.main()