Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 20 additions & 13 deletions coriolis/osmorphing/osmount/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -239,8 +239,9 @@ def _check_mount_fstab_partitions(
self, os_root_dir, skip_mounts=["/", "/boot"],
skip_filesystems=["swap"], mountable_lvm_devs=None):
""" Reads the contents of /etc/fstab from the VM's root directory and
tries to mount all clearly identified (by UUID or path) filesystems.
Returns the list of the new directories which were mounted.
tries to mount all clearly identified (by UUID, label, or path)
filesystems. Returns the list of the new directories which were
mounted.
param: skip_mounts: list(str()): list of directories (inside the
chroot) to not try to mount.
param: skip_filesystems: list(str()): list of filesystem types to skip
Expand Down Expand Up @@ -304,6 +305,7 @@ def _check_mount_fstab_partitions(

# regex for supported fstab device references:
fs_uuid_entry_regex = "^((UUID=|/dev/disk/by-uuid/)(.+))$"
fs_label_entry_regex = "^((LABEL=|/dev/disk/by-label/)(.+))$"
if not mountable_lvm_devs:
mountable_lvm_devs = []
device_paths = self._get_device_file_paths(mountable_lvm_devs)
Expand All @@ -313,17 +315,22 @@ def _check_mount_fstab_partitions(
if fs_uuid_match:
device = "/dev/disk/by-uuid/%s" % fs_uuid_match.group(3)
else:
device_file_path = self._get_symlink_target(device)
if device not in mountable_lvm_devs and (
device_file_path not in device_paths):
LOG.warn(
"Found fstab entry for dir %s which references device "
"%s. Only LVM volumes or devices referenced by UUID=* "
"or /dev/disk/by-uuid/* notation are supported. "
"Devicemapper paths for LVM volumes are also "
"supported. Skipping mounting directory." %
(mountpoint, device))
continue
fs_label_match = re.match(fs_label_entry_regex, device)
if fs_label_match:
device = "/dev/disk/by-label/%s" % fs_label_match.group(3)
else:
device_file_path = self._get_symlink_target(device)
if device not in mountable_lvm_devs and (
device_file_path not in device_paths):
LOG.warn(
"Found fstab entry for dir %s which references "
"device %s. Only LVM volumes or devices "
"referenced by UUID=*, /dev/disk/by-uuid/*, "
"LABEL=*, or /dev/disk/by-label/* notation are "
"supported. Devicemapper paths for LVM volumes "
"are also supported. Skipping mounting directory."
% (mountpoint, device))
continue
if mountpoint in skip_mounts:
LOG.debug(
"Skipping undesired mount: %s: %s", mountpoint, details)
Expand Down
37 changes: 37 additions & 0 deletions coriolis/tests/osmorphing/osmount/test_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -346,6 +346,43 @@ def test__check_mount_fstab_partitions(self, mock_get_device_file_paths,
self.base_os_mount_tools._ssh, "/dev/disk/by-uuid/uuid2")
])

@mock.patch.object(base.utils, 'test_ssh_path')
@mock.patch.object(base.utils, 'read_ssh_file')
@mock.patch.object(base.BaseSSHOSMountTools, '_exec_cmd')
@mock.patch.object(base.BaseLinuxOSMountTools, '_get_device_file_paths')
def test__check_mount_fstab_partitions_with_label(
self, mock_get_device_file_paths, mock_exec_cmd,
mock_read_ssh_file, mock_test_ssh_path):
mocked_full_path = self.os_root_dir + "/etc/fstab"
mock_test_ssh_path.return_value = True
mock_get_device_file_paths.return_value = ["/dev/sda1", "/dev/sda2"]
mock_exec_cmd.return_value = (
b"/dev/sda1:LABEL=boot\n/dev/sda2:LABEL=UEFI"
)
mock_read_ssh_file.return_value = (
b"LABEL=UEFI /boot/efi vfat defaults 0 0"
)

result = self.base_os_mount_tools._check_mount_fstab_partitions(
self.os_root_dir, mountable_lvm_devs=["/dev/sda1"])

expected_result = [self.os_root_dir + "/boot/efi"]

self.assertEqual(result, expected_result)

mock_get_device_file_paths.assert_called_once()
mock_exec_cmd.assert_called_once_with(
"sudo mount -t vfat -o defaults"
" /dev/disk/by-label/UEFI '/root/boot/efi'"
)
mock_read_ssh_file.assert_called_once_with(
self.base_os_mount_tools._ssh, mocked_full_path)
mock_test_ssh_path.assert_has_calls([
mock.call(self.base_os_mount_tools._ssh, mocked_full_path),
mock.call(
self.base_os_mount_tools._ssh, "/dev/disk/by-label/UEFI")
])

@mock.patch.object(base.utils, 'test_ssh_path')
def test__check_mount_fstab_partitions_no_fstab(self, mock_test_ssh_path):
mock_test_ssh_path.return_value = False
Expand Down
Loading