# Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import functools import logging import os from unittest import mock import diskimage_builder.block_device.tests.test_config as tc from diskimage_builder.block_device.config import config_tree_to_graph from diskimage_builder.block_device.config import create_graph from diskimage_builder.block_device.level2.mkfs import FilesystemNode from diskimage_builder.block_device.level3.mount import cmp_mount_order from diskimage_builder.block_device.level3.mount import MountPointNode from diskimage_builder.block_device.tests.test_base import TestBase logger = logging.getLogger(__name__) class TestMountComparator(TestBase): def test_mount_comparator(self): # This tests cmp_mount_order to ensure it sorts in the # expected order. The comparator takes a tuple of # (mount_point, node_name) but we can ignore the name partitions = [ ('/var/log', 'fake_log'), ('/boot', 'fake_boot'), ('/', 'fake_name'), ('/var', 'fake_name')] partitions.sort(key=functools.cmp_to_key(cmp_mount_order)) res = list(x[0] for x in partitions) # "/" must be first self.assertEqual(res[0], '/') # /var before /var/log var_pos = res.index('/var') var_log_pos = res.index('/var/log') self.assertGreater(var_log_pos, var_pos) class TestMountOrder(tc.TestGraphGeneration): def _exec_sudo_log(*args, **kwargs): # Used as a side-effect from exec_sudo mocking so we can see # the call in-place in logs logger.debug("exec_sudo: %s", " ".join(args[0])) @mock.patch('diskimage_builder.block_device.level3.mount.exec_sudo', side_effect=_exec_sudo_log) @mock.patch('diskimage_builder.block_device.level2.mkfs.exec_sudo', side_effect=_exec_sudo_log) def test_mfks_and_mount_order(self, mock_exec_sudo_mkfs, mock_exec_sudo_mount): # XXX: better mocking for the os.path.exists calls to avoid # failing if this exists. self.assertFalse(os.path.exists('/fake/')) # This is probably in order after graph creation, so ensure it # remains stable. We test the mount and umount call sequences config = self.load_config_file('multiple_partitions_graph.yaml') state = {} graph, call_order = create_graph(config, self.fake_default_config, state) # Mocked block device state state['blockdev'] = {} state['blockdev']['root'] = {'device': '/dev/loopXp1/root'} state['blockdev']['var'] = {'device': '/dev/loopXp2/var'} state['blockdev']['var_log'] = {'device': '/dev/loopXp3/var_log'} state['blockdev']['swap'] = {'device': '/dev/loopXp4/swap'} for node in call_order: if isinstance(node, (FilesystemNode, MountPointNode)): node.create() for node in reversed(call_order): if isinstance(node, (FilesystemNode, MountPointNode)): node.umount() # ensure that partitions were mounted in order root->var->var/log self.assertListEqual(state['mount_order'], ['/', '/var', '/var/log', 'none']) # fs creation sequence (note we don't care about order of this # as they're all independent) cmd_sequence = [ mock.call(['mkfs', '-t', 'xfs', '-L', 'mkfs_root', '-m', 'uuid=root-uuid-1234', '-q', '/dev/loopXp1/root']), mock.call(['mkfs', '-t', 'xfs', '-L', 'mkfs_var', '-m', 'uuid=var-uuid-1234', '-q', '/dev/loopXp2/var']), mock.call(['mkfs', '-t', 'vfat', '-n', 'VARLOG', '/dev/loopXp3/var_log']), mock.call(['mkswap', '-L', 'mkfs_swap', '-U', 'swap-uuid-1234', '/dev/loopXp4/swap']) ] self.assertEqual(mock_exec_sudo_mkfs.call_count, len(cmd_sequence)) mock_exec_sudo_mkfs.assert_has_calls(cmd_sequence, any_order=True) # Check mount sequence cmd_sequence = [ # mount sequence mock.call(['mkdir', '-p', '/fake/']), mock.call(['mount', '/dev/loopXp1/root', '/fake/']), mock.call(['mkdir', '-p', '/fake/var']), mock.call(['mount', '/dev/loopXp2/var', '/fake/var']), mock.call(['mkdir', '-p', '/fake/var/log']), mock.call(['mount', '/dev/loopXp3/var_log', '/fake/var/log']), # umount sequence mock.call(['sync']), # note /fake/var/log is a vfs partition to make sure # we don't try to fstrim it mock.call(['umount', '/fake/var/log']), mock.call(['sync']), mock.call(['fstrim', '--verbose', '/fake/var']), mock.call(['umount', '/fake/var']), mock.call(['sync']), mock.call(['fstrim', '--verbose', '/fake/']), mock.call(['umount', '/fake/']) ] self.assertListEqual(mock_exec_sudo_mount.call_args_list, cmd_sequence) @mock.patch('diskimage_builder.block_device.level3.mount.exec_sudo', side_effect=_exec_sudo_log) def test_mount_order_unsorted(self, mock_exec_sudo): # As above, but this is out of order and gets sorted # so that root is mounted first (and skips the mkfs testing). config = self.load_config_file('lvm_tree_partition_ordering.yaml') parsed_graph = config_tree_to_graph(config) state = {} graph, call_order = create_graph(parsed_graph, self.fake_default_config, state) state['filesys'] = { 'mkfs_root': { 'device': '/dev/loopXp1', 'fstype': 'xfs' }, 'mkfs_var': { 'device': '/dev/loopXp2', 'fstype': 'xfs', }, 'mkfs_boot': { 'device': '/dev/loopXp3', 'fstype': 'vfat', }, 'mkfs_swap': { 'device': '/dev/loopXp4', 'fstype': 'swap', }, } for node in call_order: if isinstance(node, MountPointNode): node.create() for node in reversed(call_order): if isinstance(node, MountPointNode): node.umount() # ensure that partitions are mounted in order / -> /boot -> /var # swap not mounted in runtime, but should be in 'mount_order' list self.assertListEqual(state['mount_order'], ['/', '/boot', '/var', 'none']) cmd_sequence = [ # mount sequence mock.call(['mkdir', '-p', '/fake/']), mock.call(['mount', '/dev/loopXp1', '/fake/']), mock.call(['mkdir', '-p', '/fake/boot']), mock.call(['mount', '/dev/loopXp3', '/fake/boot']), mock.call(['mkdir', '-p', '/fake/var']), mock.call(['mount', '/dev/loopXp2', '/fake/var']), # umount sequence mock.call(['sync']), mock.call(['fstrim', '--verbose', '/fake/var']), mock.call(['umount', '/fake/var']), mock.call(['sync']), # no trim on vfat /fake/boot mock.call(['umount', '/fake/boot']), mock.call(['sync']), mock.call(['fstrim', '--verbose', '/fake/']), mock.call(['umount', '/fake/']) ] self.assertListEqual(mock_exec_sudo.call_args_list, cmd_sequence)