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

Refactor use of mount/umount calls

Browse Source 
Provide a MountManager class and handle all mount/umount
calls in instances of MountManager
This commit is contained in:
Marcus Schäfer 2016-02-26 16:39:24 +01:00
parent 31bffb7b3d
commit 04764a42df
18 changed files with 599 additions and 706 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
kiwi/bootloader_install_grub2.py
View File

@ -15,15 +15,12 @@
# You should have received a copy of the GNU General Public License
# along with kiwi. If not, see <http://www.gnu.org/licenses/>
#
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()

55
kiwi/bootloader_install_zipl.py
View File

@ -15,14 +15,11 @@
# You should have received a copy of the GNU General Public License
# along with kiwi. If not, see <http://www.gnu.org/licenses/>
#
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()

62
kiwi/filesystem_base.py
View File

@ -16,14 +16,12 @@
# along with kiwi. If not, see <http://www.gnu.org/licenses/>
#
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()

97
kiwi/mount_manager.py Normal file
View File

@ -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 <http://www.gnu.org/licenses/>
#
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

51
kiwi/root_bind.py
View File

@ -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

21
kiwi/uri.py
View File

@ -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()

22
kiwi/volume_manager_base.py
View File

@ -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):
"""

78
kiwi/volume_manager_btrfs.py
View File

@ -15,13 +15,13 @@
# You should have received a copy of the GNU General Public License
# along with kiwi. If not, see <http://www.gnu.org/licenses/>
#
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()

70
kiwi/volume_manager_lvm.py
View File

@ -15,12 +15,10 @@
# You should have received a copy of the GNU General Public License
# along with kiwi. If not, see <http://www.gnu.org/licenses/>
#
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
)

92
test/unit/bootloader_install_grub2_test.py
View File

@ -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()

49
test/unit/bootloader_install_zipl_test.py
View File

@ -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()

81
test/unit/filesystem_base_test.py
View File

@ -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()

97
test/unit/mount_manager_test.py Normal file
View File

@ -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

138
test/unit/root_bind_test.py
View File

@ -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']

35
test/unit/uri_test.py
View File

@ -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()

19
test/unit/volume_manager_base_test.py
View File

@ -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

157
test/unit/volume_manager_btrfs_test.py
View File

@ -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()

99
test/unit/volume_manager_lvm_test.py
View File

@ -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