diff --git a/kiwi/bootloader_install_grub2.py b/kiwi/bootloader_install_grub2.py
index 7924ead2..1db1289f 100644
--- a/kiwi/bootloader_install_grub2.py
+++ b/kiwi/bootloader_install_grub2.py
@@ -15,15 +15,12 @@
# You should have received a copy of the GNU General Public License
# along with kiwi. If not, see
#
-from tempfile import mkdtemp
-import time
-
# project
from .bootloader_install_base import BootLoaderInstallBase
from .command import Command
from .logger import log
-from .path import Path
from .defaults import Defaults
+from .mount_manager import MountManager
from .exceptions import(
KiwiBootLoaderGrubInstallError
@@ -45,14 +42,12 @@ class BootLoaderInstallGrub2(BootLoaderInstallBase):
'root device node name required for grub2 installation'
)
- self.mountpoint_root = mkdtemp()
- self.mountpoint_boot = mkdtemp()
- self.grub2_boot_device = custom_args['boot_device']
- self.grub2_root_device = custom_args['root_device']
-
- self.extra_boot_partition = False
- if not self.grub2_boot_device == self.grub2_root_device:
- self.extra_boot_partition = True
+ self.root_mount = MountManager(
+ custom_args['root_device']
+ )
+ self.boot_mount = MountManager(
+ custom_args['boot_device']
+ )
def install(self):
"""
@@ -60,15 +55,18 @@ class BootLoaderInstallGrub2(BootLoaderInstallBase):
"""
log.info('Installing grub2 on disk %s', self.device)
- if self.extra_boot_partition:
- self.__mount_boot_partition()
- self.__mount_root_partition()
- module_directory = self.mountpoint_root + '/usr/lib/grub2/i386-pc'
- boot_directory = self.mountpoint_boot
+ if not self.root_mount.device == self.boot_mount.device:
+ self.root_mount.mount()
+ self.boot_mount.mount()
+ module_directory = self.root_mount.mountpoint \
+ + '/usr/lib/grub2/i386-pc'
+ boot_directory = self.boot_mount.mountpoint
else:
- self.__mount_root_partition()
- module_directory = self.mountpoint_root + '/usr/lib/grub2/i386-pc'
- boot_directory = self.mountpoint_root + '/boot'
+ self.root_mount.mount()
+ module_directory = self.root_mount.mountpoint \
+ + '/usr/lib/grub2/i386-pc'
+ boot_directory = self.root_mount.mountpoint \
+ + '/boot'
Command.run(
[
@@ -82,47 +80,7 @@ class BootLoaderInstallGrub2(BootLoaderInstallBase):
]
)
- def __mount_boot_partition(self):
- self.__mount(self.grub2_boot_device, self.mountpoint_boot)
-
- def __mount_root_partition(self):
- self.__mount(self.grub2_root_device, self.mountpoint_root)
-
- def __is_mounted(self, mountpoint):
- try:
- Command.run(['mountpoint', mountpoint])
- return True
- except Exception:
- return False
-
- def __mount(self, device, mountpoint):
- Command.run(['mount', device, mountpoint])
- return True
-
- def __umount(self, mountpoint):
- if self.__is_mounted(mountpoint):
- umounted_successfully = False
- for busy in [1, 2, 3]:
- try:
- Command.run(['umount', mountpoint])
- umounted_successfully = True
- break
- except Exception:
- log.warning(
- '%d umount of %s failed, try again in 1sec',
- busy, mountpoint
- )
- time.sleep(1)
- if not umounted_successfully:
- log.warning(
- '%s still busy at %s', mountpoint, type(self).__name__
- )
- # skip removing the mountpoint directory
- return
-
- Path.remove(mountpoint)
-
def __del__(self):
log.info('Cleaning up %s instance', type(self).__name__)
- self.__umount(self.mountpoint_root)
- self.__umount(self.mountpoint_boot)
+ self.root_mount.umount()
+ self.boot_mount.umount()
diff --git a/kiwi/bootloader_install_zipl.py b/kiwi/bootloader_install_zipl.py
index 7a836758..54a09619 100644
--- a/kiwi/bootloader_install_zipl.py
+++ b/kiwi/bootloader_install_zipl.py
@@ -15,14 +15,11 @@
# You should have received a copy of the GNU General Public License
# along with kiwi. If not, see
#
-import time
-from tempfile import mkdtemp
-
# project
from .bootloader_install_base import BootLoaderInstallBase
from .command import Command
-from .path import Path
from .logger import log
+from .mount_manager import MountManager
from .exceptions import(
KiwiBootLoaderZiplInstallError
@@ -40,9 +37,9 @@ class BootLoaderInstallZipl(BootLoaderInstallBase):
'boot device node name required for zipl installation'
)
- self.mountpoint = mkdtemp()
- self.is_mounted = False
- self.zipl_boot_device = custom_args['boot_device']
+ self.boot_mount = MountManager(
+ custom_args['boot_device']
+ )
def install(self):
"""
@@ -50,12 +47,12 @@ class BootLoaderInstallZipl(BootLoaderInstallBase):
"""
log.info('Installing zipl on disk %s', self.device)
- self.__mount_boot_partition()
+ self.boot_mount.mount()
bash_command = ' '.join(
[
- 'cd', self.mountpoint, '&&',
- 'zipl', '-V', '-c', self.mountpoint + '/config',
+ 'cd', self.boot_mount.mountpoint, '&&',
+ 'zipl', '-V', '-c', self.boot_mount.mountpoint + '/config',
'-m', 'menu'
]
)
@@ -64,40 +61,6 @@ class BootLoaderInstallZipl(BootLoaderInstallBase):
)
log.debug('zipl install succeeds with: %s', zipl_call.output)
- self.__umount_boot_partition()
-
- def __mount_boot_partition(self):
- Command.run(
- ['mount', self.zipl_boot_device, self.mountpoint]
- )
- self.is_mounted = True
-
- def __umount_boot_partition(self):
- Command.run(
- ['umount', self.mountpoint]
- )
- Path.remove(self.mountpoint)
- self.is_mounted = False
-
def __del__(self):
- if self.is_mounted:
- log.info('Cleaning up %s instance', type(self).__name__)
- umounted_successfully = False
- for busy in [1, 2, 3]:
- try:
- Command.run(['umount', self.mountpoint])
- umounted_successfully = True
- break
- except Exception:
- log.warning(
- '%d umount of %s failed, try again in 1sec',
- busy, self.mountpoint
- )
- time.sleep(1)
- if not umounted_successfully:
- log.warning(
- '%s still busy at %s',
- self.mountpoint, type(self).__name__
- )
- else:
- Path.remove(self.mountpoint)
+ log.info('Cleaning up %s instance', type(self).__name__)
+ self.boot_mount.umount()
diff --git a/kiwi/filesystem_base.py b/kiwi/filesystem_base.py
index 55566eea..35bfd169 100644
--- a/kiwi/filesystem_base.py
+++ b/kiwi/filesystem_base.py
@@ -16,14 +16,12 @@
# along with kiwi. If not, see
#
import os
-import time
from tempfile import mkdtemp
# project
-from .command import Command
from .logger import log
-from .path import Path
from .data_sync import DataSync
+from .mount_manager import MountManager
from .exceptions import (
KiwiFileSystemSyncError
@@ -39,7 +37,7 @@ class FileSystemBase(object):
# here. The file name of the file containing the filesystem is
# stored in the device_provider if the filesystem is represented
# as a file there
- self.mountpoint = None
+ self.filesystem_mount = None
# bind the block device providing class instance to this object.
# This is done to guarantee the correct destructor order when
@@ -79,52 +77,18 @@ class FileSystemBase(object):
raise KiwiFileSystemSyncError(
'given root directory %s does not exist' % self.root_dir
)
- device = self.device_provider.get_device()
- Command.run(
- ['mount', device, self.__setup_mountpoint()]
+ self.filesystem_mount = MountManager(
+ device=self.device_provider.get_device(),
+ mountpoint=mkdtemp(prefix='kiwi_filesystem.')
)
- if self.mountpoint and self.is_mounted():
- data = DataSync(self.root_dir, self.mountpoint)
- data.sync_data(exclude)
- Command.run(
- ['umount', self.mountpoint]
- )
- Path.remove(self.mountpoint)
- self.mountpoint = None
-
- def is_mounted(self):
- if self.mountpoint:
- try:
- Command.run(['mountpoint', self.mountpoint])
- return True
- except Exception:
- pass
- return False
-
- def __setup_mountpoint(self):
- self.mountpoint = mkdtemp(prefix='kiwi_filesystem.')
- return self.mountpoint
+ self.filesystem_mount.mount()
+ data = DataSync(
+ self.root_dir, self.filesystem_mount.mountpoint
+ )
+ data.sync_data(exclude)
+ self.filesystem_mount.umount()
def __del__(self):
- if self.mountpoint:
+ if self.filesystem_mount:
log.info('Cleaning up %s instance', type(self).__name__)
- if self.is_mounted():
- umounted_successfully = False
- for busy in [1, 2, 3]:
- try:
- Command.run(['umount', self.mountpoint])
- umounted_successfully = True
- break
- except Exception:
- log.warning(
- '%d umount of %s failed, try again in 1sec',
- busy, self.mountpoint
- )
- time.sleep(1)
- if not umounted_successfully:
- log.warning(
- '%s still busy at %s',
- self.mountpoint, type(self).__name__
- )
- else:
- Path.remove(self.mountpoint)
+ self.filesystem_mount.umount()
diff --git a/kiwi/mount_manager.py b/kiwi/mount_manager.py
new file mode 100644
index 00000000..a1f34256
--- /dev/null
+++ b/kiwi/mount_manager.py
@@ -0,0 +1,97 @@
+# Copyright (c) 2015 SUSE Linux GmbH. All rights reserved.
+#
+# This file is part of kiwi.
+#
+# kiwi is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# kiwi is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with kiwi. If not, see
+#
+import time
+from tempfile import mkdtemp
+
+# project
+from .command import Command
+from .path import Path
+from .logger import log
+
+
+class MountManager(object):
+ """
+ Provide 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
+ """
+ def __init__(self, device, mountpoint=None):
+ self.device = device
+ if not mountpoint:
+ self.mountpoint = mkdtemp()
+ else:
+ self.mountpoint = mountpoint
+
+ def bind_mount(self):
+ if not self.is_mounted():
+ Command.run(
+ ['mount', '-n', '--bind', self.device, self.mountpoint]
+ )
+
+ def mount(self, options=None):
+ if not self.is_mounted():
+ option_list = []
+ if options:
+ option_list = ['-o'] + options
+ Command.run(
+ ['mount'] + option_list + [self.device, self.mountpoint]
+ )
+
+ def umount_lazy(self, delete_mountpoint=True):
+ if self.is_mounted():
+ Command.run(['umount', '-l', self.mountpoint])
+ if delete_mountpoint:
+ Path.wipe(self.mountpoint)
+
+ def umount(self, delete_mountpoint=True):
+ if self.is_mounted():
+ umounted_successfully = False
+ for busy in [1, 2, 3]:
+ try:
+ Command.run(['umount', self.mountpoint])
+ umounted_successfully = True
+ break
+ except Exception:
+ log.warning(
+ '%d umount of %s failed, try again in 1sec',
+ busy, self.mountpoint
+ )
+ time.sleep(1)
+ if not umounted_successfully:
+ log.warning(
+ '%s still busy at %s', self.mountpoint, type(self).__name__
+ )
+ # skip removing the mountpoint directory
+ return False
+
+ if delete_mountpoint:
+ Path.wipe(self.mountpoint)
+
+ return True
+
+ def is_mounted(self):
+ mountpoint_call = Command.run(
+ command=['mountpoint', self.mountpoint],
+ raise_on_error=False
+ )
+ if mountpoint_call.returncode == 0:
+ return True
+ else:
+ return False
diff --git a/kiwi/root_bind.py b/kiwi/root_bind.py
index e49fbbb5..d928b25b 100644
--- a/kiwi/root_bind.py
+++ b/kiwi/root_bind.py
@@ -21,6 +21,7 @@ import os
from .command import Command
from .logger import log
from .path import Path
+from .mount_manager import MountManager
from .exceptions import (
KiwiMountKernelFileSystemsError,
@@ -60,13 +61,11 @@ class RootBind(object):
try:
for location in self.bind_locations:
if os.path.exists(location):
- Command.run(
- [
- 'mount', '-n', '--bind', location,
- self.root_dir + location
- ]
+ shared_mount = MountManager(
+ device=location, mountpoint=self.root_dir + location
)
- self.mount_stack.append(location)
+ shared_mount.bind_mount()
+ self.mount_stack.append(shared_mount)
except Exception as e:
self.cleanup()
raise KiwiMountKernelFileSystemsError(
@@ -78,12 +77,11 @@ class RootBind(object):
host_dir = self.shared_location
try:
Path.create(self.root_dir + host_dir)
- Command.run(
- [
- 'mount', '-n', '--bind', host_dir, self.root_dir + host_dir
- ]
+ shared_mount = MountManager(
+ device=host_dir, mountpoint=self.root_dir + host_dir
)
- self.mount_stack.append(host_dir)
+ shared_mount.bind_mount()
+ self.mount_stack.append(shared_mount)
self.dir_stack.append(host_dir)
except Exception as e:
self.cleanup()
@@ -146,15 +144,17 @@ class RootBind(object):
)
def __cleanup_mount_stack(self):
- mount_list = self.__build_mount_list()
- if mount_list:
- try:
- Command.run(['umount', '-l'] + mount_list)
- except Exception as e:
- log.warning(
- 'Image root directory %s not cleanly umounted: %s',
- self.root_dir, format(e)
- )
+ for mount in reversed(self.mount_stack):
+ if mount.is_mounted():
+ try:
+ mount.umount_lazy(delete_mountpoint=False)
+ except Exception as e:
+ log.warning(
+ 'Image root directory %s not cleanly umounted: %s',
+ self.root_dir, format(e)
+ )
+ else:
+ log.warning('Path %s not a mountpoint', mount.mountpoint)
del self.mount_stack[:]
@@ -167,14 +167,3 @@ class RootBind(object):
'Failed to remove directory %s: %s', location, format(e)
)
del self.dir_stack[:]
-
- def __build_mount_list(self):
- mount_points = []
- for location in reversed(self.mount_stack):
- mount_path = self.root_dir + location
- try:
- Command.run(['mountpoint', '-q', mount_path])
- mount_points.append(mount_path)
- except Exception:
- log.warning('Path %s not a mountpoint', mount_path)
- return mount_points
diff --git a/kiwi/uri.py b/kiwi/uri.py
index a5f96cfb..e2372009 100644
--- a/kiwi/uri.py
+++ b/kiwi/uri.py
@@ -21,8 +21,7 @@ from urllib.parse import urlparse
import hashlib
# project
-from .command import Command
-from .path import Path
+from .mount_manager import MountManager
from .exceptions import (
KiwiUriStyleUnknown,
@@ -107,10 +106,12 @@ class Uri(object):
)
def __iso_mount_path(self, path):
- iso_path = mkdtemp(prefix='iso-mount.')
- Command.run(['mount', path, iso_path])
- self.mount_stack.append(iso_path)
- return iso_path
+ iso_mount = MountManager(
+ device=path, mountpoint=mkdtemp(prefix='iso-mount.')
+ )
+ iso_mount.mount()
+ self.mount_stack.append(iso_mount)
+ return iso_mount.mountpoint
def __local_directory(self, path):
return os.path.normpath(path)
@@ -138,9 +139,5 @@ class Uri(object):
)
def __del__(self):
- try:
- for mount in reversed(self.mount_stack):
- Command.run(['umount', mount])
- Path.remove(mount)
- except Exception:
- pass
+ for mount in reversed(self.mount_stack):
+ mount.umount()
diff --git a/kiwi/volume_manager_base.py b/kiwi/volume_manager_base.py
index ffcde878..377e7f3c 100644
--- a/kiwi/volume_manager_base.py
+++ b/kiwi/volume_manager_base.py
@@ -20,8 +20,8 @@ from tempfile import mkdtemp
import os
# project
-from .command import Command
from .device_provider import DeviceProvider
+from .mount_manager import MountManager
from .data_sync import DataSync
from .path import Path
from .system_size import SystemSize
@@ -174,25 +174,15 @@ class VolumeManagerBase(DeviceProvider):
)
return mbsize
- def is_mounted(self):
- """
- Implements check if volumes are mounted
- """
- if self.mountpoint:
- try:
- Command.run(['mountpoint', self.mountpoint])
- return True
- except Exception:
- pass
- return False
-
def sync_data(self, exclude=None):
"""
Implements sync of root directory to mounted volumes
"""
- if self.mountpoint and self.is_mounted():
- data = DataSync(self.root_dir, self.mountpoint)
- data.sync_data(exclude)
+ if self.mountpoint:
+ root_mount = MountManager(device=None, mountpoint=self.mountpoint)
+ if root_mount.is_mounted():
+ data = DataSync(self.root_dir, self.mountpoint)
+ data.sync_data(exclude)
def setup_mountpoint(self):
"""
diff --git a/kiwi/volume_manager_btrfs.py b/kiwi/volume_manager_btrfs.py
index b27dba32..27e26501 100644
--- a/kiwi/volume_manager_btrfs.py
+++ b/kiwi/volume_manager_btrfs.py
@@ -15,13 +15,13 @@
# You should have received a copy of the GNU General Public License
# along with kiwi. If not, see
#
-import time
import re
import os
# project
from .command import Command
from .volume_manager_base import VolumeManagerBase
+from .mount_manager import MountManager
from .mapped_device import MappedDevice
from .filesystem import FileSystem
from .data_sync import DataSync
@@ -48,6 +48,9 @@ class VolumeManagerBtrfs(VolumeManagerBase):
self.custom_args['root_is_snapshot'] = False
self.subvol_mount_list = []
+ self.toplevel_mount = None
+
+ self.setup_mountpoint()
def get_device(self):
return {'root': self}
@@ -59,10 +62,10 @@ class VolumeManagerBtrfs(VolumeManagerBase):
filesystem.create_on_device(
label=self.custom_args['root_label']
)
- self.setup_mountpoint()
- Command.run(
- ['mount', self.device, self.mountpoint]
+ self.toplevel_mount = MountManager(
+ device=self.device, mountpoint=self.mountpoint
)
+ self.toplevel_mount.mount()
root_volume = self.mountpoint + '/@'
Command.run(
['btrfs', 'subvolume', 'create', root_volume]
@@ -114,29 +117,27 @@ class VolumeManagerBtrfs(VolumeManagerBase):
os.path.normpath(toplevel + volume.realpath)
]
)
- self.subvol_mount_list.append(
- volume.realpath
- )
+ if self.custom_args['root_is_snapshot']:
+ snapshot = self.mountpoint + '/@/.snapshots/1/snapshot/'
+ volume_mount = MountManager(
+ device=self.device,
+ mountpoint=os.path.normpath(snapshot + volume.realpath)
+ )
+ self.subvol_mount_list.append(
+ volume_mount
+ )
def mount_volumes(self):
- if self.custom_args['root_is_snapshot']:
- snapshot = self.mountpoint + '/@/.snapshots/1/snapshot/'
- for subvol in self.subvol_mount_list:
- volume_parent_path = os.path.normpath(
- snapshot + subvol
- )
- if not os.path.exists(volume_parent_path):
- Path.create(volume_parent_path)
- Command.run(
- [
- 'mount', self.device,
- os.path.normpath(snapshot + subvol),
- '-o', 'subvol=' + os.path.normpath('@/' + subvol)
- ]
- )
+ for volume_mount in self.subvol_mount_list:
+ if not os.path.exists(volume_mount.mountpoint):
+ Path.create(volume_mount.mountpoint)
+ subvol_name = os.path.basename(volume_mount.mountpoint)
+ volume_mount.mount(
+ options=['subvol=' + os.path.normpath('@/' + subvol_name)]
+ )
def sync_data(self, exclude=None):
- if self.mountpoint and self.is_mounted():
+ if self.toplevel_mount:
sync_target = self.mountpoint + '/@'
if self.custom_args['root_is_snapshot']:
sync_target = self.mountpoint + '/@/.snapshots/1/snapshot'
@@ -165,33 +166,10 @@ class VolumeManagerBtrfs(VolumeManagerBase):
'Failed to find btrfs volume: %s' % default_volume
)
- def __try_umount(self, mount_point, wipe=True):
- umounted_successfully = False
- for busy in [1, 2, 3]:
- try:
- Command.run(['umount', os.path.normpath(mount_point)])
- umounted_successfully = True
- break
- except Exception:
- log.warning(
- '%d umount of %s failed, try again in 1sec',
- busy, mount_point
- )
- time.sleep(1)
- if umounted_successfully and wipe:
- Path.wipe(self.mountpoint)
- elif not umounted_successfully:
- log.warning(
- 'mount path %s still busy', mount_point
- )
-
def __del__(self):
- if self.is_mounted():
+ if self.toplevel_mount:
log.info('Cleaning up %s instance', type(self).__name__)
- if self.custom_args['root_is_snapshot']:
- for subvol in reversed(self.subvol_mount_list):
- subvol_mount = \
- self.mountpoint + '/@/.snapshots/1/snapshot/' + subvol
- self.__try_umount(mount_point=subvol_mount, wipe=False)
+ for volume_mount in reversed(self.subvol_mount_list):
+ volume_mount.umount(delete_mountpoint=False)
- self.__try_umount(mount_point=self.device)
+ self.toplevel_mount.umount()
diff --git a/kiwi/volume_manager_lvm.py b/kiwi/volume_manager_lvm.py
index 2bb29027..1a1ff46c 100644
--- a/kiwi/volume_manager_lvm.py
+++ b/kiwi/volume_manager_lvm.py
@@ -15,12 +15,10 @@
# You should have received a copy of the GNU General Public License
# along with kiwi. If not, see
#
-from collections import namedtuple
-import time
-
# project
from .command import Command
from .volume_manager_base import VolumeManagerBase
+from .mount_manager import MountManager
from .mapped_device import MappedDevice
from .filesystem import FileSystem
from .path import Path
@@ -43,6 +41,8 @@ class VolumeManagerLVM(VolumeManagerBase):
if 'root_label' not in self.custom_args:
self.custom_args['root_label'] = 'ROOT'
+ self.setup_mountpoint()
+
def get_device(self):
"""
return names of volume devices, note that the mapping
@@ -124,12 +124,9 @@ class VolumeManagerLVM(VolumeManagerBase):
)
def mount_volumes(self):
- self.setup_mountpoint()
- for mount in self.mount_list:
- Path.create(self.mountpoint + mount.mountpoint)
- Command.run(
- ['mount', mount.device, self.mountpoint + mount.mountpoint]
- )
+ for volume_mount in self.mount_list:
+ Path.create(volume_mount.mountpoint)
+ volume_mount.mount()
def __create_filesystem(self, volume_name, filesystem_name):
device_node = self.volume_map[volume_name]
@@ -145,25 +142,20 @@ class VolumeManagerLVM(VolumeManagerBase):
)
def __add_to_mount_list(self, volume_name, realpath):
- mount_type = namedtuple(
- 'mount_type', [
- 'device', 'mountpoint'
- ]
- )
device_node = self.volume_map[volume_name]
if volume_name == 'LVRoot':
# root volume must be first in the list
self.mount_list.insert(
- 0, mount_type(
+ 0, MountManager(
device=device_node,
- mountpoint='/'
+ mountpoint=self.mountpoint
)
)
else:
self.mount_list.append(
- mount_type(
+ MountManager(
device=device_node,
- mountpoint='/' + realpath
+ mountpoint=self.mountpoint + '/' + realpath
)
)
@@ -184,33 +176,17 @@ class VolumeManagerLVM(VolumeManagerBase):
def __del__(self):
if self.volume_group:
log.info('Cleaning up %s instance', type(self).__name__)
- if self.is_mounted():
- all_volumes_umounted = True
- for mount in reversed(self.mount_list):
- umounted_successfully = False
- for busy in [1, 2, 3]:
- try:
- Command.run(['umount', mount.device])
- umounted_successfully = True
- break
- except Exception:
- log.warning(
- '%d umount of %s failed, try again in 1sec',
- busy, mount.device
- )
- time.sleep(1)
- if not umounted_successfully:
+ all_volumes_umounted = True
+ for volume_mount in reversed(self.mount_list):
+ if volume_mount.is_mounted():
+ if not volume_mount.umount():
all_volumes_umounted = False
- log.warning(
- '%s still busy at %s',
- self.mountpoint + mount.mountpoint,
- type(self).__name__
- )
- if all_volumes_umounted:
- Path.wipe(self.mountpoint)
- try:
- Command.run(['vgchange', '-an', self.volume_group])
- except Exception:
- log.warning(
- 'volume group %s still busy', self.volume_group
- )
+
+ if all_volumes_umounted:
+ Path.wipe(self.mountpoint)
+ try:
+ Command.run(['vgchange', '-an', self.volume_group])
+ except Exception:
+ log.warning(
+ 'volume group %s still busy', self.volume_group
+ )
diff --git a/test/unit/bootloader_install_grub2_test.py b/test/unit/bootloader_install_grub2_test.py
index 95d0ccf3..f976ada4 100644
--- a/test/unit/bootloader_install_grub2_test.py
+++ b/test/unit/bootloader_install_grub2_test.py
@@ -12,29 +12,35 @@ from kiwi.defaults import Defaults
class TestBootLoaderInstallGrub2(object):
- @patch('kiwi.bootloader_install_grub2.mkdtemp')
- def setup(self, mock_mkdtemp):
- tmpdirs = ['tmp_boot', 'tmp_root']
+ @patch('kiwi.bootloader_install_grub2.MountManager')
+ def setup(self, mock_mount):
+ custom_args = {
+ 'boot_device': '/dev/mapper/loop0p2',
+ 'root_device': '/dev/mapper/loop0p1'
+ }
+ root_mount = mock.Mock()
+ root_mount.device = custom_args['root_device']
+ root_mount.mountpoint = 'tmp_root'
- def side_effect():
- return tmpdirs.pop()
+ boot_mount = mock.Mock()
+ boot_mount.device = custom_args['boot_device']
+ boot_mount.mountpoint = 'tmp_boot'
+
+ mount_managers = [boot_mount, root_mount]
+
+ def side_effect(arg):
+ return mount_managers.pop()
+
+ mock_mount.side_effect = side_effect
- mock_mkdtemp.side_effect = side_effect
device_provider = mock.Mock()
device_provider.get_device = mock.Mock(
return_value='/dev/some-device'
)
- self.path = mock.Mock()
- kiwi.bootloader_install_grub2.Path = self.path
+
self.bootloader = BootLoaderInstallGrub2(
- 'root_dir', device_provider, {
- 'boot_device': '/dev/mapper/loop0p2',
- 'root_device': '/dev/mapper/loop0p1'
- }
+ 'root_dir', device_provider, custom_args
)
- assert self.bootloader.mountpoint_root == 'tmp_root'
- assert self.bootloader.mountpoint_boot == 'tmp_boot'
- assert self.bootloader.extra_boot_partition is True
@raises(KiwiBootLoaderGrubInstallError)
def test_post_init_no_boot_device(self):
@@ -47,62 +53,36 @@ class TestBootLoaderInstallGrub2(object):
@patch('kiwi.bootloader_install_grub2.Command.run')
def test_install_with_extra_boot_partition(self, mock_command):
self.bootloader.install()
- assert mock_command.call_args_list == [
- call(['mount', '/dev/mapper/loop0p2', 'tmp_boot']),
- call(['mount', '/dev/mapper/loop0p1', 'tmp_root']),
- call([
+ self.bootloader.root_mount.mount.assert_called_once_with()
+ self.bootloader.boot_mount.mount.assert_called_once_with()
+ mock_command.assert_called_once_with(
+ [
'grub2-install', '--skip-fs-probe',
'--directory', 'tmp_root/usr/lib/grub2/i386-pc',
'--boot-directory', 'tmp_boot',
'--target', 'i386-pc',
'--modules', ' '.join(Defaults.get_grub_bios_modules()),
'/dev/some-device'
- ])
- ]
+ ]
+ )
@patch('kiwi.bootloader_install_grub2.Command.run')
def test_install(self, mock_command):
- self.bootloader.extra_boot_partition = False
+ self.bootloader.boot_mount.device = self.bootloader.root_mount.device
self.bootloader.install()
- assert mock_command.call_args_list == [
- call(['mount', '/dev/mapper/loop0p1', 'tmp_root']),
- call([
+ self.bootloader.root_mount.mount.assert_called_once_with()
+ mock_command.assert_called_once_with(
+ [
'grub2-install', '--skip-fs-probe',
'--directory', 'tmp_root/usr/lib/grub2/i386-pc',
'--boot-directory', 'tmp_root/boot',
'--target', 'i386-pc',
'--modules', ' '.join(Defaults.get_grub_bios_modules()),
'/dev/some-device'
- ])
- ]
+ ]
+ )
- @patch('kiwi.bootloader_install_grub2.Command.run')
- @patch('kiwi.logger.log.warning')
- @patch('time.sleep')
- def test_desstructor(self, mock_sleep, mock_warn, mock_command):
- command_return_values = [False, False, False, True, True, True]
-
- def side_effect(arg):
- if not command_return_values.pop():
- raise Exception
-
- mock_command.side_effect = side_effect
+ def test_destructor(self):
self.bootloader.__del__()
- self.path.remove.assert_called_once_with('tmp_root')
- assert mock_command.call_args_list == [
- call(['mountpoint', 'tmp_root']),
- call(['umount', 'tmp_root']),
- call(['mountpoint', 'tmp_boot']),
- call(['umount', 'tmp_boot']),
- call(['umount', 'tmp_boot']),
- call(['umount', 'tmp_boot'])
- ]
-
- @patch('kiwi.bootloader_install_grub2.Command.run')
- def test_desstructor_nothing_mounted(self, mock_command):
- mock_command.side_effect = Exception
- self.bootloader.__del__()
- assert mock_command.call_args_list == [
- call(['mountpoint', 'tmp_root']),
- call(['mountpoint', 'tmp_boot'])
- ]
+ self.bootloader.root_mount.umount.assert_called_once_with()
+ self.bootloader.boot_mount.umount.assert_called_once_with()
diff --git a/test/unit/bootloader_install_zipl_test.py b/test/unit/bootloader_install_zipl_test.py
index 2bf8b833..056e6e14 100644
--- a/test/unit/bootloader_install_zipl_test.py
+++ b/test/unit/bootloader_install_zipl_test.py
@@ -11,15 +11,23 @@ from kiwi.bootloader_install_zipl import BootLoaderInstallZipl
class TestBootLoaderInstallZipl(object):
- @patch('kiwi.bootloader_install_zipl.mkdtemp')
- def setup(self, mock_mkdtemp):
- mock_mkdtemp.return_value = 'tmpdir'
+ @patch('kiwi.bootloader_install_zipl.MountManager')
+ def setup(self, mock_mount):
+ custom_args = {
+ 'boot_device': '/dev/mapper/loop0p1'
+ }
+ boot_mount = mock.Mock()
+ boot_mount.device = custom_args['boot_device']
+ boot_mount.mountpoint = 'tmp_boot'
+
+ mock_mount.return_value = boot_mount
+
device_provider = mock.Mock()
device_provider.get_device = mock.Mock(
return_value='/dev/some-device'
)
self.bootloader = BootLoaderInstallZipl(
- 'root_dir', device_provider, {'boot_device': '/dev/mapper/loop0p1'}
+ 'root_dir', device_provider, custom_args
)
@raises(KiwiBootLoaderZiplInstallError)
@@ -27,13 +35,10 @@ class TestBootLoaderInstallZipl(object):
self.bootloader.post_init(None)
@patch('kiwi.bootloader_install_zipl.Command.run')
- @patch('kiwi.bootloader_install_zipl.Path.remove')
- def test_install(self, mock_path, mock_command):
+ def test_install(self, mock_command):
self.bootloader.install()
- assert mock_command.call_args_list == [
- call(
- ['mount', '/dev/mapper/loop0p1', 'tmpdir']
- ),
+ self.bootloader.boot_mount.mount.assert_called_once_with()
+ mock_command.call_args_list == [
call(
['bash', '-c', 'cd tmpdir && zipl -V -c tmpdir/config -m menu']
),
@@ -41,27 +46,7 @@ class TestBootLoaderInstallZipl(object):
['umount', 'tmpdir']
)
]
- mock_path.assert_called_once_with('tmpdir')
- @patch('kiwi.bootloader_install_zipl.Command.run')
- def test_destructor_valid_mountpoint(self, mock_command):
- self.bootloader.is_mounted = True
+ def test_destructor(self):
self.bootloader.__del__()
- self.bootloader.is_mounted = False
- mock_command.call_args_list == [
- call(['umount', 'tmpdir']),
- call(['rmdir', 'tmpdir'])
- ]
-
- @patch('kiwi.bootloader_install_zipl.Command.run')
- @patch('kiwi.logger.log.warning')
- @patch('kiwi.bootloader_install_zipl.Path.remove')
- @patch('time.sleep')
- def test_destructor_mountpoint_busy(
- self, mock_sleep, mock_path, mock_warn, mock_command
- ):
- self.bootloader.is_mounted = True
- mock_command.side_effect = Exception
- self.bootloader.__del__()
- self.bootloader.is_mounted = False
- assert mock_warn.called
+ self.bootloader.boot_mount.umount.assert_called_once_with()
diff --git a/test/unit/filesystem_base_test.py b/test/unit/filesystem_base_test.py
index a429f6bc..e68153f1 100644
--- a/test/unit/filesystem_base_test.py
+++ b/test/unit/filesystem_base_test.py
@@ -35,75 +35,32 @@ class TestFileSystemBase(object):
def test_create_on_file(self):
self.fsbase.create_on_file('myimage')
- @patch('kiwi.filesystem_base.Command.run')
- def test_is_mounted_true(self, mock_command):
- self.fsbase.mountpoint = 'tmpdir'
- assert self.fsbase.is_mounted()
- mock_command.assert_called_once_with(['mountpoint', 'tmpdir'])
- self.fsbase.mountpoint = None
-
- @patch('kiwi.filesystem_base.Command.run')
- def test_is_mounted_false(self, mock_command):
- mock_command.side_effect = Exception
- self.fsbase.mountpoint = 'tmpdir'
- assert self.fsbase.is_mounted() is False
- self.fsbase.mountpoint = None
-
- @patch('kiwi.filesystem_base.Command.run')
+ @patch('kiwi.filesystem_base.MountManager')
@patch('kiwi.filesystem_base.DataSync')
- @patch('kiwi.filesystem_base.FileSystemBase.is_mounted')
@patch('kiwi.filesystem_base.mkdtemp')
@patch('os.path.exists')
- def test_sync_data(
- self, mock_exists, mock_mkdtemp, mock_mounted, mock_sync, mock_command
- ):
+ def test_sync_data(self, mock_exists, mock_mkdtemp, mock_sync, mock_mount):
+ mock_mkdtemp.return_value = 'tmpdir'
+ mock_exists.return_value = True
+
+ filesystem_mount = mock.Mock()
+ filesystem_mount.mountpoint = mock_mkdtemp.return_value
+ mock_mount.return_value = filesystem_mount
+
data_sync = mock.Mock()
mock_sync.return_value = data_sync
- mock_exists.return_value = True
- mock_mkdtemp.return_value = 'tmpdir'
+
self.fsbase.sync_data(['exclude_me'])
+
mock_sync.assert_called_once_with('root_dir', 'tmpdir')
data_sync.sync_data.assert_called_once_with(['exclude_me'])
- call = mock_command.call_args_list[0]
- assert mock_command.call_args_list[0] == \
- call([
- 'mount', '/dev/loop0', 'tmpdir'
- ])
- call = mock_command.call_args_list[1]
- assert mock_command.call_args_list[1] == \
- call([
- 'umount', 'tmpdir'
- ])
- call = mock_command.call_args_list[2]
- assert mock_command.call_args_list[2] == \
- call([
- 'rmdir', 'tmpdir'
- ])
+ mock_mount.assert_called_once_with(
+ device='/dev/loop0', mountpoint='tmpdir'
+ )
+ filesystem_mount.mount.assert_called_once_with()
+ filesystem_mount.umount.assert_called_once_with()
- @patch('kiwi.filesystem_base.Command.run')
- @patch('kiwi.filesystem_base.FileSystemBase.is_mounted')
- def test_destructor_valid_mountpoint(self, mock_mounted, mock_command):
- mock_mounted.return_value = True
- self.fsbase.mountpoint = 'tmpdir'
+ def test_destructor_valid_mountpoint(self):
+ self.fsbase.filesystem_mount = mock.Mock()
self.fsbase.__del__()
- self.fsbase.mountpoint = None
- call = mock_command.call_args_list[0]
- assert mock_command.call_args_list[0] == \
- call(['umount', 'tmpdir'])
- call = mock_command.call_args_list[1]
- assert mock_command.call_args_list[1] == \
- call(['rmdir', 'tmpdir'])
-
- @patch('kiwi.filesystem_base.Command.run')
- @patch('kiwi.filesystem_base.FileSystemBase.is_mounted')
- @patch('kiwi.logger.log.warning')
- @patch('time.sleep')
- def test_destructor_mountpoint_busy(
- self, mock_sleep, mock_warn, mock_mounted, mock_command
- ):
- mock_command.side_effect = Exception
- mock_mounted.return_value = True
- self.fsbase.mountpoint = 'tmpdir'
- self.fsbase.__del__()
- self.fsbase.mountpoint = None
- assert mock_warn.called
+ self.fsbase.filesystem_mount.umount.assert_called_once_with()
diff --git a/test/unit/mount_manager_test.py b/test/unit/mount_manager_test.py
new file mode 100644
index 00000000..2903d61c
--- /dev/null
+++ b/test/unit/mount_manager_test.py
@@ -0,0 +1,97 @@
+from nose.tools import *
+from mock import patch
+from mock import call
+import mock
+
+from . import nose_helper
+
+from kiwi.exceptions import *
+from kiwi.mount_manager import MountManager
+
+
+class TestMountManager(object):
+ def setup(self):
+ self.mount_manager = MountManager(
+ '/dev/some-device', '/some/mountpoint'
+ )
+
+ @patch('kiwi.mount_manager.mkdtemp')
+ def test_setup_empty_mountpoint(self, mock_mkdtemp):
+ mock_mkdtemp.return_value = 'tmpdir'
+ mount_manager = MountManager('/dev/some-device')
+ assert mount_manager.mountpoint == 'tmpdir'
+
+ @patch('kiwi.mount_manager.Command.run')
+ @patch('kiwi.mount_manager.MountManager.is_mounted')
+ def test_bind_mount(self, mock_mounted, mock_command):
+ mock_mounted.return_value = False
+ self.mount_manager.bind_mount()
+ mock_command.assert_called_once_with(
+ ['mount', '-n', '--bind', '/dev/some-device', '/some/mountpoint']
+ )
+
+ @patch('kiwi.mount_manager.Command.run')
+ @patch('kiwi.mount_manager.MountManager.is_mounted')
+ def test_mount(self, mock_mounted, mock_command):
+ mock_mounted.return_value = False
+ self.mount_manager.mount(['options'])
+ mock_command.assert_called_once_with(
+ ['mount', '-o', 'options', '/dev/some-device', '/some/mountpoint']
+ )
+
+ @patch('kiwi.mount_manager.Command.run')
+ @patch('kiwi.mount_manager.Path.wipe')
+ @patch('kiwi.mount_manager.MountManager.is_mounted')
+ def test_umount_lazy(self, mock_mounted, mock_path, mock_command):
+ mock_mounted.return_value = True
+ self.mount_manager.umount_lazy()
+ mock_command.assert_called_once_with(
+ ['umount', '-l', '/some/mountpoint']
+ )
+ mock_path.assert_called_once_with('/some/mountpoint')
+
+ @patch('kiwi.mount_manager.Command.run')
+ @patch('kiwi.mount_manager.MountManager.is_mounted')
+ @patch('time.sleep')
+ @patch('kiwi.logger.log.warning')
+ def test_umount_with_errors(
+ self, mock_warn, mock_sleep, mock_mounted, mock_command
+ ):
+ mock_command.side_effect = Exception
+ mock_mounted.return_value = True
+ assert self.mount_manager.umount() is False
+ assert mock_command.call_args_list == [
+ call(['umount', '/some/mountpoint']),
+ call(['umount', '/some/mountpoint']),
+ call(['umount', '/some/mountpoint'])
+ ]
+ assert mock_warn.called
+
+ @patch('kiwi.mount_manager.Command.run')
+ @patch('kiwi.mount_manager.Path.wipe')
+ @patch('kiwi.mount_manager.MountManager.is_mounted')
+ def test_umount_success(
+ self, mock_mounted, mock_path, mock_command
+ ):
+ mock_mounted.return_value = True
+ assert self.mount_manager.umount() is True
+ mock_command.assert_called_once_with(
+ ['umount', '/some/mountpoint']
+ )
+ mock_path.assert_called_once_with(
+ '/some/mountpoint'
+ )
+
+ @patch('kiwi.mount_manager.Command.run')
+ def test_is_mounted_true(self, mock_command):
+ command = mock.Mock()
+ command.returncode = 0
+ mock_command.return_value = command
+ assert self.mount_manager.is_mounted() is True
+
+ @patch('kiwi.mount_manager.Command.run')
+ def test_is_mounted_false(self, mock_command):
+ command = mock.Mock()
+ command.returncode = 1
+ mock_command.return_value = command
+ assert self.mount_manager.is_mounted() is False
diff --git a/test/unit/root_bind_test.py b/test/unit/root_bind_test.py
index bdf4d793..3151e99f 100644
--- a/test/unit/root_bind_test.py
+++ b/test/unit/root_bind_test.py
@@ -27,29 +27,33 @@ class TestRootBind(object):
self.bind_root.bind_locations = ['/proc']
# stub files/dirs and mountpoints to cleanup
+ self.mount_manager = mock.Mock()
self.bind_root.cleanup_files = ['/foo.kiwi']
- self.bind_root.mount_stack = ['/mountpoint']
+ self.bind_root.mount_stack = [self.mount_manager]
self.bind_root.dir_stack = ['/mountpoint']
@raises(KiwiMountKernelFileSystemsError)
- @patch('kiwi.command.Command.run')
+ @patch('kiwi.root_bind.MountManager.bind_mount')
@patch('kiwi.root_bind.RootBind.cleanup')
@patch('os.path.exists')
def test_kernel_file_systems_raises_error(
- self, mock_exists, mock_cleanup, mock_command
+ self, mock_exists, mock_cleanup, mock_mount
):
mock_exists.return_value = True
- mock_command.side_effect = KiwiMountKernelFileSystemsError(
+ mock_mount.side_effect = KiwiMountKernelFileSystemsError(
'mount-error'
)
self.bind_root.mount_kernel_file_systems()
mock.cleanup.assert_called_once_with()
@raises(KiwiMountSharedDirectoryError)
- @patch('kiwi.command.Command.run')
+ @patch('kiwi.root_bind.MountManager.bind_mount')
+ @patch('kiwi.root_bind.Path.create')
@patch('kiwi.root_bind.RootBind.cleanup')
- def test_shared_directory_raises_error(self, mock_cleanup, mock_command):
- mock_command.side_effect = KiwiMountSharedDirectoryError(
+ def test_shared_directory_raises_error(
+ self, mock_cleanup, mock_path, mock_mount
+ ):
+ mock_mount.side_effect = KiwiMountSharedDirectoryError(
'mount-error'
)
self.bind_root.mount_shared_directory()
@@ -69,25 +73,27 @@ class TestRootBind(object):
self.bind_root.setup_intermediate_config()
mock.cleanup.assert_called_once_with()
- @patch('kiwi.command.Command.run')
- def test_mount_kernel_file_systems(self, mock_command):
+ @patch('kiwi.root_bind.MountManager')
+ def test_mount_kernel_file_systems(self, mock_mount):
+ shared_mount = mock.Mock()
+ mock_mount.return_value = shared_mount
self.bind_root.mount_kernel_file_systems()
- mock_command.assert_called_once_with(
- ['mount', '-n', '--bind', '/proc', 'root-dir/proc']
+ mock_mount.assert_called_once_with(
+ device='/proc', mountpoint='root-dir/proc'
)
+ shared_mount.bind_mount.assert_called_once_with()
- @patch('kiwi.command.Command.run')
- def test_mount_shared_directory(self, mock_command):
+ @patch('kiwi.root_bind.MountManager')
+ @patch('kiwi.root_bind.Path.create')
+ def test_mount_shared_directory(self, mock_path, mock_mount):
+ shared_mount = mock.Mock()
+ mock_mount.return_value = shared_mount
self.bind_root.mount_shared_directory()
- assert mock_command.call_args_list == [
- call([
- 'mkdir', '-p', 'root-dir/var/cache/kiwi'
- ]),
- call([
- 'mount', '-n', '--bind', '/var/cache/kiwi',
- 'root-dir/var/cache/kiwi'
- ])
- ]
+ mock_path.assert_called_once_with('root-dir/var/cache/kiwi')
+ mock_mount.assert_called_once_with(
+ device='/var/cache/kiwi', mountpoint='root-dir/var/cache/kiwi'
+ )
+ shared_mount.bind_mount.assert_called_once_with()
@patch('kiwi.command.Command.run')
@patch('os.path.exists')
@@ -103,68 +109,58 @@ class TestRootBind(object):
])
]
- @patch('kiwi.command.Command.run')
+ @patch('kiwi.root_bind.Command.run')
+ @patch('kiwi.root_bind.Path.remove_hierarchy')
@patch('os.path.islink')
- def test_cleanup(self, mock_islink, mock_command):
+ def test_cleanup(self, mock_islink, mock_remove_hierarchy, mock_command):
mock_islink.return_value = True
self.bind_root.cleanup()
+ self.mount_manager.umount_lazy.assert_called_once_with(
+ delete_mountpoint=False
+ )
+ mock_remove_hierarchy.assert_called_once_with('root-dir/mountpoint')
+ mock_command.assert_called_once_with(
+ ['rm', '-f', 'root-dir/foo.kiwi', 'root-dir/foo']
+ )
- assert mock_command.call_args_list == [
- call([
- 'mountpoint', '-q', 'root-dir/mountpoint'
- ]),
- call([
- 'umount', '-l', 'root-dir/mountpoint'
- ]),
- call([
- 'rmdir', '-p', '--ignore-fail-on-non-empty',
- 'root-dir/mountpoint'
- ]),
- call([
- 'rm', '-f', 'root-dir/foo.kiwi', 'root-dir/foo'
- ])
- ]
-
- @patch('kiwi.command.Command.run')
@patch('os.path.islink')
@patch('kiwi.logger.log.warning')
- def test_cleanup_continue_on_raise(
- self, mock_warn, mock_islink, mock_command
- ):
-
- def side_effect(arg):
- if arg[0] == 'mountpoint':
- return
- else:
- raise Exception
-
- mock_command.side_effect = side_effect
- mock_islink.return_value = True
- self.bind_root.cleanup()
- assert mock_warn.called
-
@patch('kiwi.command.Command.run')
- @patch('os.path.islink')
- @patch('kiwi.logger.log.warning')
- def test_cleanup_mountpoint_invalid(
- self, mock_warn, mock_islink, mock_command
+ @patch('kiwi.root_bind.Path.remove_hierarchy')
+ def test_cleanup_continue_on_error(
+ self, mock_remove_hierarchy, mock_command, mock_warn, mock_islink
):
mock_islink.return_value = True
+ mock_remove_hierarchy.side_effect = Exception
mock_command.side_effect = Exception
+ self.mount_manager.umount_lazy.side_effect = Exception
self.bind_root.cleanup()
- assert mock_command.call_args_list == [
- call([
- 'mountpoint', '-q', 'root-dir/mountpoint'
- ]),
- call([
- 'rmdir', '-p', '--ignore-fail-on-non-empty',
- 'root-dir/mountpoint'
- ]),
- call([
- 'rm', '-f', 'root-dir/foo.kiwi', 'root-dir/foo'
- ])
+ assert mock_warn.call_args_list == [
+ call(
+ 'Image root directory %s not cleanly umounted: %s',
+ 'root-dir', ''
+ ),
+ call(
+ 'Failed to remove directory %s: %s', '/mountpoint', ''
+ ),
+ call(
+ 'Failed to remove intermediate config files: %s', ''
+ )
]
+ @patch('kiwi.logger.log.warning')
+ @patch('kiwi.command.Command.run')
+ @patch('kiwi.root_bind.Path.remove_hierarchy')
+ def test_cleanup_nothing_mounted(
+ self, mock_remove_hierarchy, mock_command, mock_warn
+ ):
+ self.mount_manager.is_mounted.return_value = False
+ self.mount_manager.mountpoint = '/mountpoint'
+ self.bind_root.cleanup()
+ mock_warn.assert_called_once_with(
+ 'Path %s not a mountpoint', '/mountpoint'
+ )
+
def test_move_to_root(self):
assert self.bind_root.move_to_root(
[self.bind_root.root_dir + '/argument']
diff --git a/test/unit/uri_test.py b/test/unit/uri_test.py
index c4d571b3..b1b469ec 100644
--- a/test/unit/uri_test.py
+++ b/test/unit/uri_test.py
@@ -5,11 +5,7 @@ import mock
from . import nose_helper
-from kiwi.exceptions import (
- KiwiUriStyleUnknown,
- KiwiUriTypeUnknown
-)
-
+from kiwi.exceptions import *
from kiwi.uri import Uri
import hashlib
@@ -71,27 +67,34 @@ class TestUri(object):
uri = Uri('http://example.com/foo', 'rpm-md')
assert uri.translate() == 'http://example.com/foo'
- @patch('kiwi.command.Command.run')
+ @patch('kiwi.uri.MountManager')
@patch('kiwi.uri.mkdtemp')
- def test_translate_iso_path(self, mock_mkdtemp, mock_command):
- mock_mkdtemp.return_value = '/tmp/foo'
+ def test_translate_iso_path(self, mock_mkdtemp, mock_manager):
+ mock_mkdtemp.return_value = 'tmpdir'
+ manager = mock.Mock()
+ manager.mountpoint = mock_mkdtemp.return_value
+ mock_manager.return_value = manager
uri = Uri('iso:///image/CDs/openSUSE-13.2-DVD-x86_64.iso', 'yast2')
result = uri.translate()
- mock_command.assert_called_once_with(
- ['mount', '/image/CDs/openSUSE-13.2-DVD-x86_64.iso', '/tmp/foo']
+ mock_manager.assert_called_once_with(
+ device='/image/CDs/openSUSE-13.2-DVD-x86_64.iso',
+ mountpoint='tmpdir'
)
- assert result == '/tmp/foo'
+ manager.mount.assert_called_once_with()
+ assert result == 'tmpdir'
def test_translate_suse_buildservice_path(self):
uri = Uri('suse://openSUSE:13.2/standard', 'yast2')
assert uri.translate() == \
'/usr/src/packages/SOURCES/repos/openSUSE:13.2/standard'
- @patch('kiwi.command.Command.run')
+ @patch('kiwi.uri.MountManager')
@patch('kiwi.uri.mkdtemp')
- def test_destructor(self, mock_mkdtemp, mock_command):
- mock_mkdtemp.return_value = '/tmp/foo'
+ def test_destructor(self, mock_mkdtemp, mock_manager):
+ manager = mock.Mock()
+ mock_mkdtemp.return_value = 'tmpdir'
+ mock_manager.return_value = manager
uri = Uri('iso:///image/CDs/openSUSE-13.2-DVD-x86_64.iso', 'yast2')
result = uri.translate()
- mock_command.side_effect = KeyError
- del uri
+ uri.__del__()
+ manager.umount.assert_called_once_with()
diff --git a/test/unit/volume_manager_base_test.py b/test/unit/volume_manager_base_test.py
index bb53ebc5..7556d9a8 100644
--- a/test/unit/volume_manager_base_test.py
+++ b/test/unit/volume_manager_base_test.py
@@ -99,25 +99,8 @@ class TestVolumeManagerBase(object):
def test_mount_volumes(self):
self.volume_manager.mount_volumes()
- @patch('kiwi.volume_manager_base.Command.run')
- def test_is_mounted_true(self, mock_command):
- self.volume_manager.mountpoint = 'mountpoint'
- assert self.volume_manager.is_mounted() is True
- mock_command.assert_called_once_with(
- ['mountpoint', 'mountpoint']
- )
-
- @patch('kiwi.volume_manager_base.Command.run')
- def test_is_mounted_false(self, mock_command):
- mock_command.side_effect = Exception
- self.volume_manager.mountpoint = 'mountpoint'
- assert self.volume_manager.is_mounted() is False
- mock_command.assert_called_once_with(
- ['mountpoint', 'mountpoint']
- )
-
@patch('kiwi.volume_manager_base.DataSync')
- @patch('kiwi.volume_manager_base.VolumeManagerBase.is_mounted')
+ @patch('kiwi.volume_manager_base.MountManager.is_mounted')
def test_sync_data(self, mock_mounted, mock_sync):
data_sync = mock.Mock()
mock_sync.return_value = data_sync
diff --git a/test/unit/volume_manager_btrfs_test.py b/test/unit/volume_manager_btrfs_test.py
index eae32684..8c4c4443 100644
--- a/test/unit/volume_manager_btrfs_test.py
+++ b/test/unit/volume_manager_btrfs_test.py
@@ -12,7 +12,9 @@ from kiwi.volume_manager_btrfs import VolumeManagerBtrfs
class TestVolumeManagerBtrfs(object):
@patch('os.path.exists')
- def setup(self, mock_path):
+ @patch('kiwi.volume_manager_base.mkdtemp')
+ def setup(self, mock_mkdtemp, mock_path):
+ mock_mkdtemp.return_value = 'tmpdir'
self.volume_type = namedtuple(
'volume_type', [
'name',
@@ -64,12 +66,13 @@ class TestVolumeManagerBtrfs(object):
@patch('kiwi.volume_manager_btrfs.Command.run')
@patch('kiwi.volume_manager_btrfs.FileSystem')
@patch('kiwi.volume_manager_btrfs.MappedDevice')
- @patch('kiwi.volume_manager_base.mkdtemp')
+ @patch('kiwi.volume_manager_btrfs.MountManager')
def test_setup_no_snapshot(
- self, mock_temp, mock_mapped_device, mock_fs,
+ self, mock_mount, mock_mapped_device, mock_fs,
mock_command, mock_os_exists
):
- mock_temp.return_value = 'tmpdir'
+ toplevel_mount = mock.Mock()
+ mock_mount.return_value = toplevel_mount
command_call = mock.Mock()
command_call.output = 'ID 256 gen 23 top level 5 path @'
mock_mapped_device.return_value = 'mapped_device'
@@ -78,8 +81,11 @@ class TestVolumeManagerBtrfs(object):
self.volume_manager.setup()
+ mock_mount.assert_called_once_with(
+ device='/dev/storage', mountpoint='tmpdir'
+ )
+ toplevel_mount.mount.assert_called_once_with()
assert mock_command.call_args_list == [
- call(['mount', '/dev/storage', 'tmpdir']),
call(['btrfs', 'subvolume', 'create', 'tmpdir/@']),
call(['btrfs', 'subvolume', 'list', 'tmpdir']),
call(['btrfs', 'subvolume', 'set-default', '256', 'tmpdir'])
@@ -89,12 +95,13 @@ class TestVolumeManagerBtrfs(object):
@patch('kiwi.volume_manager_btrfs.Command.run')
@patch('kiwi.volume_manager_btrfs.FileSystem')
@patch('kiwi.volume_manager_btrfs.MappedDevice')
- @patch('kiwi.volume_manager_base.mkdtemp')
+ @patch('kiwi.volume_manager_btrfs.MountManager')
def test_setup_with_snapshot(
- self, mock_temp, mock_mapped_device, mock_fs,
+ self, mock_mount, mock_mapped_device, mock_fs,
mock_command, mock_os_exists
):
- mock_temp.return_value = 'tmpdir'
+ toplevel_mount = mock.Mock()
+ mock_mount.return_value = toplevel_mount
command_call = mock.Mock()
command_call.output = \
'ID 258 gen 26 top level 257 path @/.snapshots/1/snapshot'
@@ -105,8 +112,11 @@ class TestVolumeManagerBtrfs(object):
self.volume_manager.setup()
+ mock_mount.assert_called_once_with(
+ device='/dev/storage', mountpoint='tmpdir'
+ )
+ toplevel_mount.mount.assert_called_once_with()
assert mock_command.call_args_list == [
- call(['mount', '/dev/storage', 'tmpdir']),
call(['btrfs', 'subvolume', 'create', 'tmpdir/@']),
call(['btrfs', 'subvolume', 'create', 'tmpdir/@/.snapshots']),
call(['mkdir', '-p', 'tmpdir/@/.snapshots/1']),
@@ -123,9 +133,9 @@ class TestVolumeManagerBtrfs(object):
@patch('kiwi.volume_manager_btrfs.Command.run')
@patch('kiwi.volume_manager_btrfs.FileSystem')
@patch('kiwi.volume_manager_btrfs.MappedDevice')
- @patch('kiwi.volume_manager_base.mkdtemp')
+ @patch('kiwi.volume_manager_btrfs.MountManager')
def test_setup_volume_id_not_detected(
- self, mock_temp, mock_mapped_device, mock_fs,
+ self, mock_mount, mock_mapped_device, mock_fs,
mock_command, mock_os_exists
):
command_call = mock.Mock()
@@ -137,96 +147,87 @@ class TestVolumeManagerBtrfs(object):
@patch('os.path.exists')
@patch('kiwi.volume_manager_btrfs.Command.run')
- def test_create_volumes(self, mock_command, mock_os_exists):
+ @patch('kiwi.volume_manager_btrfs.MountManager')
+ @patch('kiwi.volume_manager_btrfs.Path.create')
+ def test_create_volumes(
+ self, mock_path, mock_mount, mock_command, mock_os_exists
+ ):
+ volume_mount = mock.Mock()
+ mock_mount.return_value = volume_mount
self.volume_manager.mountpoint = 'tmpdir'
+ self.volume_manager.custom_args['root_is_snapshot'] = True
mock_os_exists.return_value = False
+
self.volume_manager.create_volumes('btrfs')
+ assert mock_path.call_args_list == [
+ call('root_dir/etc'),
+ call('root_dir/data'),
+ call('root_dir/home'),
+ call('tmpdir/@'),
+ call('tmpdir/@'),
+ call('tmpdir/@')
+ ]
assert mock_command.call_args_list == [
- call(['mkdir', '-p', 'root_dir/etc']),
- call(['mkdir', '-p', 'root_dir/data']),
- call(['mkdir', '-p', 'root_dir/home']),
- call(['mkdir', '-p', 'tmpdir/@']),
call(['btrfs', 'subvolume', 'create', 'tmpdir/@/data']),
- call(['mkdir', '-p', 'tmpdir/@']),
call(['btrfs', 'subvolume', 'create', 'tmpdir/@/etc']),
- call(['mkdir', '-p', 'tmpdir/@']),
call(['btrfs', 'subvolume', 'create', 'tmpdir/@/home'])
]
+ assert mock_mount.call_args_list == [
+ call(
+ device='/dev/storage',
+ mountpoint='tmpdir/@/.snapshots/1/snapshot/data'
+ ),
+ call(
+ device='/dev/storage',
+ mountpoint='tmpdir/@/.snapshots/1/snapshot/etc'
+ ),
+ call(
+ device='/dev/storage',
+ mountpoint='tmpdir/@/.snapshots/1/snapshot/home'
+ )
+ ]
@patch('os.path.exists')
- @patch('kiwi.volume_manager_btrfs.Command.run')
- def test_mount_volumes(self, mock_command, mock_os_exists):
+ @patch('kiwi.volume_manager_btrfs.Path.create')
+ def test_mount_volumes(self, mock_path, mock_os_exists):
mock_os_exists.return_value = False
- self.volume_manager.mountpoint = 'tmpdir'
- self.volume_manager.custom_args['root_is_snapshot'] = True
- self.volume_manager.subvol_mount_list = [
- '/var/log', '/etc'
- ]
+ volume_mount = mock.Mock()
+ volume_mount.mountpoint = 'tmpdir/@/.snapshots/1/snapshot/subvol'
+ self.volume_manager.subvol_mount_list = [volume_mount]
+
self.volume_manager.mount_volumes()
- assert mock_command.call_args_list == [
- call([
- 'mkdir', '-p', 'tmpdir/@/.snapshots/1/snapshot/var/log'
- ]),
- call([
- 'mount', '/dev/storage',
- 'tmpdir/@/.snapshots/1/snapshot/var/log',
- '-o', 'subvol=@/var/log']),
- call([
- 'mkdir', '-p', 'tmpdir/@/.snapshots/1/snapshot/etc'
- ]),
- call([
- 'mount', '/dev/storage',
- 'tmpdir/@/.snapshots/1/snapshot/etc',
- '-o', 'subvol=@/etc'
- ])
- ]
+
+ mock_path.assert_called_once_with(volume_mount.mountpoint)
+ volume_mount.mount.assert_called_once_with(
+ options=['subvol=@/subvol']
+ )
@patch('kiwi.volume_manager_btrfs.DataSync')
- @patch('kiwi.volume_manager_btrfs.VolumeManagerBtrfs.is_mounted')
- def test_sync_data(self, mock_mounted, mock_sync):
- mock_mounted.return_value = True
+ def test_sync_data(self, mock_sync):
+ self.volume_manager.toplevel_mount = mock.Mock()
self.volume_manager.mountpoint = 'tmpdir'
self.volume_manager.custom_args['root_is_snapshot'] = True
sync = mock.Mock()
mock_sync.return_value = sync
+
self.volume_manager.sync_data(['exclude_me'])
+
mock_sync.assert_called_once_with(
'root_dir', 'tmpdir/@/.snapshots/1/snapshot'
)
- sync.sync_data.assert_called_once_with(['exclude_me'])
+ sync.sync_data.assert_called_once_with(
+ ['exclude_me']
+ )
- @patch('time.sleep')
- @patch('kiwi.volume_manager_btrfs.VolumeManagerBtrfs.is_mounted')
- @patch('kiwi.volume_manager_btrfs.Path.wipe')
- @patch('kiwi.volume_manager_btrfs.Command.run')
- def test_destructor_success(
- self, mock_command, mock_wipe, mock_mounted, mock_time
- ):
- self.volume_manager.mountpoint = 'tmpdir'
- self.volume_manager.custom_args['root_is_snapshot'] = True
- self.volume_manager.subvol_mount_list = ['/var/log', 'etc']
- mock_mounted.return_value = True
- self.volume_manager.__del__()
- assert mock_command.call_args_list == [
- call(['umount', 'tmpdir/@/.snapshots/1/snapshot/etc']),
- call(['umount', 'tmpdir/@/.snapshots/1/snapshot/var/log']),
- call(['umount', '/dev/storage'])
- ]
- mock_wipe.assert_called_once_with('tmpdir')
- self.volume_manager.mountpoint = None
+ def test_destructor(self):
+ self.volume_manager.toplevel_mount = mock.Mock()
+ volume_mount = mock.Mock()
+ self.volume_manager.subvol_mount_list = [volume_mount]
- @patch('time.sleep')
- @patch('kiwi.volume_manager_btrfs.VolumeManagerBtrfs.is_mounted')
- @patch('kiwi.volume_manager_btrfs.Path')
- @patch('kiwi.volume_manager_btrfs.Command.run')
- @patch('kiwi.logger.log.warning')
- def test_destructor_failed(
- self, mock_log_warn, mock_command, mock_path, mock_mounted, mock_time
- ):
- self.volume_manager.mountpoint = 'tmpdir'
- mock_command.side_effect = Exception
- mock_mounted.return_value = True
self.volume_manager.__del__()
- assert mock_log_warn.called
- self.volume_manager.mountpoint = None
+
+ volume_mount.umount.assert_called_once_with(
+ delete_mountpoint=False
+ )
+ self.volume_manager.toplevel_mount.umount.assert_called_once_with()
diff --git a/test/unit/volume_manager_lvm_test.py b/test/unit/volume_manager_lvm_test.py
index 4e363b5e..2ec24d63 100644
--- a/test/unit/volume_manager_lvm_test.py
+++ b/test/unit/volume_manager_lvm_test.py
@@ -13,12 +13,9 @@ from kiwi.defaults import Defaults
class TestVolumeManagerLVM(object):
@patch('os.path.exists')
- def setup(self, mock_path):
- self.mount_type = namedtuple(
- 'mount_type', [
- 'device', 'mountpoint'
- ]
- )
+ @patch('kiwi.volume_manager_base.mkdtemp')
+ def setup(self, mock_mkdtemp, mock_path):
+ mock_mkdtemp.return_value = 'tmpdir'
self.volume_type = namedtuple(
'volume_type', [
'name',
@@ -112,9 +109,10 @@ class TestVolumeManagerLVM(object):
@patch('kiwi.volume_manager_lvm.Command.run')
@patch('kiwi.volume_manager_lvm.FileSystem')
@patch('kiwi.volume_manager_lvm.MappedDevice')
+ @patch('kiwi.volume_manager_lvm.MountManager')
def test_create_volumes(
- self, mock_mapped_device, mock_fs, mock_command, mock_size,
- mock_os_exists
+ self, mock_mount, mock_mapped_device, mock_fs, mock_command,
+ mock_size, mock_os_exists
):
mock_mapped_device.return_value = 'mapped_device'
size = mock.Mock()
@@ -128,6 +126,13 @@ class TestVolumeManagerLVM(object):
myvol_size = 500
etc_size = 200 + 42 + Defaults.get_min_volume_mbytes()
root_size = 100 + 42 + Defaults.get_min_volume_mbytes()
+
+ assert mock_mount.call_args_list == [
+ call(device='/dev/volume_group/LVRoot', mountpoint='tmpdir'),
+ call(device='/dev/volume_group/myvol', mountpoint='tmpdir//data'),
+ call(device='/dev/volume_group/LVetc', mountpoint='tmpdir//etc'),
+ call(device='/dev/volume_group/LVhome', mountpoint='tmpdir//home')
+ ]
assert mock_command.call_args_list == [
call(['mkdir', '-p', 'root_dir/etc']),
call(['mkdir', '-p', 'root_dir/data']),
@@ -154,73 +159,47 @@ class TestVolumeManagerLVM(object):
call('ext3', 'mapped_device'),
call('ext3', 'mapped_device')
]
- assert self.volume_manager.mount_list == [
- self.mount_type(
- device='/dev/volume_group/LVRoot', mountpoint='/'
- ),
- self.mount_type(
- device='/dev/volume_group/myvol', mountpoint='//data'
- ),
- self.mount_type(
- device='/dev/volume_group/LVetc', mountpoint='//etc'
- ),
- self.mount_type(
- device='/dev/volume_group/LVhome', mountpoint='//home'
- )
- ]
self.volume_manager.volume_group = None
@patch('kiwi.volume_manager_lvm.Path')
- @patch('kiwi.volume_manager_base.mkdtemp')
- @patch('kiwi.volume_manager_lvm.Command.run')
- def test_mount_volumes(self, mock_command, mock_temp, mock_path):
- mock_temp.return_value = 'tmpdir'
- self.volume_manager.mount_list = [
- self.mount_type(device='/dev/volume_group/LVRoot', mountpoint='/')
- ]
+ def test_mount_volumes(self, mock_path):
+ volume_mount = mock.Mock()
+ volume_mount.mountpoint = 'volume_mount_point'
+ self.volume_manager.mount_list = [volume_mount]
self.volume_manager.mount_volumes()
- mock_path.create.assert_called_once_with('tmpdir/')
- mock_command.assert_called_once_with(
- ['mount', '/dev/volume_group/LVRoot', 'tmpdir/']
- )
+ mock_path.create.assert_called_once_with(volume_mount.mountpoint)
+ volume_mount.mount.assert_called_once_with()
- @patch('time.sleep')
- @patch('kiwi.volume_manager_lvm.VolumeManagerLVM.is_mounted')
@patch('kiwi.volume_manager_lvm.Path.wipe')
@patch('kiwi.volume_manager_lvm.Command.run')
- def test_destructor_success(
- self, mock_command, mock_wipe, mock_mounted, mock_time
- ):
+ def test_destructor_busy_volumes(self, mock_command, mock_wipe):
self.volume_manager.mountpoint = 'tmpdir'
- self.volume_manager.mount_list = [
- self.mount_type(device='/dev/volume_group/LVRoot', mountpoint='/')
- ]
- mock_mounted.return_value = True
self.volume_manager.volume_group = 'volume_group'
+ volume_mount = mock.Mock()
+ volume_mount.is_mounted.return_value = True
+ volume_mount.umount.return_value = False
+ volume_mount.mountpoint = 'volume_mount_point'
+ volume_mount.device = '/dev/volume_group/LVRoot'
+ self.volume_manager.mount_list = [volume_mount]
+
self.volume_manager.__del__()
- call = mock_command.call_args_list[0]
- assert mock_command.call_args_list == [
- call(['umount', '/dev/volume_group/LVRoot']),
- call(['vgchange', '-an', 'volume_group'])
- ]
- mock_wipe.assert_called_once_with('tmpdir')
+
+ volume_mount.umount.assert_called_once_with()
self.volume_manager.volume_group = None
- @patch('time.sleep')
- @patch('kiwi.volume_manager_lvm.VolumeManagerLVM.is_mounted')
- @patch('kiwi.volume_manager_lvm.Path')
+ @patch('kiwi.volume_manager_lvm.Path.wipe')
@patch('kiwi.volume_manager_lvm.Command.run')
@patch('kiwi.logger.log.warning')
- def test_destructor_failed(
- self, mock_log_warn, mock_command, mock_path, mock_mounted, mock_time
- ):
- self.volume_manager.mountpoint = 'tmpdir'
+ def test_destructor(self, mock_warn, mock_command, mock_wipe):
mock_command.side_effect = Exception
- self.volume_manager.mount_list = [
- self.mount_type(device='/dev/volume_group/LVRoot', mountpoint='/')
- ]
- mock_mounted.return_value = True
+ self.volume_manager.mountpoint = 'tmpdir'
self.volume_manager.volume_group = 'volume_group'
+
self.volume_manager.__del__()
- assert mock_log_warn.called
+
+ mock_wipe.assert_called_once_with('tmpdir')
+ mock_command.assert_called_once_with(
+ ['vgchange', '-an', 'volume_group']
+ )
+ assert mock_warn.called
self.volume_manager.volume_group = None