wrolpi/controller/test/test_disks.py
lrnselfreliance 826f1cdc42
The drive wizard never offers an EFI system partition (#833)
On a Debian install the onboarding wizard offered the EFI system partition
(unlabeled vfat at /boot/efi) as the primary WROLPi drive; choosing it would
mount the boot partition over /media/wrolpi.  Candidates were filtered by
the "EFI" label only, and the Debian installer leaves the ESP unlabeled.

/boot/efi is now an excluded mountpoint, and partitions with the EFI system
partition type (GPT GUID or MBR 0xef, from lsblk PARTTYPE) are excluded even
when unmounted, e.g. on a second internal disk.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 16:26:18 -06:00

541 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_reads_partition_type(self):
"""The partition type identifies an EFI system partition whatever its label or mount."""
lsblk_output = json.dumps({"blockdevices": [{
"name": "sda", "path": "/dev/sda", "size": "500G", "type": "disk", "children": [{
"name": "sda1", "path": "/dev/sda1", "size": "512M", "type": "part", "fstype": "vfat",
"parttype": "c12a7328-f81f-11d2-ba4b-00a0c93ec93b",
}],
}]})
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 "PARTTYPE" in mock_run.call_args.args[0][-1].split(",")
assert result[0].parttype == "c12a7328-f81f-11d2-ba4b-00a0c93ec93b"
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_ntfs_mount_runs_ntfsfix_first(self):
"""NTFS mounts run ntfsfix -d before mount to clear the Windows dirty flag."""
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False), \
mock.patch("controller.lib.disks.get_wrolpi_uid_gid", return_value=(1001, 1001)), \
mock.patch("pathlib.Path.mkdir"), \
mock.patch("subprocess.run") as mock_run:
mock_run.return_value = mock.Mock(returncode=0, stdout="", stderr="")
result = mount_drive("/dev/sdb2", "/media/test", fstype="ntfs")
assert result["success"] is True
commands = [c[0][0] for c in mock_run.call_args_list]
assert commands[0][:2] == ["ntfsfix", "-d"]
assert "/dev/sdb2" in commands[0]
assert commands[1][0] == "mount"
def test_ntfsfix_failure_does_not_block_mount(self):
"""A failed ntfsfix is logged but the mount is still attempted."""
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False), \
mock.patch("controller.lib.disks.get_wrolpi_uid_gid", return_value=(1001, 1001)), \
mock.patch("pathlib.Path.mkdir"), \
mock.patch("subprocess.run") as mock_run:
mock_run.side_effect = [
mock.Mock(returncode=1, stdout="", stderr="ntfsfix failed"), # ntfsfix
mock.Mock(returncode=0, stdout="", stderr=""), # mount
]
result = mount_drive("/dev/sdb2", "/media/test", fstype="ntfs")
assert result["success"] is True
def test_ext4_mount_does_not_run_ntfsfix(self):
"""Non-NTFS mounts must not invoke ntfsfix."""
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False), \
mock.patch("pathlib.Path.mkdir"), \
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
commands = [c[0][0] for c in mock_run.call_args_list]
assert len(commands) == 1
assert commands[0][0] == "mount"
@pytest.mark.parametrize("fstype,expect_uid_gid", [
("exfat", True),
("vfat", True), # FAT32
("ntfs", True), # ntfs-3g userspace driver
("ntfs3", True), # in-kernel ntfs3 driver (kernel 5.15+)
("ext4", False), # POSIX permissions; no uid/gid mount opts
])
def test_mount_uid_gid_options_by_fstype(self, fstype, expect_uid_gid):
"""Foreign-fs mounts (exfat/vfat/ntfs/ntfs3) inject uid/gid; ext4 must not."""
with mock.patch("controller.lib.disks.is_docker_mode", return_value=False), \
mock.patch("controller.lib.disks.get_wrolpi_uid_gid", return_value=(1001, 1001)), \
mock.patch("pathlib.Path.mkdir"), \
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=fstype)
assert result["success"] is True
call_args = mock_run.call_args[0][0]
assert "-o" in call_args
options = call_args[call_args.index("-o") + 1]
if expect_uid_gid:
assert "uid=1001" in options
assert "gid=1001" in options
else:
assert "uid=" not in options
assert "gid=" not 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):
"""A `config/` holding only the CA from generate_certificates.sh is expected pre-mount state."""
target = tmp_path / "wrolpi"
target.mkdir()
(target / "config" / "ssl").mkdir(parents=True)
(target / "config" / "ssl" / "ca.crt").write_text("cert")
assert check_shadowed_data(str(target)) is None
def test_detects_config_with_user_data(self, tmp_path):
"""A library kept on the internal disk has its database and configs in `config/`."""
target = tmp_path / "wrolpi"
target.mkdir()
(target / "config" / "ssl").mkdir(parents=True)
(target / "config" / "ssl" / "ca.crt").write_text("cert")
(target / "config" / "wrolpi.db").write_bytes(b"x" * 10)
result = check_shadowed_data(str(target))
assert result["entries"] == ["config"]
assert result["size_bytes"] == 14
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