""" Tests for storage details collection and parsing. Tests cover: - Storage details schema validation - Normalization of storage data from different OS types - Parsing of lsblk, df, findmnt, zpool, zfs outputs - Feature flags detection - Collection metadata handling """ import pytest from datetime import datetime from typing import Dict, Any from app.schemas.host_metrics import HostMetricsCreate, HostMetricsSummary # ============================================================================= # Fixtures: Sample storage_details data for different OS types # ============================================================================= @pytest.fixture def linux_debian_storage_details() -> Dict[str, Any]: """Sample storage_details from a Debian Linux host with ext4 + docker overlay.""" return { "os_type": "linux-debian", "status": "ok", "collected_at": "2025-12-19T10:00:00Z", "commands_run": [ {"cmd": "lsblk", "status": "ok"}, {"cmd": "findmnt", "status": "ok"}, {"cmd": "df", "status": "ok"}, {"cmd": "lvm", "status": "ok"} ], "partial_failures": [], "summary": { "total_bytes": 500107862016, "used_bytes": 125026965504, "free_bytes": 349600112640, "used_pct": 25.0 }, "block_devices": [ { "name": "sda", "type": "disk", "size": 500107862016, "model": "Samsung SSD 860", "serial": "S3YNXXXXXXXX", "children": [ { "name": "sda1", "type": "part", "size": 536870912, "fstype": "vfat", "mountpoint": "/boot/efi", "uuid": "XXXX-XXXX" }, { "name": "sda2", "type": "part", "size": 499570991104, "fstype": "ext4", "mountpoint": "/", "uuid": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" } ] } ], "mounts": [ { "target": "/", "source": "/dev/sda2", "fstype": "ext4", "options": "rw,relatime" }, { "target": "/boot/efi", "source": "/dev/sda1", "fstype": "vfat", "options": "rw,relatime" } ], "filesystems": [ { "device": "/dev/sda2", "fstype": "ext4", "size_bytes": 499570991104, "used_bytes": 125026965504, "avail_bytes": 349600112640, "use_pct": 26, "mountpoint": "/" }, { "device": "/dev/sda1", "fstype": "vfat", "size_bytes": 536870912, "used_bytes": 5242880, "avail_bytes": 531628032, "use_pct": 1, "mountpoint": "/boot/efi" } ], "lvm": { "pvs": [], "vgs": [], "lvs": [] }, "zfs": { "pools": [], "datasets": [] }, "feature_flags": { "has_lvm": False, "has_zfs": False, "has_lsblk": True, "has_findmnt": True } } @pytest.fixture def linux_alpine_storage_details() -> Dict[str, Any]: """Sample storage_details from an Alpine Linux host (minimal busybox).""" return { "os_type": "linux-alpine", "status": "partial", "collected_at": "2025-12-19T10:00:00Z", "commands_run": [ {"cmd": "lsblk", "status": "ok"}, {"cmd": "df", "status": "ok"} ], "partial_failures": ["findmnt parse failed"], "summary": { "total_bytes": 10737418240, "used_bytes": 2147483648, "free_bytes": 8589934592, "used_pct": 20.0 }, "block_devices": [ { "name": "vda", "type": "disk", "size": 10737418240, "children": [ { "name": "vda1", "type": "part", "size": 10736369664, "fstype": "ext4", "mountpoint": "/" } ] } ], "mounts": [], "filesystems": [ { "device": "/dev/vda1", "fstype": "ext4", "size_bytes": 10736369664, "used_bytes": 2147483648, "avail_bytes": 8588886016, "use_pct": 20, "mountpoint": "/" } ], "lvm": {"pvs": [], "vgs": [], "lvs": []}, "zfs": {"pools": [], "datasets": []}, "feature_flags": { "has_lvm": False, "has_zfs": False, "has_lsblk": True, "has_findmnt": False } } @pytest.fixture def truenas_core_storage_details() -> Dict[str, Any]: """Sample storage_details from TrueNAS CORE (FreeBSD + ZFS).""" return { "os_type": "truenas-core", "status": "ok", "collected_at": "2025-12-19T10:00:00Z", "commands_run": [ {"cmd": "df", "status": "ok"}, {"cmd": "zpool list", "status": "ok"}, {"cmd": "zfs list", "status": "ok"} ], "partial_failures": [], "summary": { "total_bytes": 4000787030016, "used_bytes": 2800550921011, "free_bytes": 1200236109005, "used_pct": 70.0 }, "block_devices": [], "mounts": [], "filesystems": [ { "device": "tank", "fstype": "zfs", "size_bytes": 4000787030016, "used_bytes": 2800550921011, "avail_bytes": 1200236109005, "use_pct": 70, "mountpoint": "/mnt/tank" } ], "lvm": {"pvs": [], "vgs": [], "lvs": []}, "zfs": { "pools": [ { "name": "tank", "size_bytes": 4000787030016, "alloc_bytes": 2800550921011, "free_bytes": 1200236109005, "cap_pct": 70, "health": "ONLINE" } ], "datasets": [ { "name": "tank", "used_bytes": 2800550921011, "avail_bytes": 1200236109005, "refer_bytes": 98304, "mountpoint": "/mnt/tank", "type": "filesystem" }, { "name": "tank/apps", "used_bytes": 107374182400, "avail_bytes": 1200236109005, "refer_bytes": 107374182400, "mountpoint": "/mnt/tank/apps", "type": "filesystem" }, { "name": "tank/media", "used_bytes": 2693176738611, "avail_bytes": 1200236109005, "refer_bytes": 2693176738611, "mountpoint": "/mnt/tank/media", "type": "filesystem" } ] }, "feature_flags": { "has_lvm": False, "has_zfs": True, "has_lsblk": False, "has_findmnt": False } } @pytest.fixture def truenas_scale_storage_details() -> Dict[str, Any]: """Sample storage_details from TrueNAS SCALE (Linux + ZFS).""" return { "os_type": "truenas-scale", "status": "ok", "collected_at": "2025-12-19T10:00:00Z", "commands_run": [ {"cmd": "lsblk", "status": "ok"}, {"cmd": "findmnt", "status": "ok"}, {"cmd": "df", "status": "ok"}, {"cmd": "zpool list", "status": "ok"}, {"cmd": "zfs list", "status": "ok"} ], "partial_failures": [], "summary": { "total_bytes": 8001574060032, "used_bytes": 4800944436019, "free_bytes": 3200629624013, "used_pct": 60.0 }, "block_devices": [ { "name": "sda", "type": "disk", "size": 4000787030016, "model": "WDC WD40EFAX", "rota": True }, { "name": "sdb", "type": "disk", "size": 4000787030016, "model": "WDC WD40EFAX", "rota": True } ], "mounts": [], "filesystems": [ { "device": "boot-pool", "fstype": "zfs", "size_bytes": 64424509440, "used_bytes": 3221225472, "avail_bytes": 61203283968, "use_pct": 5, "mountpoint": "/boot" } ], "lvm": {"pvs": [], "vgs": [], "lvs": []}, "zfs": { "pools": [ { "name": "boot-pool", "size_bytes": 64424509440, "alloc_bytes": 3221225472, "free_bytes": 61203283968, "cap_pct": 5, "health": "ONLINE" }, { "name": "data", "size_bytes": 8001574060032, "alloc_bytes": 4800944436019, "free_bytes": 3200629624013, "cap_pct": 60, "health": "ONLINE" } ], "datasets": [ { "name": "data", "used_bytes": 4800944436019, "avail_bytes": 3200629624013, "refer_bytes": 98304, "mountpoint": "/mnt/data", "type": "filesystem" } ] }, "feature_flags": { "has_lvm": False, "has_zfs": True, "has_lsblk": True, "has_findmnt": True } } @pytest.fixture def linux_lvm_storage_details() -> Dict[str, Any]: """Sample storage_details from a Linux host with LVM.""" return { "os_type": "linux-debian", "status": "ok", "collected_at": "2025-12-19T10:00:00Z", "commands_run": [ {"cmd": "lsblk", "status": "ok"}, {"cmd": "findmnt", "status": "ok"}, {"cmd": "df", "status": "ok"}, {"cmd": "lvm", "status": "ok"} ], "partial_failures": [], "summary": { "total_bytes": 1000204886016, "used_bytes": 400081954406, "free_bytes": 549062746931, "used_pct": 40.0 }, "block_devices": [ { "name": "sda", "type": "disk", "size": 1000204886016, "children": [ { "name": "sda1", "type": "part", "size": 536870912, "fstype": "vfat", "mountpoint": "/boot/efi" }, { "name": "sda2", "type": "part", "size": 999667015104, "fstype": "LVM2_member" } ] } ], "mounts": [], "filesystems": [ { "device": "/dev/mapper/vg0-root", "fstype": "ext4", "size_bytes": 107374182400, "used_bytes": 32212254720, "avail_bytes": 69793193984, "use_pct": 30, "mountpoint": "/" }, { "device": "/dev/mapper/vg0-home", "fstype": "ext4", "size_bytes": 429496729600, "used_bytes": 214748364800, "avail_bytes": 192673095680, "use_pct": 50, "mountpoint": "/home" }, { "device": "/dev/mapper/vg0-var", "fstype": "ext4", "size_bytes": 214748364800, "used_bytes": 107374182400, "avail_bytes": 96636764160, "use_pct": 50, "mountpoint": "/var" } ], "lvm": { "pvs": [ { "pv_name": "/dev/sda2", "vg_name": "vg0", "pv_size": "930.00g", "pv_free": "178.00g" } ], "vgs": [ { "vg_name": "vg0", "vg_size": "930.00g", "vg_free": "178.00g" } ], "lvs": [ { "lv_name": "root", "vg_name": "vg0", "lv_size": "100.00g" }, { "lv_name": "home", "vg_name": "vg0", "lv_size": "400.00g" }, { "lv_name": "var", "vg_name": "vg0", "lv_size": "200.00g" } ] }, "zfs": {"pools": [], "datasets": []}, "feature_flags": { "has_lvm": True, "has_zfs": False, "has_lsblk": True, "has_findmnt": True } } # ============================================================================= # Tests # ============================================================================= class TestStorageDetailsSchema: """Tests for storage_details schema validation.""" def test_host_metrics_create_with_storage_details(self, linux_debian_storage_details): """Test that HostMetricsCreate accepts storage_details.""" metrics = HostMetricsCreate( host_id="test-host", metric_type="storage_details", storage_details=linux_debian_storage_details ) assert metrics.host_id == "test-host" assert metrics.metric_type == "storage_details" assert metrics.storage_details is not None assert metrics.storage_details["os_type"] == "linux-debian" assert metrics.storage_details["status"] == "ok" def test_host_metrics_summary_includes_storage_details(self, truenas_core_storage_details): """Test that HostMetricsSummary can include storage_details.""" summary = HostMetricsSummary( host_id="truenas-host", storage_details=truenas_core_storage_details, collection_status="success" ) assert summary.storage_details is not None assert summary.storage_details["feature_flags"]["has_zfs"] is True class TestStorageDetailsOsTypes: """Tests for different OS type storage details.""" def test_linux_debian_has_lsblk_and_findmnt(self, linux_debian_storage_details): """Test that Debian Linux has lsblk and findmnt data.""" flags = linux_debian_storage_details["feature_flags"] assert flags["has_lsblk"] is True assert flags["has_findmnt"] is True assert flags["has_lvm"] is False assert flags["has_zfs"] is False assert len(linux_debian_storage_details["block_devices"]) > 0 assert len(linux_debian_storage_details["mounts"]) > 0 def test_linux_alpine_partial_status(self, linux_alpine_storage_details): """Test that Alpine Linux with partial failures has correct status.""" assert linux_alpine_storage_details["status"] == "partial" assert "findmnt parse failed" in linux_alpine_storage_details["partial_failures"] assert linux_alpine_storage_details["feature_flags"]["has_findmnt"] is False def test_truenas_core_has_zfs_no_lsblk(self, truenas_core_storage_details): """Test that TrueNAS CORE has ZFS but no lsblk.""" flags = truenas_core_storage_details["feature_flags"] assert flags["has_zfs"] is True assert flags["has_lsblk"] is False assert flags["has_findmnt"] is False zfs = truenas_core_storage_details["zfs"] assert len(zfs["pools"]) == 1 assert zfs["pools"][0]["name"] == "tank" assert zfs["pools"][0]["health"] == "ONLINE" assert len(zfs["datasets"]) == 3 def test_truenas_scale_has_both_zfs_and_lsblk(self, truenas_scale_storage_details): """Test that TrueNAS SCALE has both ZFS and lsblk.""" flags = truenas_scale_storage_details["feature_flags"] assert flags["has_zfs"] is True assert flags["has_lsblk"] is True assert flags["has_findmnt"] is True assert len(truenas_scale_storage_details["block_devices"]) == 2 assert len(truenas_scale_storage_details["zfs"]["pools"]) == 2 def test_linux_lvm_has_volume_groups(self, linux_lvm_storage_details): """Test that Linux with LVM has volume groups.""" flags = linux_lvm_storage_details["feature_flags"] assert flags["has_lvm"] is True assert flags["has_zfs"] is False lvm = linux_lvm_storage_details["lvm"] assert len(lvm["pvs"]) == 1 assert len(lvm["vgs"]) == 1 assert len(lvm["lvs"]) == 3 assert lvm["vgs"][0]["vg_name"] == "vg0" class TestStorageDetailsSummary: """Tests for storage summary calculations.""" def test_summary_total_bytes(self, linux_debian_storage_details): """Test that summary contains correct total bytes.""" summary = linux_debian_storage_details["summary"] assert summary["total_bytes"] == 500107862016 assert summary["used_bytes"] == 125026965504 assert summary["free_bytes"] == 349600112640 assert summary["used_pct"] == 25.0 def test_summary_used_percentage_calculation(self, truenas_core_storage_details): """Test that used percentage is calculated correctly.""" summary = truenas_core_storage_details["summary"] total = summary["total_bytes"] used = summary["used_bytes"] # Check that used_pct is approximately correct expected_pct = (used / total) * 100 assert abs(summary["used_pct"] - expected_pct) < 1.0 class TestStorageDetailsFilesystems: """Tests for filesystem data parsing.""" def test_filesystem_has_required_fields(self, linux_debian_storage_details): """Test that filesystems have required fields.""" fs = linux_debian_storage_details["filesystems"][0] required_fields = ["device", "fstype", "size_bytes", "used_bytes", "avail_bytes", "use_pct", "mountpoint"] for field in required_fields: assert field in fs, f"Missing field: {field}" def test_filesystem_sizes_in_bytes(self, linux_debian_storage_details): """Test that filesystem sizes are in bytes.""" fs = linux_debian_storage_details["filesystems"][0] # Size should be larger than 1 GB (in bytes) assert fs["size_bytes"] > 1024 * 1024 * 1024 # Used + available should approximately equal total total = fs["size_bytes"] used = fs["used_bytes"] avail = fs["avail_bytes"] # Allow for some filesystem overhead assert abs((used + avail) - total) < total * 0.1 class TestStorageDetailsZfs: """Tests for ZFS data parsing.""" def test_zfs_pool_has_required_fields(self, truenas_core_storage_details): """Test that ZFS pools have required fields.""" pool = truenas_core_storage_details["zfs"]["pools"][0] required_fields = ["name", "size_bytes", "alloc_bytes", "free_bytes", "cap_pct", "health"] for field in required_fields: assert field in pool, f"Missing field: {field}" def test_zfs_dataset_has_required_fields(self, truenas_core_storage_details): """Test that ZFS datasets have required fields.""" ds = truenas_core_storage_details["zfs"]["datasets"][0] required_fields = ["name", "used_bytes", "avail_bytes", "refer_bytes", "mountpoint", "type"] for field in required_fields: assert field in ds, f"Missing field: {field}" def test_zfs_pool_health_status(self, truenas_core_storage_details): """Test ZFS pool health status.""" pool = truenas_core_storage_details["zfs"]["pools"][0] assert pool["health"] in ["ONLINE", "DEGRADED", "FAULTED", "OFFLINE", "UNAVAIL", "REMOVED"] class TestStorageDetailsCommands: """Tests for command execution metadata.""" def test_commands_run_list(self, linux_debian_storage_details): """Test that commands_run contains expected commands.""" commands = linux_debian_storage_details["commands_run"] cmd_names = [c["cmd"] for c in commands] assert "lsblk" in cmd_names assert "df" in cmd_names def test_command_status_is_ok(self, truenas_core_storage_details): """Test that successful commands have status 'ok'.""" for cmd in truenas_core_storage_details["commands_run"]: assert cmd["status"] == "ok" def test_partial_failures_recorded(self, linux_alpine_storage_details): """Test that partial failures are recorded.""" assert len(linux_alpine_storage_details["partial_failures"]) > 0 assert linux_alpine_storage_details["status"] == "partial" class TestStorageDetailsWarningThresholds: """Tests for warning threshold detection.""" def test_high_usage_filesystem_detection(self): """Test detection of filesystems with high usage.""" storage_details = { "filesystems": [ {"mountpoint": "/", "use_pct": 50}, {"mountpoint": "/var", "use_pct": 85}, # Warning {"mountpoint": "/home", "use_pct": 95}, # Critical ] } warning_mounts = [ fs["mountpoint"] for fs in storage_details["filesystems"] if fs["use_pct"] >= 85 ] assert "/var" in warning_mounts assert "/home" in warning_mounts assert "/" not in warning_mounts def test_critical_usage_detection(self): """Test detection of critical usage (>95%).""" storage_details = { "filesystems": [ {"mountpoint": "/", "use_pct": 96}, ] } critical_mounts = [ fs["mountpoint"] for fs in storage_details["filesystems"] if fs["use_pct"] >= 95 ] assert "/" in critical_mounts