1
from mock import patch, call
2
from argparse import Namespace
4
from curtin.commands import block_meta
5
from .helpers import CiTestCase
8
class TestBlockMetaSimple(CiTestCase):
10
super(TestBlockMetaSimple, self).setUp()
11
self.target = "my_target"
14
basepath = 'curtin.commands.block_meta.'
15
self.add_patch(basepath + 'get_bootpt_cfg', 'mock_bootpt_cfg')
16
self.add_patch(basepath + 'get_partition_format_type',
19
self.add_patch('curtin.block.stop_all_unused_multipath_devices',
21
self.add_patch('curtin.block.get_installable_blockdevs',
22
'mock_block_get_installable_bdevs')
23
self.add_patch('curtin.block.get_dev_name_entry',
24
'mock_block_get_dev_name_entry')
25
self.add_patch('curtin.block.get_root_device',
26
'mock_block_get_root_device')
27
self.add_patch('curtin.block.is_valid_device',
28
'mock_block_is_valid_device')
30
self.add_patch('curtin.config.load_command_config',
33
self.add_patch('curtin.util.subp', 'mock_subp')
34
self.add_patch('curtin.util.load_command_environment',
37
def test_write_image_to_disk(self):
40
'uri': 'http://myhost/curtin-unittest-dd.xz'
42
devname = "fakedisk1p1"
43
devnode = "/dev/" + devname
44
self.mock_block_get_dev_name_entry.return_value = (devname, devnode)
46
block_meta.write_image_to_disk(source, devname)
49
'wget "$1" --progress=dot:mega -O - |xzcat| dd bs=4M of="$2"',
50
'--', source['uri'], devnode]
51
self.mock_block_get_dev_name_entry.assert_called_with(devname)
52
self.mock_subp.assert_has_calls([call(args=wget),
53
call(['partprobe', devnode]),
54
call(['udevadm', 'settle'])])
55
paths = ["curtin", "system-data/var/lib/snapd"]
56
self.mock_block_get_root_device.assert_called_with([devname],
59
def test_write_image_to_disk_ddtgz(self):
62
'uri': 'http://myhost/curtin-unittest-dd.tgz'
64
devname = "fakedisk1p1"
65
devnode = "/dev/" + devname
66
self.mock_block_get_dev_name_entry.return_value = (devname, devnode)
68
block_meta.write_image_to_disk(source, devname)
71
'wget "$1" --progress=dot:mega -O - |'
72
'tar -xOzf -| dd bs=4M of="$2"',
73
'--', source['uri'], devnode]
74
self.mock_block_get_dev_name_entry.assert_called_with(devname)
75
self.mock_subp.assert_has_calls([call(args=wget),
76
call(['partprobe', devnode]),
77
call(['udevadm', 'settle'])])
78
paths = ["curtin", "system-data/var/lib/snapd"]
79
self.mock_block_get_root_device.assert_called_with([devname],
82
@patch('curtin.commands.block_meta.write_image_to_disk')
83
def test_meta_simple_calls_write_img(self, mock_write_image):
84
devname = "fakedisk1p1"
85
devnode = "/dev/" + devname
87
'unittest': {'type': 'dd-xz',
88
'uri': 'http://myhost/curtin-unittest-dd.xz'}
91
'block-meta': {'devices': [devname]},
94
self.mock_config_load.return_value = config
95
self.mock_load_env.return_value = {'target': self.target}
96
self.mock_block_is_valid_device.return_value = True
97
self.mock_block_get_dev_name_entry.return_value = (devname, devnode)
98
mock_write_image.return_value = devname
100
args = Namespace(target=self.target, devices=None, mode=None,
101
boot_fstype=None, fstype=None)
103
block_meta.meta_simple(args)
105
mock_write_image.assert_called_with(sources.get('unittest'), devname)
106
self.mock_subp.assert_has_calls(
107
[call(['mount', devname, self.target])])
110
class TestBlockMeta(CiTestCase):
113
super(TestBlockMeta, self).setUp()
115
basepath = 'curtin.commands.block_meta.'
116
self.add_patch(basepath + 'get_path_to_storage_volume', 'mock_getpath')
117
self.add_patch(basepath + 'make_dname', 'mock_make_dname')
118
self.add_patch('curtin.util.load_command_environment',
120
self.add_patch('curtin.util.subp', 'mock_subp')
121
self.add_patch('curtin.util.ensure_dir', 'mock_ensure_dir')
122
self.add_patch('curtin.block.get_part_table_type',
123
'mock_block_get_part_table_type')
124
self.add_patch('curtin.block.wipe_volume',
125
'mock_block_wipe_volume')
126
self.add_patch('curtin.block.path_to_kname',
127
'mock_block_path_to_kname')
128
self.add_patch('curtin.block.sys_block_path',
129
'mock_block_sys_block_path')
130
self.add_patch('curtin.block.clear_holders.get_holders',
132
self.add_patch('curtin.block.clear_holders.clear_holders',
133
'mock_clear_holders')
134
self.add_patch('curtin.block.clear_holders.assert_clear',
136
self.add_patch('curtin.block.iscsi.volpath_is_iscsi',
137
'mock_volpath_is_iscsi')
138
self.add_patch('curtin.block.get_volume_uuid',
139
'mock_block_get_volume_uuid')
140
self.add_patch('curtin.block.zero_file_at_offsets',
141
'mock_block_zero_file')
143
self.target = "my_target"
148
{'grub_device': True,
154
'wipe': 'superblock'},
160
'offset': '4194304B',
161
'size': '511705088B',
163
'uuid': 'fc7ab24c-b6bf-460f-8446-d3ac362c0625',
164
'wipe': 'superblock'},
168
'volume': 'sda-part1'},
169
{'id': 'sda-part1-mnt-root',
172
'device': 'sda1-root'},
173
{'id': 'sda-part1-mnt-root-ro',
177
'device': 'sda1-root'}
181
self.storage_config = (
182
block_meta.extract_storage_ordered_dict(self.config))
184
def test_disk_handler_calls_clear_holder(self):
185
info = self.storage_config.get('sda')
186
disk = info.get('path')
187
self.mock_getpath.return_value = disk
188
self.mock_block_get_part_table_type.return_value = 'dos'
189
self.mock_subp.side_effect = iter([
190
(0, 0), # parted mklabel
193
self.mock_get_holders.return_value = holders
195
block_meta.disk_handler(info, self.storage_config)
197
print("clear_holders: %s" % self.mock_clear_holders.call_args_list)
198
print("assert_clear: %s" % self.mock_assert_clear.call_args_list)
199
self.mock_clear_holders.assert_called_with(disk)
200
self.mock_assert_clear.assert_called_with(disk)
202
def test_partition_handler_wipes_at_partition_offset(self):
203
""" Test wiping partition at offset prior to creating partition"""
204
disk_info = self.storage_config.get('sda')
205
part_info = self.storage_config.get('sda-part1')
206
disk_kname = disk_info.get('path')
207
part_kname = disk_kname + '1'
208
self.mock_getpath.side_effect = iter([
212
self.mock_block_get_part_table_type.return_value = 'dos'
214
self.mock_block_path_to_kname.return_value = kname
215
self.mock_block_sys_block_path.return_value = '/sys/class/block/xxx'
217
block_meta.partition_handler(part_info, self.storage_config)
218
part_offset = 2048 * 512
219
self.mock_block_zero_file.assert_called_with(disk_kname, [part_offset],
221
self.mock_subp.assert_called_with(['parted', disk_kname, '--script',
222
'mkpart', 'primary', '2048s',
223
'1001471s'], capture=True)
225
@patch('curtin.util.write_file')
226
def test_mount_handler_defaults(self, mock_write_file):
227
"""Test mount_handler has defaults to 'defaults' for mount options"""
228
fstab = self.tmp_path('fstab')
229
self.mock_load_env.return_value = {'fstab': fstab,
230
'target': self.target}
231
disk_info = self.storage_config.get('sda')
232
fs_info = self.storage_config.get('sda1-root')
233
mount_info = self.storage_config.get('sda-part1-mnt-root')
235
self.mock_getpath.return_value = '/wark/xxx'
236
self.mock_volpath_is_iscsi.return_value = False
237
self.mock_block_get_volume_uuid.return_value = None
239
block_meta.mount_handler(mount_info, self.storage_config)
241
expected = "%s %s %s %s 0 0\n" % (disk_info['path'],
243
fs_info['fstype'], options)
245
mock_write_file.assert_called_with(fstab, expected, omode='a')
247
@patch('curtin.util.write_file')
248
def test_mount_handler_uses_mount_options(self, mock_write_file):
249
"""Test mount_handler 'options' string is present in fstab entry"""
250
fstab = self.tmp_path('fstab')
251
self.mock_load_env.return_value = {'fstab': fstab,
252
'target': self.target}
253
disk_info = self.storage_config.get('sda')
254
fs_info = self.storage_config.get('sda1-root')
255
mount_info = self.storage_config.get('sda-part1-mnt-root-ro')
257
self.mock_getpath.return_value = '/wark/xxx'
258
self.mock_volpath_is_iscsi.return_value = False
259
self.mock_block_get_volume_uuid.return_value = None
261
block_meta.mount_handler(mount_info, self.storage_config)
263
expected = "%s %s %s %s 0 0\n" % (disk_info['path'],
265
fs_info['fstype'], options)
267
mock_write_file.assert_called_with(fstab, expected, omode='a')
269
@patch('curtin.util.write_file')
270
def test_mount_handler_empty_options_string(self, mock_write_file):
271
"""Test mount_handler with empty 'options' string, selects defaults"""
272
fstab = self.tmp_path('fstab')
273
self.mock_load_env.return_value = {'fstab': fstab,
274
'target': self.target}
275
disk_info = self.storage_config.get('sda')
276
fs_info = self.storage_config.get('sda1-root')
277
mount_info = self.storage_config.get('sda-part1-mnt-root-ro')
278
mount_info['options'] = ''
280
self.mock_getpath.return_value = '/wark/xxx'
281
self.mock_volpath_is_iscsi.return_value = False
282
self.mock_block_get_volume_uuid.return_value = None
284
block_meta.mount_handler(mount_info, self.storage_config)
286
expected = "%s %s %s %s 0 0\n" % (disk_info['path'],
288
fs_info['fstype'], options)
290
mock_write_file.assert_called_with(fstab, expected, omode='a')
292
def test_mount_handler_appends_to_fstab(self):
293
"""Test mount_handler appends fstab lines to existing file"""
294
fstab = self.tmp_path('fstab')
295
with open(fstab, 'w') as fh:
296
fh.write("#curtin-test\n")
298
self.mock_load_env.return_value = {'fstab': fstab,
299
'target': self.target}
300
disk_info = self.storage_config.get('sda')
301
fs_info = self.storage_config.get('sda1-root')
302
mount_info = self.storage_config.get('sda-part1-mnt-root-ro')
303
mount_info['options'] = ''
305
self.mock_getpath.return_value = '/wark/xxx'
306
self.mock_volpath_is_iscsi.return_value = False
307
self.mock_block_get_volume_uuid.return_value = None
309
block_meta.mount_handler(mount_info, self.storage_config)
311
expected = "#curtin-test\n%s %s %s %s 0 0\n" % (disk_info['path'],
316
with open(fstab, 'r') as fh:
317
rendered_fstab = fh.read()
319
print(rendered_fstab)
320
self.assertEqual(rendered_fstab, expected)
added
removed
tests/unittests/test_commands_block_meta.py
