almalinux/kiwi-el8
14
0
Fork 0
Code Issues Pull Requests Releases Activity

Move LoopDevice class to context manager

Browse Source 
Change the LoopDevice class to be a context manager.
All code using LoopDevice was updated to the following
with statement:

with LoopDevice(...) as loop_provider:
    loop_provider.some_member()

This is related to Issue #2412
This commit is contained in:
Marcus Schäfer 2023-12-30 14:49:15 +01:00
parent d7bcc323d0
commit cdbe5060ca
No known key found for this signature in database
GPG Key ID: A16C1128698C8CAC
12 changed files with 490 additions and 461 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
doc/source/contributing/kiwi_from_python.rst
View File

@ -56,15 +56,15 @@ which contains your host `tmp` directory.
from kiwi.storage.loop_device import LoopDevice
from kiwi.filesystem import FileSystem
loop_provider = LoopDevice(
with LoopDevice(
filename='my_tmp.ext4', filesize_mbytes=100
)
loop_provider.create()
) as loop_provider:
loop_provider.create()
filesystem = FileSystem.new(
'ext4', loop_provider, '/tmp/'
)
filesystem.create_on_device(
label='TMP'
)
filesystem.sync_data()
filesystem = FileSystem.new(
'ext4', loop_provider, '/tmp/'
)
filesystem.create_on_device(
label='TMP'
)
filesystem.sync_data()

86
kiwi/bootloader/config/systemd_boot.py
View File

@ -104,49 +104,49 @@ class BootLoaderSystemdBoot(BootLoaderSpecBase):
Command.run(
['sgdisk', '-n', ':1.0', '-t', '1:EF00', path]
)
loop_provider = LoopDevice(path)
loop_provider.create(overwrite=False)
disk = Disk('gpt', loop_provider)
disk.map_partitions()
disk.partitioner.partition_id = 1
disk._add_to_map('efi')
Command.run(
['mkdosfs', '-n', 'BOOT', disk.partition_map['efi']]
)
Path.create(f'{self.root_dir}/boot/efi')
efi_mount = MountManager(
device=disk.partition_map['efi'],
mountpoint=f'{self.root_dir}/boot/efi'
)
device_mount = MountManager(
device='/dev',
mountpoint=f'{self.root_dir}/dev'
)
proc_mount = MountManager(
device='/proc',
mountpoint=f'{self.root_dir}/proc'
)
sys_mount = MountManager(
device='/sys',
mountpoint=f'{self.root_dir}/sys'
)
efi_mount.mount()
device_mount.bind_mount()
proc_mount.bind_mount()
sys_mount.bind_mount()
try:
self._run_bootctl(self.root_dir)
self.set_loader_entry(self.root_dir, self.target.live)
finally:
efi_mount.umount()
device_mount.umount()
proc_mount.umount()
sys_mount.umount()
Command.run(
['dd', f'if={disk.partition_map["efi"]}', f'of={path}.img']
)
del disk
del loop_provider
with LoopDevice(path) as loop_provider:
loop_provider.create(overwrite=False)
disk = Disk('gpt', loop_provider)
disk.map_partitions()
disk.partitioner.partition_id = 1
disk._add_to_map('efi')
Command.run(
['mkdosfs', '-n', 'BOOT', disk.partition_map['efi']]
)
Path.create(f'{self.root_dir}/boot/efi')
efi_mount = MountManager(
device=disk.partition_map['efi'],
mountpoint=f'{self.root_dir}/boot/efi'
)
device_mount = MountManager(
device='/dev',
mountpoint=f'{self.root_dir}/dev'
)
proc_mount = MountManager(
device='/proc',
mountpoint=f'{self.root_dir}/proc'
)
sys_mount = MountManager(
device='/sys',
mountpoint=f'{self.root_dir}/sys'
)
efi_mount.mount()
device_mount.bind_mount()
proc_mount.bind_mount()
sys_mount.bind_mount()
try:
self._run_bootctl(self.root_dir)
self.set_loader_entry(self.root_dir, self.target.live)
finally:
efi_mount.umount()
device_mount.umount()
proc_mount.umount()
sys_mount.umount()
Command.run(
['dd', f'if={disk.partition_map["efi"]}', f'of={path}.img']
)
del disk
Command.run(
['mv', f'{path}.img', path]
)

607
kiwi/builder/disk.py
View File

