diff --git a/kiwi/bootloader/config/grub2.py b/kiwi/bootloader/config/grub2.py
index 5a73b3ffaf8..7f30b8e4b56 100644
--- a/kiwi/bootloader/config/grub2.py
+++ b/kiwi/bootloader/config/grub2.py
@@ -567,7 +567,7 @@ def _copy_grub_config_to_efi_path(
grub_image = Defaults.get_signed_grub_loader(self.root_dir, target_type)
if grub_image and grub_image.filename:
bash_command = [
- 'strings', grub_image.filename, '|', 'grep', 'EFI\/'
+ 'strings', grub_image.filename, '|', 'grep', r'EFI\/'
]
efi_boot_path = Command.run(
['bash', '-c', ' '.join(bash_command)],
diff --git a/kiwi/builder/disk.py b/kiwi/builder/disk.py
index 5c6fdd7f8e9..52a205e5359 100644
--- a/kiwi/builder/disk.py
+++ b/kiwi/builder/disk.py
@@ -15,6 +15,7 @@
# You should have received a copy of the GNU General Public License
# along with kiwi. If not, see
#
+from contextlib import ExitStack
import os
import sys
import logging
@@ -732,35 +733,28 @@ def _process_build(self, device_map: Dict, disk: Disk) -> None:
raid_root = self.storage_map['raid_root']
integrity_root = self.storage_map['integrity_root']
try:
- self._build_main_system(
- device_map,
- disk,
- system,
- system_boot,
- system_efi,
- system_spare,
- system_custom_parts,
- luks_root,
- raid_root,
- integrity_root
- )
+ with ExitStack() as stack:
+ self._build_main_system(
+ stack,
+ device_map,
+ disk,
+ system,
+ system_boot,
+ system_efi,
+ system_spare,
+ system_custom_parts,
+ luks_root,
+ raid_root,
+ integrity_root
+ )
finally:
- for map_name in sorted(system_custom_parts.keys()):
- system_custom_parts[map_name].umount()
- if system_efi:
- system_efi.umount()
- if system_spare:
- system_spare.umount()
- if system_boot:
- system_boot.umount()
if system:
if self.volume_manager_name:
system.umount_volumes()
- else:
- system.umount()
def _build_main_system(
self,
+ stack: ExitStack,
device_map: Dict,
disk: Disk,
system: Optional[Union[FileSystemBase, VolumeManagerBase]],
@@ -861,7 +855,7 @@ def _build_main_system(
# syncing system data to disk image
self._sync_system_to_image(
- device_map, system, system_boot, system_efi, system_spare,
+ stack, device_map, system, system_boot, system_efi, system_spare,
system_custom_parts, integrity_root
)
@@ -1507,7 +1501,9 @@ def _write_bootloader_meta_data_to_system_image(
)
def _sync_system_to_image(
- self, device_map: Dict,
+ self,
+ stack: ExitStack,
+ device_map: Dict,
system: Optional[Union[FileSystemBase, VolumeManagerBase]],
system_boot: Optional[FileSystemBase],
system_efi: Optional[FileSystemBase],
@@ -1518,13 +1514,13 @@ def _sync_system_to_image(
log.info('Syncing system to image')
if system_spare:
log.info('--> Syncing spare partition data')
- system_spare.sync_data()
+ stack.push(system_spare.sync_data())
for map_name in sorted(system_custom_parts.keys()):
system_custom_part = system_custom_parts[map_name]
log.info('--> Syncing custom partition(s) data')
if not system_custom_part.filename:
- system_custom_part.sync_data()
+ stack.push(system_custom_part.sync_data())
if device_map.get(f'{map_name}clone1'):
log.info(
f'--> Dumping {map_name!r} clone data at extra partition'
@@ -1540,13 +1536,13 @@ def _sync_system_to_image(
if system_efi:
log.info('--> Syncing EFI boot data to EFI partition')
- system_efi.sync_data()
+ stack.push(system_efi.sync_data())
if system_boot:
log.info('--> Syncing boot data at extra partition')
- system_boot.sync_data(
+ stack.push(system_boot.sync_data(
self._get_exclude_list_for_boot_data_sync()
- )
+ ))
if device_map.get('bootclone1'):
log.info(
'--> Dumping boot clone data at extra partition'
@@ -1679,9 +1675,11 @@ def _sync_system_to_image(
self._get_clone_devices('rootclone', device_map)
)
elif system:
- system.sync_data(
+ system_mount = system.sync_data(
self._get_exclude_list_for_root_data_sync(device_map)
)
+ if system_mount:
+ stack.push(system_mount)
if device_map.get('rootclone1'):
log.info(
'--> Dumping root clone data at extra partition'
diff --git a/kiwi/filesystem/base.py b/kiwi/filesystem/base.py
index 2a497fa3c60..724ab9eedfe 100644
--- a/kiwi/filesystem/base.py
+++ b/kiwi/filesystem/base.py
@@ -175,11 +175,13 @@ def get_mountpoint(self) -> Optional[str]:
return self.filesystem_mount.mountpoint
return None
- def sync_data(self, exclude: List[str] = []):
+ def sync_data(self, exclude: List[str] = []) -> MountManager:
"""
Copy data tree into filesystem
:param list exclude: list of exclude dirs/files
+ :return: The mount created for syncing data. It should be used to
+ un-mount the filesystem again.
"""
if not self.root_dir:
raise KiwiFileSystemSyncError(
@@ -202,6 +204,7 @@ def sync_data(self, exclude: List[str] = []):
data.sync_data(
exclude=exclude, options=Defaults.get_sync_options()
)
+ return self.filesystem_mount
def create_verity_layer(
self, blocks: Optional[int] = None, filename: str = None
diff --git a/kiwi/mount_manager.py b/kiwi/mount_manager.py
index 91397fd6f12..fc6d0314d86 100644
--- a/kiwi/mount_manager.py
+++ b/kiwi/mount_manager.py
@@ -27,7 +27,7 @@
from kiwi.path import Path
from kiwi.utils.temporary import Temporary
from kiwi.command import Command
-from kiwi.exceptions import KiwiUmountBusyError
+from kiwi.exceptions import KiwiCommandError, KiwiUmountBusyError
log = logging.getLogger('kiwi')
@@ -36,10 +36,11 @@ class MountManager:
"""
**Implements methods for mounting, umounting and mount checking**
- If a MountManager instance is used to mount a device the caller
- must care for the time when umount needs to be called. The class
- does not automatically release the mounted device, which is
- intentional
+ The caller is responsible for unmounting the device if the MountManager is
+ used as is.
+
+ The class also supports to be used as a context manager, where the device is
+ unmounted once the context manager's with block is left
* :param string device: device node name
* :param string mountpoint: mountpoint directory name
@@ -60,6 +61,12 @@ def __init__(
Path.create(mountpoint)
self.mountpoint = mountpoint
+ def __enter__(self) -> "MountManager":
+ return self
+
+ def __exit__(self, exc_type, exc_value, traceback) -> None:
+ self.umount()
+
def get_attributes(self) -> Dict[str, str]:
"""
Return attributes dict for this mount manager
@@ -148,10 +155,12 @@ def umount(self, raise_on_busy: bool = True) -> bool:
Command.run(['umount', self.mountpoint])
umounted_successfully = True
break
- except Exception:
+ # we only want to catch KiwiCommandError, everything else
+ # indicates either a bug in the code or a bigger problem, like
+ # umount not being available
+ except KiwiCommandError as kw_err:
log.warning(
- '%d umount of %s failed, try again in 1sec',
- busy, self.mountpoint
+ f'{busy} umount of {self.mountpoint} failed with {kw_err}, try again in 1sec',
)
time.sleep(1)
if not umounted_successfully:
diff --git a/kiwi/volume_manager/base.py b/kiwi/volume_manager/base.py
index f1e6dd9f354..94fa038f001 100644
--- a/kiwi/volume_manager/base.py
+++ b/kiwi/volume_manager/base.py
@@ -16,7 +16,7 @@
# along with kiwi. If not, see
#
from collections import namedtuple
-from typing import Optional
+from typing import Any, Dict, List, Optional
import logging
import os
@@ -29,6 +29,7 @@
from kiwi.path import Path
from kiwi.system.size import SystemSize
from kiwi.defaults import Defaults
+from kiwi.xml_state import volume_type
from kiwi.exceptions import (
KiwiVolumeManagerSetupError
@@ -41,31 +42,32 @@ class VolumeManagerBase(DeviceProvider):
"""
**Implements base class for volume management interface**
- :param str mountpoint: root mountpoint for volumes
:param object device_map:
dictionary of low level DeviceProvider intances
:param str root_dir: root directory path name
:param list volumes: list of volumes from :class:`XMLState::get_volumes()`
- :param str volume_group: volume group name
- :param map volume_map: map volume name to device node
- :param list mount_list: list of volume MountManager's
- :param str device: storage device node name
:param dict custom_args: custom volume manager arguments for all
volume manager and filesystem specific tasks
- :param list custom_filesystem_args: custom filesystem creation and mount
- arguments, subset of the custom_args information suitable to
- be passed to a FileSystem instance
:raises KiwiVolumeManagerSetupError: if the given root_dir doesn't exist
"""
- def __init__(self, device_map, root_dir, volumes, custom_args=None):
- self.temp_directories = []
+
+ def __init__(
+ self,
+ device_map: Dict[str, DeviceProvider],
+ root_dir: str,
+ volumes: List[volume_type],
+ custom_args: Optional[Dict[str, Any]] = None
+ ) -> None:
+
+ self.temp_directories: List[Temporary] = []
# all volumes are combined into one mountpoint. This is
# needed at sync_data time. How to mount the volumes is
# special to the volume management class
- self.mountpoint = None
+ #: root mountpoint for volumes
+ self.mountpoint: Optional[str] = None
- # dictionary of mapped DeviceProviders
+ #: dictionary of mapped DeviceProviders
self.device_map = device_map
# bind the device providing class instance to this object.
@@ -73,17 +75,22 @@ def __init__(self, device_map, root_dir, volumes, custom_args=None):
# the device should be released.
self.device_provider_root = device_map['root']
- # An indicator for the mount of the filesystem and its volumes
- # when mounted for the first time
+ #: An indicator for the mount of the filesystem and its volumes
+ #: when mounted for the first time
self.volumes_mounted_initially = False
+ #: root directory path name
self.root_dir = root_dir
+ #: list of volumes from :class:`XMLState::get_volumes()`
self.volumes = volumes
+ #: volume group name
self.volume_group = None
- self.volume_map = {}
- self.mount_list = []
+ #: map volume name to device node
+ self.volume_map: Dict[str, str] = {}
+ #: list of volume MountManager's
+ self.mount_list: List[MountManager] = []
- # Main device to operate on
+ #: main storage device node name
self.device = self.device_provider_root.get_device()
if not os.path.exists(root_dir):
@@ -91,8 +98,11 @@ def __init__(self, device_map, root_dir, volumes, custom_args=None):
'given root directory %s does not exist' % root_dir
)
- self.custom_args = {}
- self.custom_filesystem_args = {
+ self.custom_args: Dict[str, Any] = {}
+
+ #: custom filesystem creation and mount arguments, subset of the
+ #: custom_args information suitable to be passed to a FileSystem instance
+ self.custom_filesystem_args: Dict[str, Any] = {
'create_options': [],
'mount_options': []
}
@@ -121,7 +131,7 @@ def post_init(self, custom_args):
"""
pass
- def setup(self, name=None):
+ def setup(self, name: str = None):
"""
Implements setup required prior to the creation of volumes
@@ -154,7 +164,7 @@ def apply_attributes_on_volume(self, toplevel, volume):
]
)
- def get_fstab(self, persistency_type, filesystem_name):
+ def get_fstab(self, persistency_type: str, filesystem_name: str) -> List[str]:
"""
Implements setup of the fstab entries. The method should
return a list of fstab compatible entries
@@ -274,7 +284,7 @@ def get_canonical_volume_list(self):
def get_volume_mbsize(
self, volume, all_volumes, filesystem_name, resize_on_boot=False
- ):
+ ) -> int:
"""
Implements size lookup for the given path and desired
filesystem according to the specified size type
@@ -335,7 +345,7 @@ def get_volume_mbsize(
)
return mbsize
- def get_mountpoint(self):
+ def get_mountpoint(self) -> Optional[str]:
"""
Provides mount point directory
@@ -360,20 +370,26 @@ def get_root_volume_name(self) -> str:
"""
return '/'
- def sync_data(self, exclude=None):
+ def sync_data(self, exclude: Optional[List[str]] = None) -> Optional[MountManager]:
"""
Implements sync of root directory to mounted volumes
:param list exclude: file patterns to exclude
+
+ :return: If a mount was created, then a context manager implementing the
+ unmount is returned.
"""
- if self.mountpoint:
- root_mount = MountManager(device=None, mountpoint=self.mountpoint)
- if not root_mount.is_mounted():
- self.mount_volumes()
- data = DataSync(self.root_dir, self.mountpoint)
- data.sync_data(
- options=Defaults.get_sync_options(), exclude=exclude
- )
+ if not self.mountpoint:
+ return None
+
+ root_mount = MountManager(device="", mountpoint=self.mountpoint)
+ if not root_mount.is_mounted():
+ self.mount_volumes()
+ data = DataSync(self.root_dir, self.mountpoint)
+ data.sync_data(
+ options=Defaults.get_sync_options(), exclude=exclude or []
+ )
+ return root_mount
def create_verity_layer(
self, blocks: Optional[int] = None, filename: str = None
diff --git a/kiwi/volume_manager/btrfs.py b/kiwi/volume_manager/btrfs.py
index e504373cc94..c1342d6feb9 100644
--- a/kiwi/volume_manager/btrfs.py
+++ b/kiwi/volume_manager/btrfs.py
@@ -57,7 +57,7 @@ def post_init(self, custom_args):
Store custom btrfs initialization arguments
- :param list custom_args: custom btrfs volume manager arguments
+ :param dict custom_args: custom btrfs volume manager arguments
"""
if custom_args:
self.custom_args = custom_args
@@ -407,6 +407,8 @@ def get_mountpoint(self) -> str:
:rtype: string
"""
+ if not self.mountpoint:
+ raise KiwiVolumeManagerSetupError("No mountpoint exists")
sync_target: List[str] = [self.mountpoint]
if self.root_volume_name != '/':
sync_target.append(self.root_volume_name)
diff --git a/test/unit/mount_manager_test.py b/test/unit/mount_manager_test.py
index f976f27d97a..c3eb28b4a72 100644
--- a/test/unit/mount_manager_test.py
+++ b/test/unit/mount_manager_test.py
@@ -1,11 +1,12 @@
import logging
+from typing import NoReturn
from pytest import (
fixture, raises
)
from mock import (
patch, call, Mock
)
-from kiwi.exceptions import KiwiUmountBusyError
+from kiwi.exceptions import KiwiCommandError, KiwiUmountBusyError
from kiwi.mount_manager import MountManager
@@ -78,6 +79,16 @@ def test_mount(self, mock_mounted, mock_command):
['mount', '-o', 'options', '/dev/some-device', '/some/mountpoint']
)
+ @patch('kiwi.mount_manager.Command.run')
+ @patch('kiwi.mount_manager.MountManager.is_mounted')
+ def test_context_manager(self, mock_mounted, mock_command):
+ mock_mounted.return_value = True
+ with self.mount_manager:
+ pass
+ mock_command.assert_called_once_with(
+ ['umount', '/some/mountpoint']
+ )
+
@patch('kiwi.mount_manager.Command.run')
@patch('kiwi.mount_manager.MountManager.is_mounted')
def test_umount_lazy(self, mock_mounted, mock_command):
@@ -93,7 +104,10 @@ def test_umount_lazy(self, mock_mounted, mock_command):
def test_umount_with_errors(
self, mock_sleep, mock_mounted, mock_command
):
- mock_command.side_effect = Exception
+ def _cmd_err(args) -> NoReturn:
+ raise KiwiCommandError("foo")
+
+ mock_command.side_effect = _cmd_err
mock_mounted.return_value = True
with self._caplog.at_level(logging.WARNING):
assert self.mount_manager.umount(raise_on_busy=False) is False
@@ -119,7 +133,7 @@ def test_umount_with_errors_raises_no_lsof_present(
):
def command_call(args):
if 'umount' in args:
- raise Exception
+ raise KiwiCommandError("foo")
mock_Path_which.return_value = None
mock_command.side_effect = command_call
@@ -136,7 +150,7 @@ def test_umount_with_errors_raises_lsof_present(
):
def command_call(args, raise_on_error=None):
if 'umount' in args:
- raise Exception
+ raise KiwiCommandError("foo")
else:
call_return = Mock()
call_return.output = 'HEADLINE\ndata'
diff --git a/test/unit/volume_manager/base_test.py b/test/unit/volume_manager/base_test.py
index 9aec10e4423..570cfbb86e2 100644
--- a/test/unit/volume_manager/base_test.py
+++ b/test/unit/volume_manager/base_test.py
@@ -254,6 +254,12 @@ def test_sync_data(self, mock_mount_volumes, mock_mounted, mock_sync):
)
assert self.volume_manager.get_mountpoint() == 'mountpoint'
+ @patch('kiwi.volume_manager.base.MountManager.is_mounted')
+ def test_sync_data_without_mountpoint(self, mock_mounted):
+ self.volume_manager.mountpoint = ''
+ assert self.volume_manager.sync_data() is None
+ mock_mounted.assert_not_called()
+
def test_create_verity_layer(self):
with raises(NotImplementedError):
self.volume_manager.create_verity_layer()
diff --git a/test/unit/volume_manager/btrfs_test.py b/test/unit/volume_manager/btrfs_test.py
index f0f48d70089..4e3db759b12 100644
--- a/test/unit/volume_manager/btrfs_test.py
+++ b/test/unit/volume_manager/btrfs_test.py
@@ -9,6 +9,8 @@
from lxml import etree
from xml.dom import minidom
+import pytest
+
from kiwi.volume_manager.btrfs import VolumeManagerBtrfs
from kiwi.xml_state import volume_type
@@ -84,6 +86,11 @@ def test_post_init_root_is_snapshot_without_root_volume(self):
self.volume_manager.post_init({'root_is_snapshot': True})
assert self.volume_manager.custom_args['root_is_snapshot'] is False
+ def test_fails_to_get_mountpoint_without_mountpoint(self):
+ self.volume_manager.mountpoint = ''
+ with pytest.raises(KiwiVolumeManagerSetupError):
+ self.volume_manager.get_mountpoint()
+
@patch('os.path.exists')
@patch('kiwi.volume_manager.btrfs.Command.run')
@patch('kiwi.volume_manager.btrfs.FileSystem.new')