kiwi-el8/test/unit/storage/disk_test.py
Marcus Schäfer 69c5495b79
Add support for part clones to the Disk interface
The Disk class provides methods to create partition(s)
and map names according to its scope and independent of
the actual partition tools. For example: create_root_partition().
This commit adds an additional optional clone parameter to
all methods for which we want to allow partition clones
2022-04-10 16:52:05 +02:00

328 lines
12 KiB
Python

import logging
from mock import (
patch, mock_open, call
)
from pytest import (
fixture, raises
)
import mock
from kiwi.storage.disk import ptable_entry_type
from kiwi.storage.disk import Disk
from kiwi.exceptions import KiwiCustomPartitionConflictError
class TestDisk:
@fixture(autouse=True)
def inject_fixtures(self, caplog):
self._caplog = caplog
@patch('kiwi.storage.disk.Partitioner.new')
def setup(self, mock_partitioner):
self.tempfile = mock.Mock()
self.tempfile.name = 'tempfile'
self.partitioner = mock.Mock()
self.partitioner.create = mock.Mock()
self.partitioner.get_id = mock.Mock(
return_value=1
)
mock_partitioner.return_value = self.partitioner
self.storage_provider = mock.Mock()
self.storage_provider.is_loop = mock.Mock(
return_value=True
)
self.storage_provider.get_device = mock.Mock(
return_value='/dev/loop0'
)
self.disk = Disk('gpt', self.storage_provider)
@patch('kiwi.storage.disk.Partitioner.new')
def setup_method(self, cls, mock_partitioner):
self.setup()
@patch('os.path.exists')
def test_get_device(self, mock_exists):
mock_exists.return_value = True
self.disk.partition_map['root'] = '/dev/root-device'
assert self.disk.get_device()['root'].get_device() == '/dev/root-device'
def test_is_loop(self):
self.disk.is_loop()
self.storage_provider.is_loop.called_once_with()
def test_create_root_partition(self):
self.disk.create_root_partition('100', 1)
assert self.partitioner.create.call_args_list == [
call('p.lxrootclone1', '100', 't.linux'),
call('p.lxroot', '100', 't.linux')
]
def test_create_root_which_is_also_boot_partition(self):
self.disk.create_root_partition('200')
self.partitioner.create.assert_called_once_with(
'p.lxroot', '200', 't.linux'
)
assert self.disk.public_partition_id_map['kiwi_RootPart'] == 1
assert self.disk.public_partition_id_map['kiwi_BootPart'] == 1
def test_create_root_which_is_also_read_write_partition(self):
self.disk.public_partition_id_map['kiwi_ROPart'] = 1
self.disk.create_root_partition('200')
self.partitioner.create.assert_called_once_with(
'p.lxroot', '200', 't.linux'
)
assert self.disk.public_partition_id_map['kiwi_RootPart'] == 1
assert self.disk.public_partition_id_map['kiwi_RWPart'] == 1
def test_create_root_lvm_partition(self):
self.disk.create_root_lvm_partition('100', 1)
assert self.partitioner.create.call_args_list == [
call('p.lxrootclone1', '100', 't.lvm'),
call('p.lxlvm', '100', 't.lvm')
]
assert self.disk.public_partition_id_map['kiwi_rootPartClone1'] == 1
assert self.disk.public_partition_id_map['kiwi_RootPart'] == 1
def test_create_root_raid_partition(self):
self.disk.create_root_raid_partition('100', 1)
assert self.partitioner.create.call_args_list == [
call('p.lxrootclone1', '100', 't.raid'),
call('p.lxraid', '100', 't.raid')
]
assert self.disk.public_partition_id_map['kiwi_rootPartClone1'] == 1
assert self.disk.public_partition_id_map['kiwi_RootPart'] == 1
assert self.disk.public_partition_id_map['kiwi_RaidPart'] == 1
def test_create_root_readonly_partition(self):
self.disk.create_root_readonly_partition('100', 1)
assert self.partitioner.create.call_args_list == [
call('p.lxrootclone1', '100', 't.linux'),
call('p.lxreadonly', '100', 't.linux')
]
assert self.disk.public_partition_id_map['kiwi_rootPartClone1'] == 1
assert self.disk.public_partition_id_map['kiwi_ROPart'] == 1
def test_create_boot_partition(self):
self.disk.create_boot_partition('100', 1)
assert self.partitioner.create.call_args_list == [
call('p.lxbootclone1', '100', 't.linux'),
call('p.lxboot', '100', 't.linux')
]
assert self.disk.public_partition_id_map['kiwi_bootPartClone1'] == 1
assert self.disk.public_partition_id_map['kiwi_BootPart'] == 1
def test_create_efi_csm_partition(self):
self.disk.create_efi_csm_partition('100')
self.partitioner.create.assert_called_once_with(
'p.legacy', '100', 't.csm'
)
assert self.disk.public_partition_id_map['kiwi_BiosGrub'] == 1
def test_create_efi_partition(self):
self.disk.create_efi_partition('100')
self.partitioner.create.assert_called_once_with(
'p.UEFI', '100', 't.efi'
)
assert self.disk.public_partition_id_map['kiwi_EfiPart'] == 1
def test_create_spare_partition(self):
self.disk.create_spare_partition('42')
self.partitioner.create.assert_called_once_with(
'p.spare', '42', 't.linux'
)
assert self.disk.public_partition_id_map['kiwi_SparePart'] == 1
def test_create_swap_partition(self):
self.disk.create_swap_partition('42')
self.partitioner.create.assert_called_once_with(
'p.swap', '42', 't.swap'
)
assert self.disk.public_partition_id_map['kiwi_SwapPart'] == 1
@patch('kiwi.storage.disk.Command.run')
def test_create_prep_partition(self, mock_command):
self.disk.create_prep_partition('8')
self.partitioner.create.assert_called_once_with(
'p.prep', '8', 't.prep'
)
assert self.disk.public_partition_id_map['kiwi_PrepPart'] == 1
@patch('kiwi.storage.disk.Command.run')
def test_create_custom_partitions(self, mock_command):
table_entries = {
'var': ptable_entry_type(
mbsize='100',
clone=2,
partition_name='p.lxvar',
partition_type='t.linux',
mountpoint='/var',
filesystem='ext3'
)
}
self.disk.create_custom_partitions(table_entries)
assert self.partitioner.create.call_args_list == [
call('p.lxvarclone1', '100', 't.linux'),
call('p.lxvarclone2', '100', 't.linux'),
call('p.lxvar', '100', 't.linux')
]
assert self.disk.public_partition_id_map['kiwi_varPartClone1'] == 1
assert self.disk.public_partition_id_map['kiwi_varPartClone2'] == 1
assert self.disk.public_partition_id_map['kiwi_VarPart'] == 1
def test_create_custom_partitions_reserved_name(self):
table_entries = {
'root': ptable_entry_type(
mbsize='100',
clone=0,
partition_name='p.lxroot',
partition_type='t.linux',
mountpoint='/',
filesystem='ext3'
)
}
with raises(KiwiCustomPartitionConflictError):
self.disk.create_custom_partitions(table_entries)
@patch('kiwi.storage.disk.Command.run')
def test_device_map_efi_partition(self, mock_command):
self.disk.create_efi_partition('100')
self.disk.map_partitions()
assert self.disk.partition_map == {'efi': '/dev/mapper/loop0p1'}
self.disk.is_mapped = False
@patch('kiwi.storage.disk.Command.run')
def test_device_map_prep_partition(self, mock_command):
self.disk.create_prep_partition('8')
self.disk.map_partitions()
assert self.disk.partition_map == {'prep': '/dev/mapper/loop0p1'}
self.disk.is_mapped = False
@patch('kiwi.storage.disk.Command.run')
def test_device_map_linux_dev_sda(self, mock_command):
self.storage_provider.is_loop.return_value = False
self.storage_provider.get_device = mock.Mock(
return_value='/dev/sda'
)
self.disk.create_efi_partition('100')
self.disk.map_partitions()
assert self.disk.partition_map == {'efi': '/dev/sda1'}
self.disk.is_mapped = False
@patch('kiwi.storage.disk.Command.run')
def test_device_map_linux_dev_c0d0(self, mock_command):
self.storage_provider.is_loop.return_value = False
self.storage_provider.get_device = mock.Mock(
return_value='/dev/c0d0'
)
self.disk.create_efi_partition('100')
self.disk.map_partitions()
assert self.disk.partition_map == {'efi': '/dev/c0d0p1'}
self.disk.is_mapped = False
@patch('kiwi.storage.disk.Command.run')
def test_activate_boot_partition_is_boot_partition(self, mock_command):
self.disk.create_boot_partition('100')
self.disk.create_root_partition('100')
self.disk.activate_boot_partition()
self.partitioner.set_flag(1, 'f.active')
@patch('kiwi.storage.disk.Command.run')
def test_activate_boot_partition_is_root_partition(self, mock_command):
self.disk.create_root_partition('100')
self.disk.activate_boot_partition()
self.partitioner.set_flag(1, 'f.active')
@patch('kiwi.storage.disk.Command.run')
def test_activate_boot_partition_is_prep_partition(self, mock_command):
self.disk.create_prep_partition('8')
self.disk.activate_boot_partition()
self.partitioner.set_flag(1, 'f.active')
@patch('kiwi.storage.disk.Command.run')
def test_wipe_gpt(self, mock_command):
self.disk.wipe()
mock_command.assert_called_once_with(
['sgdisk', '--zap-all', '/dev/loop0']
)
@patch('kiwi.storage.disk.Command.run')
@patch('kiwi.storage.disk.Temporary.new_file')
def test_wipe_dasd(self, mock_temp, mock_command):
mock_command.side_effect = Exception
self.disk.table_type = 'dasd'
mock_temp.return_value = self.tempfile
m_open = mock_open()
with patch('builtins.open', m_open, create=True):
self.disk.wipe()
m_open.return_value.write.assert_called_once_with(
'y\n\nw\nq\n'
)
with self._caplog.at_level(logging.DEBUG):
mock_command.assert_called_once_with(
['bash', '-c', 'cat tempfile | fdasd -f /dev/loop0']
)
@patch('kiwi.storage.disk.Command.run')
def test_map_partitions_loop(self, mock_command):
self.disk.map_partitions()
mock_command.assert_called_once_with(
['kpartx', '-s', '-a', '/dev/loop0']
)
self.disk.is_mapped = False
@patch('kiwi.storage.disk.Command.run')
def test_map_partitions_other(self, mock_command):
self.storage_provider.is_loop.return_value = False
self.disk.map_partitions()
mock_command.assert_called_once_with(
['partprobe', '/dev/loop0']
)
@patch('kiwi.storage.disk.Command.run')
def test_destructor_dm_cleanup_failed(self, mock_command):
self.disk.is_mapped = True
self.disk.partition_map = {'root': '/dev/mapper/loop0p1'}
mock_command.side_effect = Exception
self.disk.__del__()
with self._caplog.at_level(logging.WARNING):
mock_command.assert_called_once_with(
['dmsetup', 'remove', '/dev/mapper/loop0p1']
)
self.disk.is_mapped = False
@patch('kiwi.storage.disk.Command.run')
def test_destructor(self, mock_command):
self.disk.is_mapped = True
self.disk.partition_map = {'root': '/dev/mapper/loop0p1'}
self.disk.__del__()
assert mock_command.call_args_list == [
call(['dmsetup', 'remove', '/dev/mapper/loop0p1']),
call(['kpartx', '-d', '/dev/loop0'])
]
self.disk.is_mapped = False
def test_get_public_partition_id_map(self):
assert self.disk.get_public_partition_id_map() == {}
def test_create_hybrid_mbr(self):
self.disk.create_hybrid_mbr()
self.partitioner.set_hybrid_mbr.assert_called_once_with()
def test_create_mbr(self):
self.disk.create_mbr()
self.partitioner.set_mbr.assert_called_once_with()
def test_parse_size(self):
(size, _) = self.disk._parse_size('100')
assert size == '100'
(size, clone_size) = self.disk._parse_size('all_free')
assert size == 'all_free'
assert clone_size == 'all_free'
(size, clone_size) = self.disk._parse_size('clone:100:all_free')
assert size == '100'
assert clone_size == 'all_free'