@ -279,307 +279,310 @@ class DiskBuilder:
# create the disk
log.info('Creating raw disk image %s', self.diskname)
loop_provider = LoopDevice(
with LoopDevice(
self.diskname, disksize_mbytes, self.blocksize
)
loop_provider.create()
) as loop_provider:
loop_provider.create()
disk = Disk(
self.firmware.get_partition_table_type(), loop_provider,
self.disk_start_sector,
extended_layout=bool(self.dosparttable_extended_layout)
)
# create the bootloader instance
if self.bootloader != 'custom':
self.bootloader_config = BootLoaderConfig.new(
self.bootloader, self.xml_state, root_dir=self.root_dir,
boot_dir=self.root_dir, custom_args={
'targetbase':
loop_provider.get_device(),
'grub_directory_name':
Defaults.get_grub_boot_directory_name(self.root_dir),
'crypto_disk':
True if self.luks is not None else False,
'boot_is_crypto':
self.boot_is_crypto,
'config_options':
self.xml_state.get_bootloader_config_options()
}
disk = Disk(
self.firmware.get_partition_table_type(), loop_provider,
self.disk_start_sector,
extended_layout=bool(self.dosparttable_extended_layout)
)
# create disk partitions and instance device map
device_map = self._build_and_map_disk_partitions(disk, disksize_mbytes)
if self.root_filesystem_is_overlay and \
self.root_filesystem_has_write_partition is False:
device_map['root'] = device_map['readonly']
disk.public_partition_id_map['kiwi_RootPart'] = \
disk.public_partition_id_map['kiwi_ROPart']
# create raid on current root device if requested
raid_root = None
if self.mdraid:
raid_root = RaidDevice(device_map['root'])
raid_root.create_degraded_raid(raid_level=self.mdraid)
device_map['root'] = raid_root.get_device()
disk.public_partition_id_map['kiwi_RaidPart'] = \
disk.public_partition_id_map['kiwi_RootPart']
disk.public_partition_id_map['kiwi_RaidDev'] = \
device_map['root'].get_device()
# create integrity on current root device if requested
if self.integrity_root:
options = []
if self.integrity_legacy_hmac:
options.append('legacy_hmac')
self.integrity_root = IntegrityDevice(
device_map['root'], defaults.INTEGRITY_ALGORITHM,
integrity_credentials_type(
keydescription=self.integrity_key_description,
keyfile=self.integrity_keyfile,
keyfile_algorithm=defaults.INTEGRITY_KEY_ALGORITHM,
options=options
# create the bootloader instance
if self.bootloader != 'custom':
self.bootloader_config = BootLoaderConfig.new(
self.bootloader, self.xml_state, root_dir=self.root_dir,
boot_dir=self.root_dir, custom_args={
'targetbase':
loop_provider.get_device(),
'grub_directory_name':
Defaults.get_grub_boot_directory_name(self.root_dir),
'crypto_disk':
True if self.luks is not None else False,
'boot_is_crypto':
self.boot_is_crypto,
'config_options':
self.xml_state.get_bootloader_config_options()
}
)
# create disk partitions and instance device map
device_map = self._build_and_map_disk_partitions(
disk, disksize_mbytes
)
self.integrity_root.create_dm_integrity()
device_map['integrity_root'] = device_map['root']
device_map['root'] = self.integrity_root.get_device()
if self.root_filesystem_is_overlay and \
self.root_filesystem_has_write_partition is False:
device_map['readonly'] = device_map['root']
device_map['root'] = device_map['readonly']
disk.public_partition_id_map['kiwi_RootPart'] = \
disk.public_partition_id_map['kiwi_ROPart']
# create luks on current root device if requested
luks_root = None
if self.luks is not None:
luks_root = LuksDevice(device_map['root'])
self.luks_boot_keyname = '/root/.root.keyfile'
self.luks_boot_keyfile = ''.join(
[self.root_dir, self.luks_boot_keyname]
)
# use LUKS key file for the following conditions:
# 1. /boot is encrypted
# In this case grub needs to read from LUKS via the
# cryptodisk module which at the moment always asks
# for the passphrase even when empty. The keyfile
# setup makes sure only one interaction on the grub
# stage is needed
# 2. LUKS passphrase is configured as empty string
# In this case the keyfile allows to open the
# LUKS pool without asking
#
luks_need_keyfile = \
True if self.boot_is_crypto or self.luks == '' else False
luks_root.create_crypto_luks(
passphrase=self.luks,
osname=self.luks_os,
options=self.luks_format_options,
keyfile=self.luks_boot_keyname if luks_need_keyfile else '',
randomize=self.luks_randomize,
root_dir=self.root_dir
)
if luks_need_keyfile:
self.luks_boot_keyfile_setup = ''.join(
[self.root_dir, '/etc/dracut.conf.d/99-luks-boot.conf']
# create raid on current root device if requested
raid_root = None
if self.mdraid:
raid_root = RaidDevice(device_map['root'])
raid_root.create_degraded_raid(raid_level=self.mdraid)
device_map['root'] = raid_root.get_device()
disk.public_partition_id_map['kiwi_RaidPart'] = \
disk.public_partition_id_map['kiwi_RootPart']
disk.public_partition_id_map['kiwi_RaidDev'] = \
device_map['root'].get_device()
# create integrity on current root device if requested
if self.integrity_root:
options = []
if self.integrity_legacy_hmac:
options.append('legacy_hmac')
self.integrity_root = IntegrityDevice(
device_map['root'], defaults.INTEGRITY_ALGORITHM,
integrity_credentials_type(
keydescription=self.integrity_key_description,
keyfile=self.integrity_keyfile,
keyfile_algorithm=defaults.INTEGRITY_KEY_ALGORITHM,
options=options
)
)
self.boot_image.write_system_config_file(
config={'install_items': [self.luks_boot_keyname]},
config_file=self.luks_boot_keyfile_setup
self.integrity_root.create_dm_integrity()
device_map['integrity_root'] = device_map['root']
device_map['root'] = self.integrity_root.get_device()
if self.root_filesystem_is_overlay and \
self.root_filesystem_has_write_partition is False:
device_map['readonly'] = device_map['root']
# create luks on current root device if requested
luks_root = None
if self.luks is not None:
luks_root = LuksDevice(device_map['root'])
self.luks_boot_keyname = '/root/.root.keyfile'
self.luks_boot_keyfile = ''.join(
[self.root_dir, self.luks_boot_keyname]
)
self.boot_image.include_file(
'/root/' + os.path.basename(self.luks_boot_keyfile)
# use LUKS key file for the following conditions:
# 1. /boot is encrypted
# In this case grub needs to read from LUKS via the
# cryptodisk module which at the moment always asks
# for the passphrase even when empty. The keyfile
# setup makes sure only one interaction on the grub
# stage is needed
# 2. LUKS passphrase is configured as empty string
# In this case the keyfile allows to open the
# LUKS pool without asking
#
luks_need_keyfile = \
True if self.boot_is_crypto or self.luks == '' else False
luks_root.create_crypto_luks(
passphrase=self.luks,
osname=self.luks_os,
options=self.luks_format_options,
keyfile=self.luks_boot_keyname if luks_need_keyfile else '',
randomize=self.luks_randomize,
root_dir=self.root_dir
)
device_map['luks_root'] = device_map['root']
device_map['root'] = luks_root.get_device()
if self.root_filesystem_is_overlay and \
self.root_filesystem_has_write_partition is False:
device_map['readonly'] = device_map['root']
if luks_need_keyfile:
self.luks_boot_keyfile_setup = ''.join(
[self.root_dir, '/etc/dracut.conf.d/99-luks-boot.conf']
)
self.boot_image.write_system_config_file(
config={'install_items': [self.luks_boot_keyname]},
config_file=self.luks_boot_keyfile_setup
)
self.boot_image.include_file(
'/root/' + os.path.basename(self.luks_boot_keyfile)
)
device_map['luks_root'] = device_map['root']
device_map['root'] = luks_root.get_device()
if self.root_filesystem_is_overlay and \
self.root_filesystem_has_write_partition is False:
device_map['readonly'] = device_map['root']
# create spare filesystem on spare partition if present
system_spare = self._build_spare_filesystem(device_map)
# create spare filesystem on spare partition if present
system_spare = self._build_spare_filesystem(device_map)
system_custom_parts = self._build_custom_parts_filesystem(
device_map, self.custom_partitions
)
# create filesystems on boot partition(s) if any
system_boot, system_efi = self._build_boot_filesystems(device_map)
# create volumes and filesystems for root system
if self.volume_manager_name:
volume_manager_custom_parameters = {
'fs_mount_options':
self.custom_root_mount_args,
'fs_create_options':
self.custom_root_creation_args,
'root_label':
self.disk_setup.get_root_label(),
'root_is_snapshot':
self.xml_state.build_type.get_btrfs_root_is_snapshot(),
'root_is_readonly_snapshot':
self.xml_state.build_type.
get_btrfs_root_is_readonly_snapshot(),
'root_is_subvolume':
self.xml_state.build_type.
get_btrfs_root_is_subvolume(),
'btrfs_default_volume_requested':
self.btrfs_default_volume_requested,
'quota_groups':
self.xml_state.build_type.get_btrfs_quota_groups(),
'resize_on_boot':
self.disk_resize_requested
}
volume_manager = VolumeManager.new(
self.volume_manager_name, device_map,
self.root_dir + '/',
self.volumes,
volume_manager_custom_parameters
system_custom_parts = self._build_custom_parts_filesystem(
device_map, self.custom_partitions
)
volume_manager.setup(
self.volume_group_name
)
volume_manager.create_volumes(
self.requested_filesystem
)
volume_manager.mount_volumes()
system = volume_manager
device_map['root'] = volume_manager.get_device().get('root')
device_map['swap'] = volume_manager.get_device().get('swap')
else:
if not self.root_filesystem_is_overlay or \
self.root_filesystem_has_write_partition is not False:
log.info(
'Creating root(%s) filesystem on %s',
self.requested_filesystem, device_map['root'].get_device()
)
filesystem_custom_parameters = {
'mount_options': self.custom_root_mount_args,
'create_options': self.custom_root_creation_args
# create filesystems on boot partition(s) if any
system_boot, system_efi = self._build_boot_filesystems(device_map)
# create volumes and filesystems for root system
if self.volume_manager_name:
volume_manager_custom_parameters = {
'fs_mount_options':
self.custom_root_mount_args,
'fs_create_options':
self.custom_root_creation_args,
'root_label':
self.disk_setup.get_root_label(),
'root_is_snapshot':
self.xml_state.build_type.get_btrfs_root_is_snapshot(),
'root_is_readonly_snapshot':
self.xml_state.build_type.
get_btrfs_root_is_readonly_snapshot(),
'root_is_subvolume':
self.xml_state.build_type.
get_btrfs_root_is_subvolume(),
'btrfs_default_volume_requested':
self.btrfs_default_volume_requested,
'quota_groups':
self.xml_state.build_type.get_btrfs_quota_groups(),
'resize_on_boot':
self.disk_resize_requested
}
filesystem = FileSystem.new(
self.requested_filesystem, device_map['root'],
volume_manager = VolumeManager.new(
self.volume_manager_name, device_map,
self.root_dir + '/',
filesystem_custom_parameters
self.volumes,
volume_manager_custom_parameters
)
if self.root_filesystem_embed_integrity_metadata:
filesystem.create_on_device(
label=self.disk_setup.get_root_label(),
size=-defaults.DM_METADATA_OFFSET,
unit=defaults.UNIT.byte
)
else:
filesystem.create_on_device(
label=self.disk_setup.get_root_label(),
)
system = filesystem
# create swap on current root device if requested
if self.swap_mbytes:
swap = FileSystem.new(
'swap', device_map['swap']
)
swap.create_on_device(
label='SWAP'
)
# store root partition/filesystem uuid for profile
self._preserve_root_partition_uuid(device_map)
self._preserve_root_filesystem_uuid(device_map)
# create a random image identifier
self.mbrid = SystemIdentifier()
self.mbrid.calculate_id()
# create first stage metadata to boot image
self._write_partition_id_config_to_boot_image(disk)
self._write_recovery_metadata_to_boot_image()
self._write_raid_config_to_boot_image(raid_root)
self._write_generic_fstab_to_boot_image(device_map, system)
self.system_setup.export_modprobe_setup(
self.boot_image.boot_root_directory
)
# create first stage metadata to system image
self._write_image_identifier_to_system_image()
self._write_crypttab_to_system_image(luks_root)
self._write_integritytab_to_system_image(self.integrity_root)
self._write_generic_fstab_to_system_image(device_map, system)
if self.initrd_system == 'dracut':
if self.root_filesystem_is_multipath is False:
self.boot_image.omit_module('multipath')
if self.root_filesystem_is_overlay:
self.boot_image.include_module('kiwi-overlay')
self.boot_image.write_system_config_file(
config={'modules': ['kiwi-overlay']}
volume_manager.setup(
self.volume_group_name
)
if self.disk_resize_requested:
self.boot_image.include_module('kiwi-repart')
volume_manager.create_volumes(
self.requested_filesystem
)
volume_manager.mount_volumes()
system = volume_manager
device_map['root'] = volume_manager.get_device().get('root')
device_map['swap'] = volume_manager.get_device().get('swap')
else:
if not self.root_filesystem_is_overlay or \
self.root_filesystem_has_write_partition is not False:
log.info(
'Creating root(%s) filesystem on %s',
self.requested_filesystem, device_map['root'].get_device()
)
filesystem_custom_parameters = {
'mount_options': self.custom_root_mount_args,
'create_options': self.custom_root_creation_args
}
filesystem = FileSystem.new(
self.requested_filesystem, device_map['root'],
self.root_dir + '/',
filesystem_custom_parameters
)
if self.root_filesystem_embed_integrity_metadata:
filesystem.create_on_device(
label=self.disk_setup.get_root_label(),
size=-defaults.DM_METADATA_OFFSET,
unit=defaults.UNIT.byte
)
else:
filesystem.create_on_device(
label=self.disk_setup.get_root_label(),
)
system = filesystem
# create initrd
if self.boot_image.has_initrd_support():
self.boot_image.create_initrd(self.mbrid)
# create swap on current root device if requested
if self.swap_mbytes:
swap = FileSystem.new(
'swap', device_map['swap']
)
swap.create_on_device(
label='SWAP'
)
# create second stage metadata to system image
self._copy_first_boot_files_to_system_image()
# store root partition/filesystem uuid for profile
self._preserve_root_partition_uuid(device_map)
self._preserve_root_filesystem_uuid(device_map)
self._write_bootloader_meta_data_to_system_image(
device_map, disk, system
)
# create a random image identifier
self.mbrid = SystemIdentifier()
self.mbrid.calculate_id()
self.mbrid.write_to_disk(
disk.storage_provider
)
# create first stage metadata to boot image
self._write_partition_id_config_to_boot_image(disk)
# run pre sync script hook
if self.system_setup.script_exists(
defaults.PRE_DISK_SYNC_SCRIPT
):
disk_system = SystemSetup(
self.xml_state, self.root_dir
self._write_recovery_metadata_to_boot_image()
self._write_raid_config_to_boot_image(raid_root)
self._write_generic_fstab_to_boot_image(device_map, system)
self.system_setup.export_modprobe_setup(
self.boot_image.boot_root_directory
)
disk_system.call_pre_disk_script()
# syncing system data to disk image
self._sync_system_to_image(
device_map, system, system_boot, system_efi, system_spare,
system_custom_parts
)
# create first stage metadata to system image
self._write_image_identifier_to_system_image()
# run post sync script hook
if self.system_setup.script_exists(
defaults.POST_DISK_SYNC_SCRIPT
):
image_system = ImageSystem(
device_map, self.root_dir,
system.get_volumes() if self.volume_manager_name else {}
self._write_crypttab_to_system_image(luks_root)
self._write_integritytab_to_system_image(self.integrity_root)
self._write_generic_fstab_to_system_image(device_map, system)
if self.initrd_system == 'dracut':
if self.root_filesystem_is_multipath is False:
self.boot_image.omit_module('multipath')
if self.root_filesystem_is_overlay:
self.boot_image.include_module('kiwi-overlay')
self.boot_image.write_system_config_file(
config={'modules': ['kiwi-overlay']}
)
if self.disk_resize_requested:
self.boot_image.include_module('kiwi-repart')
# create initrd
if self.boot_image.has_initrd_support():
self.boot_image.create_initrd(self.mbrid)
# create second stage metadata to system image
self._copy_first_boot_files_to_system_image()
self._write_bootloader_meta_data_to_system_image(
device_map, disk, system
)
image_system.mount()
disk_system = SystemSetup(
self.xml_state, image_system.mountpoint()
self.mbrid.write_to_disk(
disk.storage_provider
)
try:
disk_system.call_disk_script()
finally:
image_system.umount()
# install boot loader
self._install_bootloader(device_map, disk, system)
# run pre sync script hook
if self.system_setup.script_exists(
defaults.PRE_DISK_SYNC_SCRIPT
):
disk_system = SystemSetup(
self.xml_state, self.root_dir
)
disk_system.call_pre_disk_script()
# set root filesystem properties
self._setup_property_root_is_readonly_snapshot(system)
# syncing system data to disk image
self._sync_system_to_image(
device_map, system, system_boot, system_efi, system_spare,
system_custom_parts
)
# run post sync script hook
if self.system_setup.script_exists(
defaults.POST_DISK_SYNC_SCRIPT
):
image_system = ImageSystem(
device_map, self.root_dir,
system.get_volumes() if self.volume_manager_name else {}
)
image_system.mount()
disk_system = SystemSetup(
self.xml_state, image_system.mountpoint()
)
try:
disk_system.call_disk_script()
finally:
image_system.umount()
# install boot loader
self._install_bootloader(device_map, disk, system)
# set root filesystem properties
self._setup_property_root_is_readonly_snapshot(system)
Result.verify_image_size(
self.runtime_config.get_max_size_constraint(),
self.diskname
)
Result.verify_image_size(
self.runtime_config.get_max_size_constraint(),
self.diskname
)
# store image bundle_format in result
if self.bundle_format:
self.result.add_bundle_format(self.bundle_format)
@ -663,12 +666,12 @@ class DiskBuilder:
)
disk_format.resize_raw_disk(self.unpartitioned_bytes, append=True)
firmware = FirmWare(self.xml_state)
loop_provider = LoopDevice(disk_format.diskname)
loop_provider.create(overwrite=False)
partitioner = Partitioner.new(
firmware.get_partition_table_type(), loop_provider
)
partitioner.resize_table()
with LoopDevice(disk_format.diskname) as loop_provider:
loop_provider.create(overwrite=False)
partitioner = Partitioner.new(
firmware.get_partition_table_type(), loop_provider
)
partitioner.resize_table()
def create_install_media(self, result_instance: Result) -> Result:
"""
@ -1430,34 +1433,32 @@ class DiskBuilder:
if self.root_filesystem_embed_verity_metadata:
verity_root_file_bytes -= defaults.DM_METADATA_OFFSET
verity_root_file = Temporary().new_file()
loop_provider = LoopDevice(
verity_root_file.name,
int(verity_root_file_bytes / 1048576)
)
loop_provider.create()
filesystem_custom_parameters = {
'mount_options': self.custom_root_mount_args,
'create_options': self.custom_root_creation_args
}
filesystem = FileSystem.new(
self.requested_filesystem, loop_provider,
self.root_dir + '/',
filesystem_custom_parameters
)
filesystem.create_on_device(
label=self.disk_setup.get_root_label(),
uuid=BlockID(root_target).get_uuid()
)
filesystem.sync_data(
self._get_exclude_list_for_root_data_sync(device_map)
)
filesystem.umount()
filesystem.create_verity_layer(
self.root_filesystem_verity_blocks if
self.root_filesystem_verity_blocks != 'all' else None,
verity_root_file.name
)
del loop_provider
with LoopDevice(
verity_root_file.name, int(verity_root_file_bytes / 1048576)
) as loop_provider:
loop_provider.create()
filesystem_custom_parameters = {
'mount_options': self.custom_root_mount_args,
'create_options': self.custom_root_creation_args
}
filesystem = FileSystem.new(
self.requested_filesystem, loop_provider,
self.root_dir + '/',
filesystem_custom_parameters
)
filesystem.create_on_device(
label=self.disk_setup.get_root_label(),
uuid=BlockID(root_target).get_uuid()
)
filesystem.sync_data(
self._get_exclude_list_for_root_data_sync(device_map)
)
filesystem.umount()
filesystem.create_verity_layer(
self.root_filesystem_verity_blocks if
self.root_filesystem_verity_blocks != 'all' else None,
verity_root_file.name
)
log.info(
'--> Dumping rootfs file({0} bytes) -> {1}({2} bytes)'.format(
os.path.getsize(verity_root_file.name),

34
kiwi/builder/filesystem.py
View File

@ -170,26 +170,26 @@ class FileSystemBuilder:
def _operate_on_loop(self) -> None:
filesystem = None
loop_provider = LoopDevice(
with LoopDevice(
self.filename,
self.filesystem_setup.get_size_mbytes(),
self.blocksize
)
loop_provider.create()
filesystem = FileSystem.new(
self.requested_filesystem, loop_provider,
self.root_dir + os.sep, self.filesystem_custom_parameters
)
filesystem.create_on_device(self.label)
self.root_uuid = loop_provider.get_uuid(loop_provider.get_device())
log.info(
f'--> Syncing data to filesystem on {loop_provider.get_device()}'
)
filesystem.sync_data(
Defaults.
get_exclude_list_for_root_data_sync() + Defaults.
get_exclude_list_from_custom_exclude_files(self.root_dir)
)
) as loop_provider:
loop_provider.create()
filesystem = FileSystem.new(
self.requested_filesystem, loop_provider,
self.root_dir + os.sep, self.filesystem_custom_parameters
)
filesystem.create_on_device(self.label)
self.root_uuid = loop_provider.get_uuid(loop_provider.get_device())
log.info(
f'--> Syncing data to filesystem on {loop_provider.get_device()}'
)
filesystem.sync_data(
Defaults.
get_exclude_list_for_root_data_sync() + Defaults.
get_exclude_list_from_custom_exclude_files(self.root_dir)
)
def _operate_on_file(self) -> None:
default_provider = DeviceProvider()

38
kiwi/builder/live.py
View File

@ -255,28 +255,28 @@ class LiveImageBuilder:
self.xml_state, self.root_dir
)
root_image = Temporary().new_file()
loop_provider = LoopDevice(
with LoopDevice(
root_image.name,
filesystem_setup.get_size_mbytes(root_filesystem),
self.xml_state.build_type.get_target_blocksize()
)
loop_provider.create()
live_filesystem = FileSystem.new(
name=root_filesystem,
device_provider=loop_provider,
root_dir=self.root_dir + os.sep,
custom_args=filesystem_custom_parameters
)
live_filesystem.create_on_device()
log.info(
'--> Syncing data to {0} root image'.format(root_filesystem)
)
live_filesystem.sync_data(
Defaults.
get_exclude_list_for_root_data_sync() + Defaults.
get_exclude_list_from_custom_exclude_files(self.root_dir)
)
live_filesystem.umount()
) as loop_provider:
loop_provider.create()
live_filesystem = FileSystem.new(
name=root_filesystem,
device_provider=loop_provider,
root_dir=self.root_dir + os.sep,
custom_args=filesystem_custom_parameters
)
live_filesystem.create_on_device()
log.info(
'--> Syncing data to {0} root image'.format(root_filesystem)
)
live_filesystem.sync_data(
Defaults.
get_exclude_list_for_root_data_sync() + Defaults.
get_exclude_list_from_custom_exclude_files(self.root_dir)
)
live_filesystem.umount()
log.info('--> Creating squashfs container for root image')
self.live_container_dir = Temporary(

18
kiwi/storage/loop_device.py
View File

@ -23,9 +23,7 @@ from kiwi.command import Command
from kiwi.storage.device_provider import DeviceProvider
from kiwi.utils.command_capabilities import CommandCapabilities
from kiwi.exceptions import (
KiwiLoopSetupError
)
from kiwi.exceptions import KiwiLoopSetupError
log = logging.getLogger('kiwi')
@ -51,6 +49,9 @@ class LoopDevice(DeviceProvider):
self.filesize_mbytes = filesize_mbytes
self.blocksize_bytes = blocksize_bytes
def __enter__(self):
return self
def get_device(self) -> str:
"""
Device node name
@ -98,12 +99,13 @@ class LoopDevice(DeviceProvider):
)
self.node_name = loop_call.output.rstrip(os.linesep)
def __del__(self):
def __exit__(self, exc_type, exc_value, traceback):
if self.node_name:
log.info('Cleaning up %s instance', type(self).__name__)
try:
Command.run(['losetup', '-d', self.node_name])
except Exception:
log.warning(
'loop device %s still busy', self.node_name
except Exception as issue:
log.error(
'loop cleanup on {0} failed with: {1}'.format(
self.node_name, issue
)
)

13
kiwi/tasks/image_resize.py
View File

@ -120,13 +120,12 @@ class ImageResizeTask(CliTask):
# resize raw disk partition table
firmware = FirmWare(self.xml_state)
loop_provider = LoopDevice(image_format.diskname)
loop_provider.create(overwrite=False)
partitioner = Partitioner.new(
firmware.get_partition_table_type(), loop_provider
)
partitioner.resize_table()
del loop_provider
with LoopDevice(image_format.diskname) as loop_provider:
loop_provider.create(overwrite=False)
partitioner = Partitioner.new(
firmware.get_partition_table_type(), loop_provider
)
partitioner.resize_table()
# resize disk format from resized raw disk
if disk_format and resize_result is True:

38
test/unit/builder/disk_test.py
View File

@ -79,10 +79,6 @@ class TestDiskBuilder:
kiwi.builder.disk.BlockID = Mock(
return_value=self.block_operation
)
self.loop_provider = Mock()
kiwi.builder.disk.LoopDevice = Mock(
return_value=self.loop_provider
)
self.disk = Mock()
provider = Mock()
provider.get_device = Mock(
@ -290,6 +286,7 @@ class TestDiskBuilder:
{'signing_keys': ['key_file_a', 'key_file_b']}
)
@patch('kiwi.builder.disk.LoopDevice')
@patch('kiwi.builder.disk.FileSystem.new')
@patch('random.randrange')
@patch('kiwi.builder.disk.Command.run')
@ -298,8 +295,10 @@ class TestDiskBuilder:
@patch('os.path.exists')
def test_create_disk_standard_root_with_kiwi_initrd(
self, mock_path, mock_ImageSystem, mock_grub_dir,
mock_command, mock_rand, mock_fs
mock_command, mock_rand, mock_fs, mock_LoopDevice
):
loop_provider = Mock()
mock_LoopDevice.return_value.__enter__.return_value = loop_provider
mock_path.return_value = True
mock_rand.return_value = 15
filesystem = Mock()
@ -320,7 +319,7 @@ class TestDiskBuilder:
self.disk_setup.get_disksize_mbytes.assert_called_once_with(
root_clone=0, boot_clone=0
)
self.loop_provider.create.assert_called_once_with()
loop_provider.create.assert_called_once_with()
self.disk.wipe.assert_called_once_with()
self.disk.create_efi_csm_partition.assert_called_once_with(
self.firmware.get_legacy_bios_partition_size()
@ -638,6 +637,7 @@ class TestDiskBuilder:
call(label='SWAP')
]
@patch('kiwi.builder.disk.LoopDevice')
@patch('kiwi.builder.disk.FileSystem.new')
@patch('random.randrange')
@patch('kiwi.builder.disk.Command.run')
@ -651,8 +651,11 @@ class TestDiskBuilder:
def test_create_disk_standard_root_with_dracut_initrd(
self, mock_Temporary_new_file, mock_VeritySetup,
mock_ImageSystem, mock_SystemSetup, mock_os_path_getsize,
mock_path, mock_grub_dir, mock_command, mock_rand, mock_fs
mock_path, mock_grub_dir, mock_command, mock_rand, mock_fs,
mock_LoopDevice
):
loop_provider = Mock()
mock_LoopDevice.return_value.__enter__.return_value = loop_provider
tempfile = Mock()
tempfile.name = 'tempfile'
mock_Temporary_new_file.return_value = tempfile
@ -682,7 +685,7 @@ class TestDiskBuilder:
self.disk_setup.get_disksize_mbytes.assert_called_once_with(
root_clone=0, boot_clone=0
)
assert self.loop_provider.create.call_args_list == [
assert loop_provider.create.call_args_list == [
call(), call()
]
self.disk.wipe.assert_called_once_with()
@ -829,6 +832,7 @@ class TestDiskBuilder:
'clone:1024:1024', 1
)
@patch('kiwi.builder.disk.LoopDevice')
@patch('kiwi.builder.disk.DeviceProvider')
@patch('kiwi.builder.disk.FileSystem.new')
@patch('kiwi.builder.disk.FileSystemSquashFs')
@ -842,8 +846,11 @@ class TestDiskBuilder:
def test_create_disk_standard_root_is_overlay(
self, mock_BlockID, mock_rand, mock_temp,
mock_getsize, mock_exists, mock_grub_dir, mock_command,
mock_squashfs, mock_fs, mock_DeviceProvider
mock_squashfs, mock_fs, mock_DeviceProvider,
mock_LoopDevice
):
loop_provider = Mock()
mock_LoopDevice.return_value.__enter__.return_value = loop_provider
block_operation = Mock()
block_operation.get_blkid.return_value = 'partuuid'
block_operation.get_filesystem.return_value = 'ext3'
@ -924,11 +931,14 @@ class TestDiskBuilder:
config={'modules': ['kiwi-overlay']}
)
@patch('kiwi.builder.disk.LoopDevice')
@patch('kiwi.builder.disk.FileSystem.new')
@patch('kiwi.builder.disk.Command.run')
def test_create_disk_standard_root_no_hypervisor_found(
self, mock_command, mock_fs
self, mock_command, mock_fs, mock_LoopDevice
):
loop_provider = Mock()
mock_LoopDevice.return_value.__enter__.return_value = loop_provider
self.kernel.get_xen_hypervisor.return_value = False
self.disk_builder.volume_manager_name = None
self.disk_builder.xen_server = True
@ -1417,10 +1427,10 @@ class TestDiskBuilder:
@patch('kiwi.builder.disk.Partitioner.new')
@patch('kiwi.builder.disk.DiskFormat.new')
def test_append_unpartitioned_space(
self, mock_diskformat, mock_partitioner, mock_loopdevice
self, mock_diskformat, mock_partitioner, mock_LoopDevice
):
loopdevice = Mock()
mock_loopdevice.return_value = loopdevice
loop_provider = Mock()
mock_LoopDevice.return_value.__enter__.return_value = loop_provider
partitioner = Mock()
mock_partitioner.return_value = partitioner
disk_format = Mock()
@ -1430,7 +1440,7 @@ class TestDiskBuilder:
disk_format.resize_raw_disk.assert_called_once_with(
1024, append=True
)
loopdevice.create.assert_called_once_with(overwrite=False)
loop_provider.create.assert_called_once_with(overwrite=False)
assert partitioner.resize_table.called
@patch('kiwi.builder.disk.FileSystem.new')

19
test/unit/builder/filesystem_test.py
View File

@ -17,11 +17,6 @@ class TestFileSystemBuilder:
@patch('kiwi.builder.filesystem.FileSystemSetup')
def setup(self, mock_fs_setup):
Defaults.set_platform_name('x86_64')
self.loop_provider = Mock()
self.loop_provider.get_device = Mock(
return_value='/dev/loop1'
)
self.loop_provider.create = Mock()
self.filesystem = Mock()
self.filesystem.create_on_device = Mock()
@ -96,22 +91,26 @@ class TestFileSystemBuilder:
@patch('kiwi.builder.filesystem.FileSystem.new')
@patch('kiwi.builder.filesystem.FileSystemSetup')
def test_create_on_loop(
self, mock_fs_setup, mock_fs, mock_loop
self, mock_fs_setup, mock_fs, mock_LoopDevice
):
Defaults.set_platform_name('x86_64')
mock_fs_setup.return_value = self.fs_setup
mock_fs.return_value = self.filesystem
mock_loop.return_value = self.loop_provider
loop_provider = Mock()
loop_provider.get_device = Mock(
return_value='/dev/loop1'
)
mock_LoopDevice.return_value.__enter__.return_value = loop_provider
fs = FileSystemBuilder(
self.xml_state, 'target_dir', 'root_dir'
)
fs.create()
mock_loop.assert_called_once_with(
mock_LoopDevice.assert_called_once_with(
'target_dir/myimage.x86_64-1.2.3.ext3', 42, 4096
)
self.loop_provider.create.assert_called_once_with()
loop_provider.create.assert_called_once_with()
mock_fs.assert_called_once_with(
'ext3', self.loop_provider, 'root_dir/', {
'ext3', loop_provider, 'root_dir/', {
'mount_options': ['async'],
'create_options': ['-O', 'option']
}

16
test/unit/builder/live_test.py
View File

@ -34,11 +34,6 @@ class TestLiveImageBuilder:
return_value=self.filesystem_setup
)
self.loop = Mock()
kiwi.builder.live.LoopDevice = Mock(
return_value=self.loop
)
self.bootloader = Mock()
kiwi.builder.live.BootLoaderConfig.new = Mock(
return_value=self.bootloader
@ -133,6 +128,7 @@ class TestLiveImageBuilder:
)
assert live_image.arch == 'ix86'
@patch('kiwi.builder.live.LoopDevice')
@patch('kiwi.builder.live.Command.run')
@patch('kiwi.builder.live.DeviceProvider')
@patch('kiwi.builder.live.IsoToolsBase.setup_media_loader_directory')
@ -150,7 +146,7 @@ class TestLiveImageBuilder:
self, mock_chmod, mock_exists, mock_grub_dir, mock_size,
mock_filesystem, mock_isofs, mock_Iso, mock_tag, mock_shutil,
mock_Temporary, mock_setup_media_loader_directory, mock_DeviceProvider,
mock_Command_run
mock_Command_run, mock_LoopDevice
):
boot_names = Mock()
boot_names.initrd_name = 'dracut_initrd_name'
@ -169,6 +165,7 @@ class TestLiveImageBuilder:
['mv', 'kiwi_used_initrd_name', 'root_dir/boot/dracut_initrd_name']
)
@patch('kiwi.builder.live.LoopDevice')
@patch('kiwi.builder.live.DeviceProvider')
@patch('kiwi.builder.live.IsoToolsBase.setup_media_loader_directory')
@patch('kiwi.builder.live.Temporary')
@ -184,8 +181,11 @@ class TestLiveImageBuilder:
def test_create_overlay_structure_boot_on_grub(
self, mock_chmod, mock_exists, mock_grub_dir, mock_size,
mock_filesystem, mock_isofs, mock_Iso, mock_tag, mock_shutil,
mock_Temporary, mock_setup_media_loader_directory, mock_DeviceProvider
mock_Temporary, mock_setup_media_loader_directory, mock_DeviceProvider,
mock_LoopDevice
):
loop_provider = Mock()
mock_LoopDevice.return_value.__enter__.return_value = loop_provider
mock_exists.return_value = True
mock_grub_dir.return_value = 'grub2'
@ -229,7 +229,7 @@ class TestLiveImageBuilder:
assert kiwi.builder.live.FileSystem.new.call_args_list == [
call(
device_provider=self.loop, name='ext4',
device_provider=loop_provider, name='ext4',
root_dir='root_dir/',
custom_args={
'mount_options': ['async'],

26
test/unit/storage/loop_device_test.py
View File

@ -1,5 +1,7 @@
import logging
from mock import patch
from mock import (
patch, call
)
from pytest import (
raises, fixture
)
@ -64,12 +66,16 @@ class TestLoopDevice:
self.loop.node_name = None
@patch('kiwi.storage.loop_device.Command.run')
def test_destructor(self, mock_command):
self.loop.node_name = '/dev/loop0'
mock_command.side_effect = Exception
self.loop.__del__()
with self._caplog.at_level(logging.WARNING):
mock_command.assert_called_once_with(
['losetup', '-d', '/dev/loop0']
)
self.loop.node_name = None
@patch('os.path.exists')
def test_context_manager_exit(self, mock_os_path_exists, mock_command_run):
mock_os_path_exists.return_value = True
mock_command_run.side_effect = Exception
with self._caplog.at_level(logging.ERROR):
with LoopDevice('loop-file', 20) as loop_provider:
loop_provider.node_name = '/dev/loop0'
with raises(Exception):
loop_provider.create(overwrite=False)
assert mock_command_run.call_args_list == [
call(['losetup', '-f', '--show', 'loop-file']),
call(['losetup', '-d', '/dev/loop0'])
]

36
test/unit/tasks/image_resize_test.py
View File

@ -41,13 +41,9 @@ class TestImageResizeTask:
self.firmware.get_partition_table_type = Mock(
return_value='gpt'
)
self.loop_provider = Mock()
kiwi.tasks.image_resize.FirmWare = Mock(
return_value=self.firmware
)
kiwi.tasks.image_resize.LoopDevice = Mock(
return_value=self.loop_provider
)
self.task = ImageResizeTask()
@ -94,9 +90,14 @@ class TestImageResizeTask:
with raises(KiwiSizeError):
self.task.process()
@patch('kiwi.tasks.image_resize.LoopDevice')
@patch('kiwi.tasks.image_resize.DiskFormat.new')
@patch('kiwi.tasks.image_resize.Partitioner.new')
def test_process_image_resize_gb(self, mock_Partitioner, mock_DiskFormat):
def test_process_image_resize_gb(
self, mock_Partitioner, mock_DiskFormat, mock_LoopDevice
):
loop_provider = Mock()
mock_LoopDevice.return_value.__enter__.return_value = loop_provider
partitioner = Mock()
mock_Partitioner.return_value = partitioner
image_format = Mock()
@ -105,16 +106,21 @@ class TestImageResizeTask:
self._init_command_args()
self.task.command_args['resize'] = True
self.task.process()
self.loop_provider.create.assert_called_once_with(overwrite=False)
loop_provider.create.assert_called_once_with(overwrite=False)
partitioner.resize_table.assert_called_once_with()
image_format.resize_raw_disk.assert_called_once_with(
42 * 1024 * 1024 * 1024
)
image_format.create_image_format.assert_called_once_with()
@patch('kiwi.tasks.image_resize.LoopDevice')
@patch('kiwi.tasks.image_resize.DiskFormat.new')
@patch('kiwi.tasks.image_resize.Partitioner.new')
def test_process_image_resize_mb(self, mock_Partitioner, mock_DiskFormat):
def test_process_image_resize_mb(
self, mock_Partitioner, mock_DiskFormat, mock_LoopDevice
):
loop_provider = Mock()
mock_LoopDevice.return_value.__enter__.return_value = loop_provider
partitioner = Mock()
mock_Partitioner.return_value = partitioner
image_format = Mock()
@ -124,18 +130,21 @@ class TestImageResizeTask:
self.task.command_args['resize'] = True
self.task.command_args['--size'] = '42m'
self.task.process()
self.loop_provider.create.assert_called_once_with(overwrite=False)
loop_provider.create.assert_called_once_with(overwrite=False)
partitioner.resize_table.assert_called_once_with()
image_format.resize_raw_disk.assert_called_once_with(
42 * 1024 * 1024
)
image_format.create_image_format.assert_called_once_with()
@patch('kiwi.tasks.image_resize.LoopDevice')
@patch('kiwi.tasks.image_resize.DiskFormat.new')
@patch('kiwi.tasks.image_resize.Partitioner.new')
def test_process_image_resize_bytes(
self, mock_Partitioner, mock_DiskFormat
self, mock_Partitioner, mock_DiskFormat, mock_LoopDevice
):
loop_provider = Mock()
mock_LoopDevice.return_value.__enter__.return_value = loop_provider
partitioner = Mock()
mock_Partitioner.return_value = partitioner
image_format = Mock()
@ -145,18 +154,21 @@ class TestImageResizeTask:
self.task.command_args['resize'] = True
self.task.command_args['--size'] = '42'
self.task.process()
self.loop_provider.create.assert_called_once_with(overwrite=False)
loop_provider.create.assert_called_once_with(overwrite=False)
partitioner.resize_table.assert_called_once_with()
image_format.resize_raw_disk.assert_called_once_with(
42
)
image_format.create_image_format.assert_called_once_with()
@patch('kiwi.tasks.image_resize.LoopDevice')
@patch('kiwi.tasks.image_resize.DiskFormat.new')
@patch('kiwi.tasks.image_resize.Partitioner.new')
def test_process_image_resize_not_needed(
self, mock_Partitioner, mock_DiskFormat
self, mock_Partitioner, mock_DiskFormat, mock_LoopDevice
):
loop_provider = Mock()
mock_LoopDevice.return_value.__enter__.return_value = loop_provider
partitioner = Mock()
mock_Partitioner.return_value = partitioner
image_format = Mock()
@ -167,7 +179,7 @@ class TestImageResizeTask:
self.task.command_args['--size'] = '42'
with self._caplog.at_level(logging.INFO):
self.task.process()
self.loop_provider.create.assert_called_once_with(overwrite=False)
loop_provider.create.assert_called_once_with(overwrite=False)
partitioner.resize_table.assert_called_once_with()
image_format.resize_raw_disk.assert_called_once_with(
42