mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 19:01:39 -04:00
529 lines
22 KiB
Python
529 lines
22 KiB
Python
"""
|
|
Unit tests for controller.lib.disks module.
|
|
"""
|
|
|
|
import json
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from controller.lib.disks import (
|
|
BlockDevice,
|
|
check_shadowed_data,
|
|
get_block_devices,
|
|
get_uuid,
|
|
get_wrolpi_uid_gid,
|
|
validate_mount_point,
|
|
mount_drive,
|
|
unmount_drive,
|
|
is_mount_busy,
|
|
get_mounts,
|
|
)
|
|
|
|
|
|
class TestBlockDevice:
|
|
"""Tests for BlockDevice dataclass."""
|
|
|
|
def test_create_block_device(self):
|
|
"""Should create a BlockDevice instance."""
|
|
device = BlockDevice(
|
|
name="sda1",
|
|
path="/dev/sda1",
|
|
size="100G",
|
|
fstype="ext4",
|
|
mountpoint="/media/data",
|
|
label="DATA",
|
|
uuid="1234-5678",
|
|
model="Samsung SSD",
|
|
)
|
|
assert device.name == "sda1"
|
|
assert device.path == "/dev/sda1"
|
|
assert device.is_wrolpi_drive is False
|
|
|
|
def test_block_device_defaults(self):
|
|
"""Should have default values."""
|
|
device = BlockDevice(
|
|
name="sda1",
|
|
path="/dev/sda1",
|
|
size="100G",
|
|
fstype=None,
|
|
mountpoint=None,
|
|
label=None,
|
|
uuid=None,
|
|
model=None,
|
|
)
|
|
assert device.is_wrolpi_drive is False
|
|
|
|
|
|
class TestGetBlockDevices:
|
|
"""Tests for get_block_devices function."""
|
|
|
|
def test_returns_empty_in_docker_mode(self):
|
|
"""Should return empty list in Docker mode."""
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=True):
|
|
result = get_block_devices()
|
|
assert result == []
|
|
|
|
def test_returns_devices_from_lsblk(self):
|
|
"""Should parse lsblk output and return devices."""
|
|
lsblk_output = json.dumps({
|
|
"blockdevices": [
|
|
{
|
|
"name": "sda",
|
|
"path": "/dev/sda",
|
|
"size": "500G",
|
|
"type": "disk",
|
|
"model": "Samsung SSD",
|
|
"children": [
|
|
{
|
|
"name": "sda1",
|
|
"path": "/dev/sda1",
|
|
"size": "500G",
|
|
"type": "part",
|
|
"fstype": "ext4",
|
|
"mountpoint": "/media/data",
|
|
"label": "DATA",
|
|
"uuid": "1234-5678",
|
|
}
|
|
]
|
|
}
|
|
]
|
|
})
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=0, stdout=lsblk_output)
|
|
result = get_block_devices()
|
|
assert len(result) == 1
|
|
assert result[0].name == "sda1"
|
|
assert result[0].fstype == "ext4"
|
|
assert result[0].model == "Samsung SSD"
|
|
|
|
def test_filters_unsupported_filesystems(self):
|
|
"""Should only return devices with supported filesystems."""
|
|
lsblk_output = json.dumps({
|
|
"blockdevices": [
|
|
{
|
|
"name": "sda",
|
|
"type": "disk",
|
|
"children": [
|
|
{"name": "sda1", "path": "/dev/sda1", "size": "100G", "type": "part", "fstype": "ext4"},
|
|
{"name": "sda2", "path": "/dev/sda2", "size": "100G", "type": "part", "fstype": "xfs"},
|
|
]
|
|
}
|
|
]
|
|
})
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=0, stdout=lsblk_output)
|
|
result = get_block_devices()
|
|
assert len(result) == 1
|
|
assert result[0].name == "sda1"
|
|
|
|
def test_returns_whole_disk_without_partitions(self):
|
|
"""Should return a disk device that has a supported filesystem but no partitions."""
|
|
lsblk_output = json.dumps({
|
|
"blockdevices": [
|
|
{
|
|
"name": "sda",
|
|
"path": "/dev/sda",
|
|
"size": "1.8T",
|
|
"type": "disk",
|
|
"fstype": "ext4",
|
|
"mountpoint": "/media/pi/crucialx9",
|
|
"label": "crucialx9",
|
|
"uuid": "abcd-1234",
|
|
"model": "CT2000X9SSD9",
|
|
}
|
|
]
|
|
})
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=0, stdout=lsblk_output)
|
|
result = get_block_devices()
|
|
assert len(result) == 1
|
|
assert result[0].name == "sda"
|
|
assert result[0].fstype == "ext4"
|
|
assert result[0].model == "CT2000X9SSD9"
|
|
|
|
def test_does_not_return_parent_disk_when_it_has_partitions(self):
|
|
"""Should not include a parent disk that has children partitions."""
|
|
lsblk_output = json.dumps({
|
|
"blockdevices": [
|
|
{
|
|
"name": "sda",
|
|
"path": "/dev/sda",
|
|
"size": "500G",
|
|
"type": "disk",
|
|
"fstype": "ext4",
|
|
"model": "Samsung SSD",
|
|
"children": [
|
|
{
|
|
"name": "sda1",
|
|
"path": "/dev/sda1",
|
|
"size": "500G",
|
|
"type": "part",
|
|
"fstype": "ext4",
|
|
"mountpoint": "/media/data",
|
|
"label": "DATA",
|
|
"uuid": "1234-5678",
|
|
}
|
|
]
|
|
}
|
|
]
|
|
})
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=0, stdout=lsblk_output)
|
|
result = get_block_devices()
|
|
assert len(result) == 1
|
|
assert result[0].name == "sda1" # Only the partition, not the parent disk
|
|
|
|
def test_handles_lsblk_failure(self):
|
|
"""Should return empty list on lsblk failure."""
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=1, stdout="", stderr="error")
|
|
result = get_block_devices()
|
|
assert result == []
|
|
|
|
|
|
class TestGetUuid:
|
|
"""Tests for get_uuid function."""
|
|
|
|
def test_returns_uuid_from_blkid(self):
|
|
"""Should return UUID from blkid output."""
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=0, stdout="1234-5678-90ab\n")
|
|
result = get_uuid("/dev/sda1")
|
|
assert result == "1234-5678-90ab"
|
|
|
|
def test_returns_none_on_failure(self):
|
|
"""Should return None if blkid fails."""
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=1, stdout="")
|
|
result = get_uuid("/dev/sda1")
|
|
assert result is None
|
|
|
|
|
|
class TestValidateMountPoint:
|
|
"""Tests for validate_mount_point function."""
|
|
|
|
def test_accepts_media_path(self):
|
|
"""Should accept paths under /media."""
|
|
validate_mount_point("/media/wrolpi")
|
|
validate_mount_point("/media/usb")
|
|
validate_mount_point("/media/wrolpi/data")
|
|
|
|
def test_rejects_non_media_paths(self):
|
|
"""Should reject paths not under /media."""
|
|
with pytest.raises(ValueError, match="must be under /media"):
|
|
validate_mount_point("/mnt/data")
|
|
|
|
with pytest.raises(ValueError, match="must be under /media"):
|
|
validate_mount_point("/home/user")
|
|
|
|
with pytest.raises(ValueError, match="must be under /media"):
|
|
validate_mount_point("/")
|
|
|
|
def test_rejects_path_traversal(self):
|
|
"""Should reject path traversal attempts."""
|
|
with pytest.raises(ValueError):
|
|
validate_mount_point("/media/../etc")
|
|
|
|
|
|
class TestMountDrive:
|
|
"""Tests for mount_drive function."""
|
|
|
|
def test_returns_error_in_docker_mode(self):
|
|
"""Should return error in Docker mode."""
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=True):
|
|
result = mount_drive("/dev/sda1", "/media/test")
|
|
assert result["success"] is False
|
|
assert "Docker" in result["error"]
|
|
|
|
def test_validates_mount_point(self):
|
|
"""Should validate mount point."""
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
result = mount_drive("/dev/sda1", "/mnt/test")
|
|
assert result["success"] is False
|
|
assert "must be under /media" in result["error"]
|
|
|
|
def test_mounts_drive_successfully(self):
|
|
"""Should mount drive successfully."""
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
with mock.patch("pathlib.Path.mkdir"):
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=0, stdout="", stderr="")
|
|
result = mount_drive("/dev/sda1", "/media/test")
|
|
assert result["success"] is True
|
|
assert result["mount_point"] == "/media/test"
|
|
|
|
def test_handles_mount_failure(self):
|
|
"""Should handle mount failure."""
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
with mock.patch("pathlib.Path.mkdir"):
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=1, stdout="", stderr="mount failed")
|
|
result = mount_drive("/dev/sda1", "/media/test")
|
|
assert result["success"] is False
|
|
assert "mount failed" in result["error"]
|
|
|
|
def test_exfat_mount_includes_uid_gid(self):
|
|
"""Should add uid/gid options when mounting exfat."""
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
with mock.patch("controller.lib.disks.get_wrolpi_uid_gid", return_value=(1001, 1001)):
|
|
with mock.patch("pathlib.Path.mkdir"):
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=0, stdout="", stderr="")
|
|
result = mount_drive("/dev/sda1", "/media/test", fstype="exfat")
|
|
assert result["success"] is True
|
|
# Verify mount was called with uid/gid options
|
|
call_args = mock_run.call_args[0][0]
|
|
assert "-o" in call_args
|
|
options_idx = call_args.index("-o") + 1
|
|
options = call_args[options_idx]
|
|
assert "uid=1001" in options
|
|
assert "gid=1001" in options
|
|
|
|
def test_ext4_mount_does_not_include_uid_gid(self):
|
|
"""Should not add uid/gid options when mounting ext4."""
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
with mock.patch("pathlib.Path.mkdir"):
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=0, stdout="", stderr="")
|
|
result = mount_drive("/dev/sda1", "/media/test", fstype="ext4")
|
|
assert result["success"] is True
|
|
# Verify mount was called without uid/gid options
|
|
call_args = mock_run.call_args[0][0]
|
|
assert "-o" in call_args
|
|
options_idx = call_args.index("-o") + 1
|
|
options = call_args[options_idx]
|
|
assert "uid=" not in options
|
|
assert "gid=" not in options
|
|
|
|
def test_vfat_mount_includes_uid_gid(self):
|
|
"""Should add uid/gid options when mounting vfat (FAT32)."""
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
with mock.patch("controller.lib.disks.get_wrolpi_uid_gid", return_value=(1001, 1001)):
|
|
with mock.patch("pathlib.Path.mkdir"):
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=0, stdout="", stderr="")
|
|
result = mount_drive("/dev/sda1", "/media/test", fstype="vfat")
|
|
assert result["success"] is True
|
|
# Verify mount was called with uid/gid options
|
|
call_args = mock_run.call_args[0][0]
|
|
assert "-o" in call_args
|
|
options_idx = call_args.index("-o") + 1
|
|
options = call_args[options_idx]
|
|
assert "uid=1001" in options
|
|
assert "gid=1001" in options
|
|
|
|
def test_ntfs_mount_includes_uid_gid(self):
|
|
"""Should add uid/gid options when mounting ntfs (Windows-formatted drive)."""
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
with mock.patch("controller.lib.disks.get_wrolpi_uid_gid", return_value=(1001, 1001)):
|
|
with mock.patch("pathlib.Path.mkdir"):
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=0, stdout="", stderr="")
|
|
result = mount_drive("/dev/sda1", "/media/test", fstype="ntfs")
|
|
assert result["success"] is True
|
|
call_args = mock_run.call_args[0][0]
|
|
options_idx = call_args.index("-o") + 1
|
|
options = call_args[options_idx]
|
|
assert "uid=1001" in options
|
|
assert "gid=1001" in options
|
|
|
|
def test_ntfs3_mount_includes_uid_gid(self):
|
|
"""Same handling for the in-kernel ntfs3 driver (kernel 5.15+)."""
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
with mock.patch("controller.lib.disks.get_wrolpi_uid_gid", return_value=(1001, 1001)):
|
|
with mock.patch("pathlib.Path.mkdir"):
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=0, stdout="", stderr="")
|
|
result = mount_drive("/dev/sda1", "/media/test", fstype="ntfs3")
|
|
assert result["success"] is True
|
|
call_args = mock_run.call_args[0][0]
|
|
options_idx = call_args.index("-o") + 1
|
|
options = call_args[options_idx]
|
|
assert "uid=1001" in options
|
|
assert "gid=1001" in options
|
|
|
|
|
|
class TestGetWrolpiUidGid:
|
|
"""Tests for get_wrolpi_uid_gid function."""
|
|
|
|
def test_returns_wrolpi_user_ids(self):
|
|
"""Should return uid/gid for wrolpi user when it exists."""
|
|
mock_pwd = mock.Mock()
|
|
mock_pwd.pw_uid = 1001
|
|
mock_grp = mock.Mock()
|
|
mock_grp.gr_gid = 1001
|
|
|
|
with mock.patch("controller.lib.disks.pwd.getpwnam", return_value=mock_pwd):
|
|
with mock.patch("controller.lib.disks.grp.getgrnam", return_value=mock_grp):
|
|
uid, gid = get_wrolpi_uid_gid()
|
|
assert uid == 1001
|
|
assert gid == 1001
|
|
|
|
def test_returns_fallback_when_user_not_found(self):
|
|
"""Should return 1001:1001 fallback when wrolpi user doesn't exist."""
|
|
with mock.patch("controller.lib.disks.pwd.getpwnam", side_effect=KeyError("wrolpi")):
|
|
uid, gid = get_wrolpi_uid_gid()
|
|
assert uid == 1001
|
|
assert gid == 1001
|
|
|
|
|
|
class TestUnmountDrive:
|
|
"""Tests for unmount_drive function."""
|
|
|
|
def test_returns_error_in_docker_mode(self):
|
|
"""Should return error in Docker mode."""
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=True):
|
|
result = unmount_drive("/media/test")
|
|
assert result["success"] is False
|
|
assert "Docker" in result["error"]
|
|
|
|
def test_unmounts_successfully(self):
|
|
"""Should unmount successfully."""
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
with mock.patch("controller.lib.disks.is_mount_busy", return_value=False):
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=0)
|
|
result = unmount_drive("/media/test")
|
|
assert result["success"] is True
|
|
|
|
def test_checks_if_mount_busy(self):
|
|
"""Should check if mount is busy."""
|
|
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False):
|
|
with mock.patch("controller.lib.disks.is_mount_busy", return_value=True):
|
|
result = unmount_drive("/media/test", lazy=False)
|
|
assert result["success"] is False
|
|
assert "busy" in result["error"]
|
|
|
|
|
|
class TestIsMountBusy:
|
|
"""Tests for is_mount_busy function."""
|
|
|
|
def test_returns_true_if_processes_using(self):
|
|
"""Should return True if processes are using mount."""
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=0, stdout="python /media/test/file.py")
|
|
result = is_mount_busy("/media/test")
|
|
# is_mount_busy returns truthy value (non-empty string) when busy
|
|
assert result
|
|
|
|
def test_returns_false_if_no_processes(self):
|
|
"""Should return False if no processes using mount."""
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=1, stdout="")
|
|
result = is_mount_busy("/media/test")
|
|
assert not result
|
|
|
|
|
|
class TestGetMounts:
|
|
"""Tests for get_mounts function."""
|
|
|
|
def test_returns_mounts_under_media(self):
|
|
"""Should return mounts under /media."""
|
|
findmnt_output = json.dumps({
|
|
"filesystems": [
|
|
{"target": "/", "source": "/dev/sda1", "fstype": "ext4", "options": "rw"},
|
|
{"target": "/media/wrolpi", "source": "/dev/sdb1", "fstype": "ext4", "options": "rw"},
|
|
{"target": "/media/usb", "source": "/dev/sdc1", "fstype": "vfat", "options": "rw"},
|
|
]
|
|
})
|
|
with mock.patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = mock.Mock(returncode=0, stdout=findmnt_output)
|
|
result = get_mounts()
|
|
assert len(result) == 2
|
|
assert result[0]["mount_point"] == "/media/wrolpi"
|
|
assert result[1]["mount_point"] == "/media/usb"
|
|
|
|
|
|
class TestCheckShadowedData:
|
|
"""Tests for check_shadowed_data."""
|
|
|
|
def test_returns_none_for_missing_target(self, tmp_path):
|
|
assert check_shadowed_data(str(tmp_path / "nope")) is None
|
|
|
|
def test_returns_none_for_empty_target(self, tmp_path):
|
|
target = tmp_path / "wrolpi"
|
|
target.mkdir()
|
|
assert check_shadowed_data(str(target)) is None
|
|
|
|
def test_returns_none_when_only_config_present(self, tmp_path):
|
|
"""`config/` is whitelisted — it's expected pre-mount state."""
|
|
target = tmp_path / "wrolpi"
|
|
target.mkdir()
|
|
(target / "config").mkdir()
|
|
(target / "config" / "ssl.crt").write_text("cert")
|
|
assert check_shadowed_data(str(target)) is None
|
|
|
|
def test_detects_directory_entry(self, tmp_path):
|
|
target = tmp_path / "wrolpi"
|
|
target.mkdir()
|
|
(target / "videos").mkdir()
|
|
(target / "videos" / "a.mp4").write_bytes(b"x" * 100)
|
|
result = check_shadowed_data(str(target))
|
|
assert result == {"size_bytes": 100, "entries": ["videos"]}
|
|
|
|
def test_detects_file_entry(self, tmp_path):
|
|
target = tmp_path / "wrolpi"
|
|
target.mkdir()
|
|
(target / "stray.txt").write_bytes(b"hello")
|
|
result = check_shadowed_data(str(target))
|
|
assert result == {"size_bytes": 5, "entries": ["stray.txt"]}
|
|
|
|
def test_sums_sizes_across_multiple_entries(self, tmp_path):
|
|
target = tmp_path / "wrolpi"
|
|
target.mkdir()
|
|
(target / "config").mkdir() # ignored
|
|
(target / "videos").mkdir()
|
|
(target / "videos" / "a.mp4").write_bytes(b"x" * 200)
|
|
(target / "videos" / "b.mp4").write_bytes(b"x" * 300)
|
|
(target / "zims").mkdir()
|
|
(target / "zims" / "z.zim").write_bytes(b"x" * 500)
|
|
result = check_shadowed_data(str(target))
|
|
assert result["size_bytes"] == 1000
|
|
assert result["entries"] == ["videos", "zims"]
|
|
|
|
def test_does_not_follow_symlinks_to_directories(self, tmp_path):
|
|
"""A symlink at the mount target must not have its target walked.
|
|
|
|
Regression: previously a symlink-to-dir was detected by is_dir()
|
|
(which follows symlinks), then rglob walked the target — so
|
|
`ln -s /etc /media/wrolpi/x` would sum the size of /etc.
|
|
"""
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
# A few large files outside the mount target.
|
|
(outside / "a").write_bytes(b"x" * 10_000)
|
|
(outside / "b").write_bytes(b"x" * 20_000)
|
|
|
|
target = tmp_path / "wrolpi"
|
|
target.mkdir()
|
|
(target / "videos").mkdir()
|
|
(target / "videos" / "movie.mp4").write_bytes(b"x" * 100)
|
|
# Symlink pointing outside the mount target.
|
|
(target / "external").symlink_to(outside)
|
|
|
|
result = check_shadowed_data(str(target))
|
|
# `external` is listed (the user should know it's there) but it
|
|
# contributes 0 to the size — only the real `videos` data counts.
|
|
assert result["entries"] == ["external", "videos"]
|
|
assert result["size_bytes"] == 100
|
|
|
|
def test_does_not_follow_symlinks_to_files(self, tmp_path):
|
|
"""A symlink-to-file is listed but contributes 0 bytes."""
|
|
outside_file = tmp_path / "big.bin"
|
|
outside_file.write_bytes(b"x" * 50_000)
|
|
|
|
target = tmp_path / "wrolpi"
|
|
target.mkdir()
|
|
(target / "link.bin").symlink_to(outside_file)
|
|
|
|
result = check_shadowed_data(str(target))
|
|
assert result["entries"] == ["link.bin"]
|
|
assert result["size_bytes"] == 0
|
|
|
|
|