bf0dc265ae
With the removal of Python 2.x we can remove the unittest2 compat wrappers and switch to assertCountEqual instead of assertItemsEqual We have been able to use them since then, because testtools required unittest2, which still included it. With testtools removing Python 2.7 support [3][4], we will lose support for assertItemsEqual, so we should switch to use assertCountEqual. [1] - https://bugs.python.org/issue17866 [2] - https://hg.python.org/cpython/rev/d9921cb6e3cd [3] - testing-cabal/testtools#286 [4] - testing-cabal/testtools#277 Change-Id: I870286a2557e41099597c22dc9747743e1077615
579 lines
24 KiB
Python
579 lines
24 KiB
Python
# 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 copy
|
|
import logging
|
|
import mock
|
|
|
|
import diskimage_builder.block_device.tests.test_config as tc
|
|
|
|
from diskimage_builder.block_device.blockdevice import BlockDeviceState
|
|
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.exception import \
|
|
BlockDeviceSetupException
|
|
from diskimage_builder.block_device.level0.localloop import LocalLoopNode
|
|
from diskimage_builder.block_device.level1.lvm import LVMNode
|
|
from diskimage_builder.block_device.level1.lvm import LVMPlugin
|
|
from diskimage_builder.block_device.level1.lvm import LvsNode
|
|
from diskimage_builder.block_device.level1.lvm import PvsNode
|
|
from diskimage_builder.block_device.level1.lvm import VgsNode
|
|
from diskimage_builder.block_device.level1.partitioning import PartitionNode
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TestLVM(tc.TestGraphGeneration):
|
|
def test_lvm_tree_to_graph(self):
|
|
# equivalence of tree-based to graph-based config
|
|
tree = self.load_config_file('lvm_tree.yaml')
|
|
graph = self.load_config_file('lvm_graph.yaml')
|
|
parsed_graph = config_tree_to_graph(tree)
|
|
self.assertCountEqual(parsed_graph, graph)
|
|
|
|
def test_lvm_invalid_config(self):
|
|
# test some invalid config paths
|
|
config = self.load_config_file('lvm_graph.yaml')
|
|
lvm_config = config[2]['lvm']
|
|
|
|
bad_config = copy.deepcopy(lvm_config)
|
|
bad_config['vgs'][0]['base'] = ['invalid_pv']
|
|
self.assertRaisesRegex(BlockDeviceSetupException,
|
|
"base:invalid_pv in vgs does not match "
|
|
"a valid pvs",
|
|
LVMPlugin, bad_config, {}, {})
|
|
|
|
bad_config = copy.deepcopy(lvm_config)
|
|
bad_config['lvs'][0]['base'] = ['invalid_vg']
|
|
self.assertRaisesRegex(BlockDeviceSetupException,
|
|
r"base:\['invalid_vg'\] in lvs does not match "
|
|
r"a valid vg",
|
|
LVMPlugin, bad_config, {}, {})
|
|
|
|
bad_config = copy.deepcopy(lvm_config)
|
|
del(bad_config['lvs'][0]['size'])
|
|
self.assertRaisesRegex(BlockDeviceSetupException,
|
|
"Missing 'size' or 'extents' in lvs config",
|
|
LVMPlugin, bad_config, {}, {})
|
|
|
|
@mock.patch('diskimage_builder.block_device.level1.lvm.exec_sudo')
|
|
def test_lvm_multi_pv(self, mock_exec_sudo):
|
|
# Test the command-sequence for a more complicated LVM setup
|
|
tree = self.load_config_file('lvm_tree_multiple_pv.yaml')
|
|
config = config_tree_to_graph(tree)
|
|
|
|
state = BlockDeviceState()
|
|
|
|
graph, call_order = create_graph(config, self.fake_default_config,
|
|
state)
|
|
|
|
# XXX: todo; test call_order. Make sure PV's come before, VG;
|
|
# VG before LV, and that mounts/etc happen afterwards.
|
|
|
|
# Fake state for the two PV's specified by this config
|
|
state['blockdev'] = {}
|
|
state['blockdev']['root'] = {}
|
|
state['blockdev']['root']['device'] = '/dev/fake/root'
|
|
state['blockdev']['data'] = {}
|
|
state['blockdev']['data']['device'] = '/dev/fake/data'
|
|
|
|
for node in call_order:
|
|
# XXX: This has not mocked out the "lower" layers of
|
|
# creating the devices, which we're assuming works OK, nor
|
|
# the upper layers.
|
|
if isinstance(node, (LVMNode, PvsNode,
|
|
VgsNode, LvsNode)):
|
|
# only the LVMNode actually does anything here...
|
|
node.create()
|
|
|
|
# ensure the sequence of calls correctly setup the devices
|
|
cmd_sequence = [
|
|
# create the pv's on the faked out block devices
|
|
mock.call(['pvcreate', '/dev/fake/root', '--force']),
|
|
mock.call(['pvcreate', '/dev/fake/data', '--force']),
|
|
# create a volume called "vg" out of these two pv's
|
|
mock.call(['vgcreate', 'vg',
|
|
'/dev/fake/root', '/dev/fake/data', '--force']),
|
|
# create a bunch of lv's on vg
|
|
mock.call(['lvcreate', '--name', 'lv_root', '-L', '1800M', 'vg']),
|
|
mock.call(['lvcreate', '--name', 'lv_tmp', '-L', '100M', 'vg']),
|
|
mock.call(['lvcreate', '--name', 'lv_var', '-L', '500M', 'vg']),
|
|
mock.call(['lvcreate', '--name', 'lv_log', '-L', '100M', 'vg']),
|
|
mock.call(['lvcreate', '--name', 'lv_audit', '-L', '100M', 'vg']),
|
|
mock.call(['lvcreate', '--name', 'lv_home', '-L', '200M', 'vg'])]
|
|
|
|
self.assertEqual(mock_exec_sudo.call_count, len(cmd_sequence))
|
|
mock_exec_sudo.assert_has_calls(cmd_sequence)
|
|
|
|
# Ensure the correct LVM state was preserved
|
|
blockdev_state = {
|
|
'data': {'device': '/dev/fake/data'},
|
|
'root': {'device': '/dev/fake/root'},
|
|
'lv_audit': {
|
|
'device': '/dev/mapper/vg-lv_audit',
|
|
'extents': None,
|
|
'opts': None,
|
|
'size': '100M',
|
|
'vgs': 'vg'
|
|
},
|
|
'lv_home': {
|
|
'device': '/dev/mapper/vg-lv_home',
|
|
'extents': None,
|
|
'opts': None,
|
|
'size': '200M',
|
|
'vgs': 'vg'
|
|
},
|
|
'lv_log': {
|
|
'device': '/dev/mapper/vg-lv_log',
|
|
'extents': None,
|
|
'opts': None,
|
|
'size': '100M',
|
|
'vgs': 'vg'
|
|
},
|
|
'lv_root': {
|
|
'device': '/dev/mapper/vg-lv_root',
|
|
'extents': None,
|
|
'opts': None,
|
|
'size': '1800M',
|
|
'vgs': 'vg'
|
|
},
|
|
'lv_tmp': {
|
|
'device': '/dev/mapper/vg-lv_tmp',
|
|
'extents': None,
|
|
'opts': None,
|
|
'size': '100M',
|
|
'vgs': 'vg'
|
|
},
|
|
'lv_var': {
|
|
'device': '/dev/mapper/vg-lv_var',
|
|
'extents': None,
|
|
'opts': None,
|
|
'size': '500M',
|
|
'vgs': 'vg'
|
|
},
|
|
}
|
|
|
|
# state.debug_dump()
|
|
self.assertDictEqual(state['blockdev'], blockdev_state)
|
|
|
|
# XXX: mount ordering? fs creation?
|
|
|
|
def test_lvm_multi_pv_vg(self):
|
|
# Test the command-sequence for a more complicated LVM setup
|
|
tree = self.load_config_file('lvm_tree_multiple_pv_vg.yaml')
|
|
config = config_tree_to_graph(tree)
|
|
|
|
state = BlockDeviceState()
|
|
|
|
graph, call_order = create_graph(config, self.fake_default_config,
|
|
state)
|
|
|
|
# XXX: todo; test call_order. Make sure PV's come before, VG;
|
|
# VG before LV, and that mounts/etc happen afterwards.
|
|
|
|
# Fake state for the two PV's specified by this config
|
|
state['blockdev'] = {}
|
|
state['blockdev']['root'] = {}
|
|
state['blockdev']['root']['device'] = '/dev/fake/root'
|
|
state['blockdev']['data'] = {}
|
|
state['blockdev']['data']['device'] = '/dev/fake/data'
|
|
|
|
# We mock patch this ... it's just a little long!
|
|
exec_sudo = 'diskimage_builder.block_device.level1.lvm.exec_sudo'
|
|
|
|
#
|
|
# Creation test
|
|
#
|
|
with mock.patch(exec_sudo) as mock_exec_sudo:
|
|
|
|
for node in call_order:
|
|
# XXX: This has not mocked out the "lower" layers of
|
|
# creating the devices, which we're assuming works OK, nor
|
|
# the upper layers.
|
|
if isinstance(node, (LVMNode, PvsNode,
|
|
VgsNode, LvsNode)):
|
|
# only the PvsNode actually does anything here...
|
|
node.create()
|
|
|
|
# ensure the sequence of calls correctly setup the devices
|
|
cmd_sequence = [
|
|
# create the pv's on the faked out block devices
|
|
mock.call(['pvcreate', '/dev/fake/root', '--force']),
|
|
mock.call(['pvcreate', '/dev/fake/data', '--force']),
|
|
# create a volume called "vg" out of these two pv's
|
|
mock.call(['vgcreate', 'vg1',
|
|
'/dev/fake/root', '--force']),
|
|
mock.call(['vgcreate', 'vg2',
|
|
'/dev/fake/data', '--force']),
|
|
# create a bunch of lv's on vg
|
|
mock.call(['lvcreate', '--name', 'lv_root',
|
|
'-L', '1800M', 'vg1']),
|
|
mock.call(['lvcreate', '--name', 'lv_tmp',
|
|
'-L', '100M', 'vg1']),
|
|
mock.call(['lvcreate', '--name', 'lv_var',
|
|
'-L', '500M', 'vg2']),
|
|
mock.call(['lvcreate', '--name', 'lv_log',
|
|
'-L', '100M', 'vg2']),
|
|
mock.call(['lvcreate', '--name', 'lv_audit',
|
|
'-L', '100M', 'vg2']),
|
|
mock.call(['lvcreate', '--name', 'lv_home',
|
|
'-L', '200M', 'vg2'])]
|
|
|
|
self.assertListEqual(mock_exec_sudo.call_args_list,
|
|
cmd_sequence)
|
|
|
|
# Ensure the correct LVM state was preserved
|
|
blockdev_state = {
|
|
'data': {'device': '/dev/fake/data'},
|
|
'root': {'device': '/dev/fake/root'},
|
|
'lv_audit': {
|
|
'device': '/dev/mapper/vg2-lv_audit',
|
|
'extents': None,
|
|
'opts': None,
|
|
'size': '100M',
|
|
'vgs': 'vg2'
|
|
},
|
|
'lv_home': {
|
|
'device': '/dev/mapper/vg2-lv_home',
|
|
'extents': None,
|
|
'opts': None,
|
|
'size': '200M',
|
|
'vgs': 'vg2'
|
|
},
|
|
'lv_log': {
|
|
'device': '/dev/mapper/vg2-lv_log',
|
|
'extents': None,
|
|
'opts': None,
|
|
'size': '100M',
|
|
'vgs': 'vg2'
|
|
},
|
|
'lv_root': {
|
|
'device': '/dev/mapper/vg1-lv_root',
|
|
'extents': None,
|
|
'opts': None,
|
|
'size': '1800M',
|
|
'vgs': 'vg1'
|
|
},
|
|
'lv_tmp': {
|
|
'device': '/dev/mapper/vg1-lv_tmp',
|
|
'extents': None,
|
|
'opts': None,
|
|
'size': '100M',
|
|
'vgs': 'vg1'
|
|
},
|
|
'lv_var': {
|
|
'device': '/dev/mapper/vg2-lv_var',
|
|
'extents': None,
|
|
'opts': None,
|
|
'size': '500M',
|
|
'vgs': 'vg2'
|
|
},
|
|
}
|
|
|
|
# state.debug_dump()
|
|
self.assertDictEqual(state['blockdev'], blockdev_state)
|
|
|
|
#
|
|
# Umount test
|
|
#
|
|
with mock.patch(exec_sudo) as mock_exec_sudo, \
|
|
mock.patch('tempfile.NamedTemporaryFile') as mock_temp, \
|
|
mock.patch('os.unlink'):
|
|
|
|
# each call to tempfile.NamedTemporaryFile will return a
|
|
# new mock with a unique filename, which we store in
|
|
# tempfiles
|
|
tempfiles = []
|
|
|
|
def new_tempfile(*args, **kwargs):
|
|
n = '/tmp/files%s' % len(tempfiles)
|
|
# trap! note mock.Mock(name = n) doesn't work like you
|
|
# think it would, since mock has a name attribute.
|
|
# That's why we override it with the configure_mock
|
|
# (this is mentioned in mock documentation if you read
|
|
# it :)
|
|
r = mock.Mock()
|
|
r.configure_mock(name=n)
|
|
tempfiles.append(n)
|
|
return r
|
|
mock_temp.side_effect = new_tempfile
|
|
|
|
def run_it(phase):
|
|
reverse_order = reversed(call_order)
|
|
for node in reverse_order:
|
|
if isinstance(node, (LVMNode, PvsNode, VgsNode, LvsNode)):
|
|
getattr(node, phase)()
|
|
else:
|
|
logger.debug("Skipping node for test: %s", node)
|
|
|
|
run_it('umount')
|
|
run_it('cleanup')
|
|
|
|
cmd_sequence = [
|
|
# delete the lv's
|
|
mock.call(['lvchange', '-an', '/dev/vg1/lv_root']),
|
|
mock.call(['lvchange', '-an', '/dev/vg1/lv_tmp']),
|
|
mock.call(['lvchange', '-an', '/dev/vg2/lv_var']),
|
|
mock.call(['lvchange', '-an', '/dev/vg2/lv_log']),
|
|
mock.call(['lvchange', '-an', '/dev/vg2/lv_audit']),
|
|
mock.call(['lvchange', '-an', '/dev/vg2/lv_home']),
|
|
# delete the vg's
|
|
mock.call(['vgchange', '-an', 'vg1']),
|
|
mock.call(['vgchange', '-an', 'vg2']),
|
|
mock.call(['udevadm', 'settle']),
|
|
mock.call(['pvscan', '--cache']),
|
|
]
|
|
|
|
self.assertListEqual(mock_exec_sudo.call_args_list, cmd_sequence)
|
|
|
|
def test_lvm_spanned_vg(self):
|
|
|
|
# Test when a volume group spans some partitions
|
|
|
|
tree = self.load_config_file('lvm_tree_spanned_vg.yaml')
|
|
config = config_tree_to_graph(tree)
|
|
|
|
state = BlockDeviceState()
|
|
|
|
graph, call_order = create_graph(config, self.fake_default_config,
|
|
state)
|
|
|
|
# XXX: todo; test call_order. Make sure PV's come before, VG;
|
|
# VG before LV, and that mounts/etc happen afterwards.
|
|
|
|
# Fake state for the two PV's specified by this config
|
|
state['blockdev'] = {}
|
|
state['blockdev']['root'] = {}
|
|
state['blockdev']['root']['device'] = '/dev/fake/root'
|
|
state['blockdev']['data1'] = {}
|
|
state['blockdev']['data1']['device'] = '/dev/fake/data1'
|
|
state['blockdev']['data2'] = {}
|
|
state['blockdev']['data2']['device'] = '/dev/fake/data2'
|
|
|
|
# We mock patch this ... it's just a little long!
|
|
exec_sudo = 'diskimage_builder.block_device.level1.lvm.exec_sudo'
|
|
|
|
#
|
|
# Creation test
|
|
#
|
|
with mock.patch(exec_sudo) as mock_exec_sudo:
|
|
|
|
for node in call_order:
|
|
# XXX: This has not mocked out the "lower" layers of
|
|
# creating the devices, which we're assuming works OK, nor
|
|
# the upper layers.
|
|
if isinstance(node, (LVMNode, PvsNode, VgsNode, LvsNode)):
|
|
# only the LVMNode actually does anything here...
|
|
node.create()
|
|
|
|
# ensure the sequence of calls correctly setup the devices
|
|
cmd_sequence = [
|
|
# create the pv's on the faked out block devices
|
|
mock.call(['pvcreate', '/dev/fake/root', '--force']),
|
|
mock.call(['pvcreate', '/dev/fake/data1', '--force']),
|
|
mock.call(['pvcreate', '/dev/fake/data2', '--force']),
|
|
# create a root and a data volume, with the data volume
|
|
# spanning data1 & data2
|
|
mock.call(['vgcreate', 'vg_root',
|
|
'/dev/fake/root', '--force']),
|
|
mock.call(['vgcreate', 'vg_data',
|
|
'/dev/fake/data1', '/dev/fake/data2', '--force']),
|
|
# create root and data volume
|
|
mock.call(['lvcreate', '--name', 'lv_root',
|
|
'-L', '1800M', 'vg_root']),
|
|
mock.call(['lvcreate', '--name', 'lv_data',
|
|
'-L', '2G', 'vg_data'])
|
|
]
|
|
|
|
self.assertListEqual(mock_exec_sudo.call_args_list,
|
|
cmd_sequence)
|
|
|
|
with mock.patch(exec_sudo) as mock_exec_sudo, \
|
|
mock.patch('tempfile.NamedTemporaryFile') as mock_temp, \
|
|
mock.patch('os.unlink'):
|
|
|
|
# see above ...
|
|
tempfiles = []
|
|
|
|
def new_tempfile(*args, **kwargs):
|
|
n = '/tmp/files%s' % len(tempfiles)
|
|
r = mock.Mock()
|
|
r.configure_mock(name=n)
|
|
tempfiles.append(n)
|
|
return r
|
|
mock_temp.side_effect = new_tempfile
|
|
|
|
def run_it(phase):
|
|
reverse_order = reversed(call_order)
|
|
for node in reverse_order:
|
|
if isinstance(node, (LVMNode, PvsNode, VgsNode, LvsNode)):
|
|
getattr(node, phase)()
|
|
else:
|
|
logger.debug("Skipping node for test: %s", node)
|
|
|
|
run_it('umount')
|
|
run_it('cleanup')
|
|
|
|
cmd_sequence = [
|
|
# deactivate lv's
|
|
mock.call(['lvchange', '-an', '/dev/vg_root/lv_root']),
|
|
mock.call(['lvchange', '-an', '/dev/vg_data/lv_data']),
|
|
|
|
# deactivate vg's
|
|
mock.call(['vgchange', '-an', 'vg_root']),
|
|
mock.call(['vgchange', '-an', 'vg_data']),
|
|
|
|
mock.call(['udevadm', 'settle']),
|
|
mock.call(['pvscan', '--cache']),
|
|
]
|
|
|
|
self.assertListEqual(mock_exec_sudo.call_args_list, cmd_sequence)
|
|
|
|
def test_lvm_multiple_partitions(self):
|
|
# Test the command-sequence for several partitions, one containing
|
|
# volumes on it
|
|
tree = self.load_config_file('lvm_tree_multiple_partitions.yaml')
|
|
config = config_tree_to_graph(tree)
|
|
|
|
state = BlockDeviceState()
|
|
|
|
graph, call_order = create_graph(config, self.fake_default_config,
|
|
state)
|
|
|
|
# Fake state for the partitions on this config
|
|
state['blockdev'] = {}
|
|
state['blockdev']['image0'] = {}
|
|
state['blockdev']['image0']['device'] = '/dev/fake/image0'
|
|
state['blockdev']['image0']['image'] = 'image'
|
|
state['blockdev']['root'] = {}
|
|
state['blockdev']['root']['device'] = '/dev/fake/root'
|
|
state['blockdev']['ESP'] = {}
|
|
state['blockdev']['ESP']['device'] = '/dev/fake/ESP'
|
|
state['blockdev']['BSP'] = {}
|
|
state['blockdev']['BSP']['device'] = '/dev/fake/BSP'
|
|
|
|
#
|
|
# Creation test
|
|
#
|
|
|
|
# We mock out the following exec_sudo and other related calls
|
|
# calls for the layers we are testing.
|
|
exec_sudo_lvm = 'diskimage_builder.block_device.level1.lvm.exec_sudo'
|
|
exec_sudo_part = ('diskimage_builder.block_device.'
|
|
'level1.partitioning.exec_sudo')
|
|
exec_sudo_loop = ('diskimage_builder.block_device.'
|
|
'level0.localloop.exec_sudo')
|
|
image_create = ('diskimage_builder.block_device.level0.'
|
|
'localloop.LocalLoopNode.create')
|
|
size_of_block = ('diskimage_builder.block_device.level1.'
|
|
'partitioning.Partitioning._size_of_block_dev')
|
|
create_mbr = ('diskimage_builder.block_device.level1.'
|
|
'partitioning.Partitioning._create_mbr')
|
|
|
|
manager = mock.MagicMock()
|
|
with mock.patch(exec_sudo_lvm) as mock_sudo_lvm, \
|
|
mock.patch(exec_sudo_part) as mock_sudo_part, \
|
|
mock.patch(exec_sudo_loop) as mock_sudo_loop, \
|
|
mock.patch(image_create) as mock_image_create, \
|
|
mock.patch(size_of_block) as mock_size_of_block, \
|
|
mock.patch(create_mbr) as mock_create_mbr:
|
|
|
|
manager.attach_mock(mock_sudo_lvm, 'sudo_lvm')
|
|
manager.attach_mock(mock_sudo_part, 'sudo_part')
|
|
manager.attach_mock(mock_sudo_loop, 'sudo_loop')
|
|
manager.attach_mock(mock_image_create, 'image_create')
|
|
manager.attach_mock(mock_size_of_block, 'size_of_block')
|
|
manager.attach_mock(mock_create_mbr, 'create_mbr')
|
|
|
|
for node in call_order:
|
|
# We're just keeping this to the partition setup and
|
|
# LVM creation; i.e. skipping mounting, mkfs, etc.
|
|
if isinstance(node, (LVMNode, PvsNode,
|
|
VgsNode, LvsNode,
|
|
LocalLoopNode, PartitionNode)):
|
|
node.create()
|
|
else:
|
|
logger.debug("Skipping node for test: %s", node)
|
|
|
|
cmd_sequence = [
|
|
# create the underlying block device
|
|
mock.call.image_create(),
|
|
mock.call.size_of_block('image'),
|
|
# write out partition table
|
|
mock.call.create_mbr(),
|
|
# now mount partitions
|
|
mock.call.sudo_part(['sync']),
|
|
mock.call.sudo_part(['kpartx', '-avs', '/dev/fake/image0']),
|
|
# now create lvm environment
|
|
mock.call.sudo_lvm(['pvcreate', '/dev/fake/root', '--force']),
|
|
mock.call.sudo_lvm(
|
|
['vgcreate', 'vg', '/dev/fake/root', '--force']),
|
|
mock.call.sudo_lvm(
|
|
['lvcreate', '--name', 'lv_root', '-l', '28%VG', 'vg']),
|
|
mock.call.sudo_lvm(
|
|
['lvcreate', '--name', 'lv_tmp', '-l', '4%VG', 'vg']),
|
|
mock.call.sudo_lvm(
|
|
['lvcreate', '--name', 'lv_var', '-l', '40%VG', 'vg']),
|
|
mock.call.sudo_lvm(
|
|
['lvcreate', '--name', 'lv_log', '-l', '23%VG', 'vg']),
|
|
mock.call.sudo_lvm(
|
|
['lvcreate', '--name', 'lv_audit', '-l', '4%VG', 'vg']),
|
|
mock.call.sudo_lvm(
|
|
['lvcreate', '--name', 'lv_home', '-l', '1%VG', 'vg']),
|
|
]
|
|
manager.assert_has_calls(cmd_sequence)
|
|
|
|
#
|
|
# Umount/cleanup test
|
|
#
|
|
manager = mock.MagicMock()
|
|
with mock.patch(exec_sudo_lvm) as mock_sudo_lvm, \
|
|
mock.patch(exec_sudo_part) as mock_sudo_part, \
|
|
mock.patch(exec_sudo_loop) as mock_sudo_loop:
|
|
|
|
manager.attach_mock(mock_sudo_lvm, 'sudo_lvm')
|
|
manager.attach_mock(mock_sudo_part, 'sudo_part')
|
|
manager.attach_mock(mock_sudo_loop, 'sudo_loop')
|
|
|
|
def run_it(phase):
|
|
reverse_order = reversed(call_order)
|
|
for node in reverse_order:
|
|
if isinstance(node, (LVMNode, PvsNode,
|
|
VgsNode, LvsNode,
|
|
LocalLoopNode, PartitionNode)):
|
|
getattr(node, phase)()
|
|
else:
|
|
logger.debug("Skipping node for test: %s", node)
|
|
|
|
run_it('umount')
|
|
run_it('cleanup')
|
|
|
|
cmd_sequence = [
|
|
# deactivate LVM first
|
|
mock.call.sudo_lvm(['lvchange', '-an', '/dev/vg/lv_root']),
|
|
mock.call.sudo_lvm(['lvchange', '-an', '/dev/vg/lv_tmp']),
|
|
mock.call.sudo_lvm(['lvchange', '-an', '/dev/vg/lv_var']),
|
|
mock.call.sudo_lvm(['lvchange', '-an', '/dev/vg/lv_log']),
|
|
mock.call.sudo_lvm(['lvchange', '-an', '/dev/vg/lv_audit']),
|
|
mock.call.sudo_lvm(['lvchange', '-an', '/dev/vg/lv_home']),
|
|
mock.call.sudo_lvm(['vgchange', '-an', 'vg']),
|
|
mock.call.sudo_lvm(['udevadm', 'settle']),
|
|
# now remove partitions (note has to happen after lvm removal)
|
|
mock.call.sudo_part(['kpartx', '-d', '/dev/fake/image0']),
|
|
# now remove loopback device
|
|
mock.call.sudo_loop(['losetup', '-d', '/dev/fake/image0']),
|
|
# now final LVM cleanup call
|
|
mock.call.sudo_lvm(['pvscan', '--cache']),
|
|
]
|
|
|
|
manager.assert_has_calls(cmd_sequence)
|