98 lines
4.8 KiB
Python
98 lines
4.8 KiB
Python
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()
|