ProxMenux/AppImage/scripts/tests/test_zfs_pool_details.py

"""ZFS pool layout parsing for the Storage view; never imports flask_server.

Fixtures are real `zpool status -j --json-int -P` output from PVE 9 /
OpenZFS 2.4.4 hosts, with disk serials and WWNs replaced:
- zpool_status_multi_pool.json: mirror, mirror, raidz2, mirror (one member with
  slow I/Os). Captured without --json-int and normalised to integers.
- zpool_status_single_rpool.json: single disk directly under the root vdev.
- zpool_status_degraded_raidz1.json: file-backed raidz1 with one member
  offlined, plus log, cache and spare devices; never scrubbed.

Run: python3 -m unittest discover -s AppImage/scripts/tests -p test_zfs_pool_details.py -v
"""
import ast
import copy
import json
import posixpath
import re
from pathlib import Path
from types import SimpleNamespace
import unittest


SOURCE = Path(__file__).resolve().parents[1] / "flask_server.py"
FIXTURES = Path(__file__).resolve().parent / "fixtures"
NAMES = {"_zfs_member_disk", "_zfs_errors", "_zfs_device", "_ZFS_PARITY_RE",
         "_zfs_vdev", "_zfs_pool_details"}


def fake_os(links=None, existing=()):
    """Minimal `os` stand-in: realpath via a symlink map, exists via a set."""
    links = links or {}
    return SimpleNamespace(path=SimpleNamespace(
        realpath=lambda p: links.get(p, p),
        exists=lambda p: p in existing,
        basename=posixpath.basename,
        dirname=posixpath.dirname,
    ))


def load(os_module=None):
    tree = ast.parse(SOURCE.read_text())
    nodes = [n for n in tree.body
             if (isinstance(n, ast.FunctionDef) and n.name in NAMES)
             or (isinstance(n, ast.Assign) and any(getattr(t, "id", None) in NAMES for t in n.targets))]
    found = {n.name if isinstance(n, ast.FunctionDef) else n.targets[0].id for n in nodes}
    assert found == NAMES, NAMES - found
    env = {"os": os_module or fake_os(), "re": re}
    exec(compile(ast.Module(body=nodes, type_ignores=[]), str(SOURCE), "exec"), env)
    return env


def pools(name):
    return json.loads((FIXTURES / f"{name}.json").read_text())["pools"]


