211 lines
8.0 KiB
Python
211 lines
8.0 KiB
Python
# Copyright 2015 Hewlett-Packard Development Company, L.P.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
import mock
|
|
|
|
from nova import test
|
|
from nova.virt.disk.mount import api
|
|
from nova.virt.disk.mount import block
|
|
from nova.virt.disk.mount import loop
|
|
from nova.virt.disk.mount import nbd
|
|
from nova.virt.image import model as imgmodel
|
|
|
|
|
|
PARTITION = 77
|
|
ORIG_DEVICE = "/dev/null"
|
|
AUTOMAP_PARTITION = "/dev/nullp77"
|
|
MAP_PARTITION = "/dev/mapper/nullp77"
|
|
|
|
|
|
class MountTestCase(test.NoDBTestCase):
|
|
def setUp(self):
|
|
super(MountTestCase, self).setUp()
|
|
|
|
def _test_map_dev(self, partition):
|
|
mount = api.Mount(mock.sentinel.image, mock.sentinel.mount_dir)
|
|
mount.device = ORIG_DEVICE
|
|
mount.partition = partition
|
|
mount.map_dev()
|
|
return mount
|
|
|
|
@mock.patch('nova.utils.trycmd')
|
|
def _test_map_dev_with_trycmd(self, partition, trycmd):
|
|
trycmd.return_value = [None, None]
|
|
mount = self._test_map_dev(partition)
|
|
self.assertEqual(1, trycmd.call_count) # don't care about args
|
|
return mount
|
|
|
|
def _exists_effect(self, data):
|
|
def exists_effect(filename):
|
|
try:
|
|
v = data[filename]
|
|
if isinstance(v, list):
|
|
if len(v) > 0:
|
|
return v.pop(0)
|
|
self.fail("Out of items for: %s" % filename)
|
|
return v
|
|
except KeyError:
|
|
self.fail("Unexpected call with: %s" % filename)
|
|
return exists_effect
|
|
|
|
def _check_calls(self, exists, filenames, trailing=0):
|
|
self.assertEqual([mock.call(x) for x in filenames],
|
|
exists.call_args_list[:len(filenames)])
|
|
self.assertEqual([mock.call(MAP_PARTITION)] * trailing,
|
|
exists.call_args_list[len(filenames):])
|
|
|
|
@mock.patch('os.path.exists')
|
|
def test_map_dev_partition_search(self, exists):
|
|
exists.side_effect = self._exists_effect({
|
|
ORIG_DEVICE: True})
|
|
mount = self._test_map_dev(-1)
|
|
self._check_calls(exists, [ORIG_DEVICE])
|
|
self.assertNotEqual("", mount.error)
|
|
self.assertFalse(mount.mapped)
|
|
|
|
@mock.patch('os.path.exists')
|
|
def test_map_dev_good(self, exists):
|
|
exists.side_effect = self._exists_effect({
|
|
ORIG_DEVICE: True,
|
|
AUTOMAP_PARTITION: False,
|
|
MAP_PARTITION: [False, True]})
|
|
mount = self._test_map_dev_with_trycmd(PARTITION)
|
|
self._check_calls(exists, [ORIG_DEVICE, AUTOMAP_PARTITION], 2)
|
|
self.assertEqual("", mount.error)
|
|
self.assertTrue(mount.mapped)
|
|
|
|
@mock.patch('os.path.exists')
|
|
def test_map_dev_error(self, exists):
|
|
exists.side_effect = self._exists_effect({
|
|
ORIG_DEVICE: True,
|
|
AUTOMAP_PARTITION: False,
|
|
MAP_PARTITION: False})
|
|
mount = self._test_map_dev_with_trycmd(PARTITION)
|
|
self._check_calls(exists, [ORIG_DEVICE, AUTOMAP_PARTITION],
|
|
api.MAX_FILE_CHECKS + 1)
|
|
self.assertNotEqual("", mount.error)
|
|
self.assertFalse(mount.mapped)
|
|
|
|
@mock.patch('os.path.exists')
|
|
def test_map_dev_error_then_pass(self, exists):
|
|
exists.side_effect = self._exists_effect({
|
|
ORIG_DEVICE: True,
|
|
AUTOMAP_PARTITION: False,
|
|
MAP_PARTITION: [False, False, True]})
|
|
mount = self._test_map_dev_with_trycmd(PARTITION)
|
|
self._check_calls(exists, [ORIG_DEVICE, AUTOMAP_PARTITION], 3)
|
|
self.assertEqual("", mount.error)
|
|
self.assertTrue(mount.mapped)
|
|
|
|
@mock.patch('os.path.exists')
|
|
def test_map_dev_automap(self, exists):
|
|
exists.side_effect = self._exists_effect({
|
|
ORIG_DEVICE: True,
|
|
AUTOMAP_PARTITION: True})
|
|
mount = self._test_map_dev(PARTITION)
|
|
self._check_calls(exists,
|
|
[ORIG_DEVICE, AUTOMAP_PARTITION, AUTOMAP_PARTITION])
|
|
self.assertEqual(AUTOMAP_PARTITION, mount.mapped_device)
|
|
self.assertTrue(mount.automapped)
|
|
self.assertTrue(mount.mapped)
|
|
|
|
@mock.patch('os.path.exists')
|
|
def test_map_dev_else(self, exists):
|
|
exists.side_effect = self._exists_effect({
|
|
ORIG_DEVICE: True,
|
|
AUTOMAP_PARTITION: True})
|
|
mount = self._test_map_dev(None)
|
|
self._check_calls(exists, [ORIG_DEVICE])
|
|
self.assertEqual(ORIG_DEVICE, mount.mapped_device)
|
|
self.assertFalse(mount.automapped)
|
|
self.assertTrue(mount.mapped)
|
|
|
|
def test_instance_for_format_raw(self):
|
|
image = imgmodel.LocalFileImage("/some/file.raw",
|
|
imgmodel.FORMAT_RAW)
|
|
mount_dir = '/mount/dir'
|
|
partition = -1
|
|
inst = api.Mount.instance_for_format(image, mount_dir, partition)
|
|
self.assertIsInstance(inst, loop.LoopMount)
|
|
|
|
def test_instance_for_format_qcow2(self):
|
|
image = imgmodel.LocalFileImage("/some/file.qcows",
|
|
imgmodel.FORMAT_QCOW2)
|
|
mount_dir = '/mount/dir'
|
|
partition = -1
|
|
inst = api.Mount.instance_for_format(image, mount_dir, partition)
|
|
self.assertIsInstance(inst, nbd.NbdMount)
|
|
|
|
def test_instance_for_format_block(self):
|
|
image = imgmodel.LocalBlockImage(
|
|
"/dev/mapper/instances--instance-0000001_disk",)
|
|
mount_dir = '/mount/dir'
|
|
partition = -1
|
|
inst = api.Mount.instance_for_format(image, mount_dir, partition)
|
|
self.assertIsInstance(inst, block.BlockMount)
|
|
|
|
def test_instance_for_device_loop(self):
|
|
image = mock.MagicMock()
|
|
mount_dir = '/mount/dir'
|
|
partition = -1
|
|
device = '/dev/loop0'
|
|
inst = api.Mount.instance_for_device(image, mount_dir, partition,
|
|
device)
|
|
self.assertIsInstance(inst, loop.LoopMount)
|
|
|
|
def test_instance_for_device_loop_partition(self):
|
|
image = mock.MagicMock()
|
|
mount_dir = '/mount/dir'
|
|
partition = 1
|
|
device = '/dev/mapper/loop0p1'
|
|
inst = api.Mount.instance_for_device(image, mount_dir, partition,
|
|
device)
|
|
self.assertIsInstance(inst, loop.LoopMount)
|
|
|
|
def test_instance_for_device_nbd(self):
|
|
image = mock.MagicMock()
|
|
mount_dir = '/mount/dir'
|
|
partition = -1
|
|
device = '/dev/nbd0'
|
|
inst = api.Mount.instance_for_device(image, mount_dir, partition,
|
|
device)
|
|
self.assertIsInstance(inst, nbd.NbdMount)
|
|
|
|
def test_instance_for_device_nbd_partition(self):
|
|
image = mock.MagicMock()
|
|
mount_dir = '/mount/dir'
|
|
partition = 1
|
|
device = '/dev/mapper/nbd0p1'
|
|
inst = api.Mount.instance_for_device(image, mount_dir, partition,
|
|
device)
|
|
self.assertIsInstance(inst, nbd.NbdMount)
|
|
|
|
def test_instance_for_device_block(self):
|
|
image = mock.MagicMock()
|
|
mount_dir = '/mount/dir'
|
|
partition = -1
|
|
device = '/dev/mapper/instances--instance-0000001_disk'
|
|
inst = api.Mount.instance_for_device(image, mount_dir, partition,
|
|
device)
|
|
self.assertIsInstance(inst, block.BlockMount)
|
|
|
|
def test_instance_for_device_block_partiton(self,):
|
|
image = mock.MagicMock()
|
|
mount_dir = '/mount/dir'
|
|
partition = 1
|
|
device = '/dev/mapper/instances--instance-0000001_diskp1'
|
|
inst = api.Mount.instance_for_device(image, mount_dir, partition,
|
|
device)
|
|
self.assertIsInstance(inst, block.BlockMount)
|