class ZfsPoolDetailsTests(unittest.TestCase):
    def setUp(self):
        self.details = load()["_zfs_pool_details"]

    def test_degraded_raidz1_with_log_cache_and_spare(self):
        pool = self.details(pools("zpool_status_degraded_raidz1")["pmx-test"])

        self.assertEqual([v["class"] for v in pool["vdevs"]], ["normal", "log", "cache", "spare"])
        raidz = pool["vdevs"][0]
        self.assertEqual((raidz["type"], raidz["parity"], raidz["state"]), ("raidz", 1, "DEGRADED"))
        self.assertEqual([d["state"] for d in raidz["devices"]], ["ONLINE", "OFFLINE", "ONLINE"])
        self.assertEqual(pool["vdevs"][3]["state"], "AVAIL")
        self.assertIsNone(pool["scan"])
        self.assertTrue(pool["status"].startswith("One or more devices has been taken offline"))
        self.assertNotIn("\t", pool["status"])
        self.assertNotIn("\n", pool["action"])
        # File vdevs have no block device behind them
        self.assertIsNone(raidz["devices"][0]["disk"])

    def test_single_disk_pool_is_a_disk_vdev(self):
        pool = self.details(pools("zpool_status_single_rpool")["rpool"])

        self.assertEqual(len(pool["vdevs"]), 1)
        vdev = pool["vdevs"][0]
        self.assertEqual((vdev["type"], vdev["class"], vdev["parity"]), ("disk", "normal", None))
        self.assertEqual(len(vdev["devices"]), 1)
        self.assertEqual(vdev["name"], vdev["devices"][0]["path"])

    def test_finished_scrub_has_no_progress_percentage(self):
        pool = self.details(pools("zpool_status_single_rpool")["rpool"])

        self.assertEqual((pool["scan"]["function"], pool["scan"]["state"]), ("SCRUB", "FINISHED"))
        self.assertEqual(pool["scan"]["end_time"], 1791674884)
        # issued/to_examine is 100.1% for this real scrub; only meaningful while running
        self.assertIsNone(pool["scan"]["percent"])

    def test_mirror_raidz2_and_slow_ios(self):
        multi = pools("zpool_status_multi_pool")
        parsed = {name: self.details(pool) for name, pool in multi.items()}

        self.assertEqual(parsed["tank-rz2"]["vdevs"][0]["parity"], 2)
        self.assertEqual(len(parsed["tank-rz2"]["vdevs"][0]["devices"]), 5)
        for name in ("rpool", "tank-m1", "tank-spin-1"):
            self.assertEqual(parsed[name]["vdevs"][0]["type"], "mirror", name)
        slow = [d["slow_ios"] for d in parsed["tank-m1"]["vdevs"][0]["devices"]]
        self.assertEqual(slow, [0, 69])
        # Member paths come from `path`, not the JSON key (which may drop -partN)
        paths = [d["path"] for p in parsed.values() for v in p["vdevs"] for d in v["devices"]]
        self.assertTrue(all(p.startswith("/dev/disk/") for p in paths), paths)

    def test_resilver_with_spare_swapping_in(self):
        # Synthetic: the degraded fixture with d2 FAULTED and the spare
        # attached under spare-1, as `zpool status` shows during a resilver.
        raw = copy.deepcopy(pools("zpool_status_degraded_raidz1")["pmx-test"])
        members = raw["vdevs"]["pmx-test"]["vdevs"]["raidz1-0"]["vdevs"]
        failed = members.pop("/root/zfstest/d2.img")
        failed.update(state="FAULTED", read_errors=12)
        spare = dict(raw["spares"]["/root/zfstest/d6.img"], state="ONLINE")
        members["spare-1"] = {"name": "spare-1", "vdev_type": "spare", "state": "DEGRADED",
                              "vdevs": {"d2": failed, "d6": spare}}
        raw["spares"]["/root/zfstest/d6.img"]["state"] = "INUSE"
        raw["scan_stats"] = {"function": "RESILVER", "state": "SCANNING", "start_time": 1,
                             "end_time": 0, "to_examine": 1000, "issued": 425, "errors": 0,
                             "processed": 0}

        pool = self.details(raw)

        devices = pool["vdevs"][0]["devices"]
        self.assertEqual([d["via"] for d in devices], [None, None, "spare-1", "spare-1"])
        self.assertEqual(devices[2]["read_errors"], 12)
        self.assertEqual(pool["scan"]["percent"], 42.5)
        self.assertEqual(pool["vdevs"][-1]["state"], "INUSE")

    def test_output_is_json_serialisable_without_guids(self):
        for name, pool in pools("zpool_status_multi_pool").items():
            text = json.dumps(self.details(pool))
            self.assertNotIn("guid", text, name)


class ZfsMemberDiskTests(unittest.TestCase):
    def test_partition_resolves_to_parent_disk(self):
        member_disk = load(fake_os(
            links={"/dev/disk/by-partlabel/ssd-t1": "/dev/sdf1",
                   "/sys/class/block/sdf1": "/sys/devices/pci0000:00/0000:02:00.1/ata1/host0/target0:0:0/0:0:0:0/block/sdf/sdf1"},
            existing={"/sys/class/block/sdf1/partition"},
        ))["_zfs_member_disk"]
        self.assertEqual(member_disk("/dev/disk/by-partlabel/ssd-t1"), "sdf")

    def test_nvme_partition_resolves_to_namespace(self):
        member_disk = load(fake_os(
            links={"/dev/disk/by-id/nvme-X-part3": "/dev/nvme0n1p3",
                   "/sys/class/block/nvme0n1p3": "/sys/devices/pci0000:00/0000:08:00.0/nvme/nvme0/nvme0n1/nvme0n1p3"},
            existing={"/sys/class/block/nvme0n1p3/partition"},
        ))["_zfs_member_disk"]
        self.assertEqual(member_disk("/dev/disk/by-id/nvme-X-part3"), "nvme0n1")

    def test_whole_disk_member(self):
        member_disk = load(fake_os(
            links={"/dev/disk/by-id/wwn-0x1": "/dev/sdj"},
            existing={"/sys/class/block/sdj"},
        ))["_zfs_member_disk"]
        self.assertEqual(member_disk("/dev/disk/by-id/wwn-0x1"), "sdj")

    def test_missing_device_and_file_vdev_give_none(self):
        member_disk = load(fake_os())["_zfs_member_disk"]
        self.assertIsNone(member_disk("/dev/disk/by-id/gone"))
        self.assertIsNone(member_disk("/root/zfstest/d1.img"))


if __name__ == "__main__":
    unittest.main()