36f310f469
When attaching block devices, we find the next device name to use by looking at the used device letters in the block device mappings and selecting the next available letter, as long as the letter is within 'a' - 'z' or 'aa' - 'zz'. This allows us to represent 702 devices (and would interestingly hit an IndexError if all 'a' - 'z' and 'aa' - 'zz' letters were in use already. This adds a method for generating device names universally, not limited to any particular maximum, based on a device index. For example: vda(index=0), sdb(index=1), vdaa(index=26), sdabc(index=730), ... This method and its helper method will be used in a subsequent patch. Part of blueprint conf-max-attach-volumes Change-Id: I2699f600a98222a4dc58903ec97ca70e57701c93
726 lines
31 KiB
Python
726 lines
31 KiB
Python
# Copyright 2011 Isaku Yamahata
|
|
# All Rights Reserved.
|
|
#
|
|
# 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.
|
|
|
|
"""
|
|
Tests for Block Device utility functions.
|
|
"""
|
|
from oslo_utils.fixture import uuidsentinel as uuids
|
|
import six
|
|
|
|
from nova import block_device
|
|
from nova import exception
|
|
from nova import objects
|
|
from nova import test
|
|
from nova.tests.unit import fake_block_device
|
|
from nova.tests.unit import matchers
|
|
|
|
|
|
class BlockDeviceTestCase(test.NoDBTestCase):
|
|
def setUp(self):
|
|
super(BlockDeviceTestCase, self).setUp()
|
|
BDM = block_device.BlockDeviceDict
|
|
|
|
self.new_mapping = [
|
|
BDM({'id': 1, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sdb1',
|
|
'source_type': 'blank',
|
|
'destination_type': 'local',
|
|
'delete_on_termination': True,
|
|
'volume_size': 1,
|
|
'guest_format': 'swap',
|
|
'boot_index': -1}),
|
|
BDM({'id': 2, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sdc1',
|
|
'source_type': 'blank',
|
|
'destination_type': 'local',
|
|
'volume_size': 10,
|
|
'delete_on_termination': True,
|
|
'boot_index': -1}),
|
|
BDM({'id': 3, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sda1',
|
|
'source_type': 'volume',
|
|
'destination_type': 'volume',
|
|
'volume_id': 'fake-volume-id-1',
|
|
'connection_info': "{'fake': 'connection_info'}",
|
|
'boot_index': 0}),
|
|
BDM({'id': 4, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sda2',
|
|
'source_type': 'snapshot',
|
|
'destination_type': 'volume',
|
|
'connection_info': "{'fake': 'connection_info'}",
|
|
'snapshot_id': 'fake-snapshot-id-1',
|
|
'volume_id': 'fake-volume-id-2',
|
|
'boot_index': -1}),
|
|
BDM({'id': 5, 'instance_uuid': uuids.instance,
|
|
'no_device': True,
|
|
'device_name': '/dev/vdc'}),
|
|
]
|
|
|
|
def test_properties(self):
|
|
root_device0 = '/dev/sda'
|
|
root_device1 = '/dev/sdb'
|
|
mappings = [{'virtual': 'root',
|
|
'device': root_device0}]
|
|
|
|
properties0 = {'mappings': mappings}
|
|
properties1 = {'mappings': mappings,
|
|
'root_device_name': root_device1}
|
|
|
|
self.assertIsNone(block_device.properties_root_device_name({}))
|
|
self.assertEqual(root_device0,
|
|
block_device.properties_root_device_name(properties0))
|
|
self.assertEqual(root_device1,
|
|
block_device.properties_root_device_name(properties1))
|
|
|
|
def test_ephemeral(self):
|
|
self.assertFalse(block_device.is_ephemeral('ephemeral'))
|
|
self.assertTrue(block_device.is_ephemeral('ephemeral0'))
|
|
self.assertTrue(block_device.is_ephemeral('ephemeral1'))
|
|
self.assertTrue(block_device.is_ephemeral('ephemeral11'))
|
|
self.assertFalse(block_device.is_ephemeral('root'))
|
|
self.assertFalse(block_device.is_ephemeral('swap'))
|
|
self.assertFalse(block_device.is_ephemeral('/dev/sda1'))
|
|
|
|
self.assertEqual(0, block_device.ephemeral_num('ephemeral0'))
|
|
self.assertEqual(1, block_device.ephemeral_num('ephemeral1'))
|
|
self.assertEqual(11, block_device.ephemeral_num('ephemeral11'))
|
|
|
|
self.assertFalse(block_device.is_swap_or_ephemeral('ephemeral'))
|
|
self.assertTrue(block_device.is_swap_or_ephemeral('ephemeral0'))
|
|
self.assertTrue(block_device.is_swap_or_ephemeral('ephemeral1'))
|
|
self.assertTrue(block_device.is_swap_or_ephemeral('swap'))
|
|
self.assertFalse(block_device.is_swap_or_ephemeral('root'))
|
|
self.assertFalse(block_device.is_swap_or_ephemeral('/dev/sda1'))
|
|
|
|
def test_mappings_prepend_dev(self):
|
|
mapping = [
|
|
{'virtual': 'ami', 'device': '/dev/sda'},
|
|
{'virtual': 'root', 'device': 'sda'},
|
|
{'virtual': 'ephemeral0', 'device': 'sdb'},
|
|
{'virtual': 'swap', 'device': 'sdc'},
|
|
{'virtual': 'ephemeral1', 'device': 'sdd'},
|
|
{'virtual': 'ephemeral2', 'device': 'sde'}]
|
|
|
|
expected = [
|
|
{'virtual': 'ami', 'device': '/dev/sda'},
|
|
{'virtual': 'root', 'device': 'sda'},
|
|
{'virtual': 'ephemeral0', 'device': '/dev/sdb'},
|
|
{'virtual': 'swap', 'device': '/dev/sdc'},
|
|
{'virtual': 'ephemeral1', 'device': '/dev/sdd'},
|
|
{'virtual': 'ephemeral2', 'device': '/dev/sde'}]
|
|
|
|
prepended = block_device.mappings_prepend_dev(mapping)
|
|
self.assertEqual(sorted(expected, key=lambda v: v['virtual']),
|
|
sorted(prepended, key=lambda v: v['virtual']))
|
|
|
|
def test_strip_dev(self):
|
|
self.assertEqual('sda', block_device.strip_dev('/dev/sda'))
|
|
self.assertEqual('sda', block_device.strip_dev('sda'))
|
|
self.assertIsNone(block_device.strip_dev(None))
|
|
|
|
def test_strip_prefix(self):
|
|
self.assertEqual('a', block_device.strip_prefix('/dev/sda'))
|
|
self.assertEqual('a', block_device.strip_prefix('a'))
|
|
self.assertEqual('a', block_device.strip_prefix('xvda'))
|
|
self.assertEqual('a', block_device.strip_prefix('vda'))
|
|
self.assertEqual('a', block_device.strip_prefix('hda'))
|
|
self.assertIsNone(block_device.strip_prefix(None))
|
|
|
|
def test_get_device_letter(self):
|
|
self.assertEqual('', block_device.get_device_letter(''))
|
|
self.assertEqual('a', block_device.get_device_letter('/dev/sda1'))
|
|
self.assertEqual('b', block_device.get_device_letter('/dev/xvdb'))
|
|
self.assertEqual('d', block_device.get_device_letter('/dev/d'))
|
|
self.assertEqual('a', block_device.get_device_letter('a'))
|
|
self.assertEqual('b', block_device.get_device_letter('sdb2'))
|
|
self.assertEqual('c', block_device.get_device_letter('vdc'))
|
|
self.assertEqual('c', block_device.get_device_letter('hdc'))
|
|
self.assertIsNone(block_device.get_device_letter(None))
|
|
|
|
def test_generate_device_name(self):
|
|
expected = (
|
|
('vda', ("vd", 0)),
|
|
('vdaa', ("vd", 26)),
|
|
('vdabc', ("vd", 730)),
|
|
('vdidpok', ("vd", 4194304)),
|
|
('sdc', ("sd", 2)),
|
|
('sdaa', ("sd", 26)),
|
|
('sdiw', ("sd", 256)),
|
|
('hdzz', ("hd", 701))
|
|
)
|
|
for res, args in expected:
|
|
self.assertEqual(res, block_device.generate_device_name(*args))
|
|
|
|
def test_volume_in_mapping(self):
|
|
swap = {'device_name': '/dev/sdb',
|
|
'swap_size': 1}
|
|
ephemerals = [{'num': 0,
|
|
'virtual_name': 'ephemeral0',
|
|
'device_name': '/dev/sdc1',
|
|
'size': 1},
|
|
{'num': 2,
|
|
'virtual_name': 'ephemeral2',
|
|
'device_name': '/dev/sdd',
|
|
'size': 1}]
|
|
block_device_mapping = [{'mount_device': '/dev/sde',
|
|
'device_path': 'fake_device'},
|
|
{'mount_device': '/dev/sdf',
|
|
'device_path': 'fake_device'}]
|
|
block_device_info = {
|
|
'root_device_name': '/dev/sda',
|
|
'swap': swap,
|
|
'ephemerals': ephemerals,
|
|
'block_device_mapping': block_device_mapping}
|
|
|
|
def _assert_volume_in_mapping(device_name, true_or_false):
|
|
in_mapping = block_device.volume_in_mapping(
|
|
device_name, block_device_info)
|
|
self.assertEqual(true_or_false, in_mapping)
|
|
|
|
_assert_volume_in_mapping('sda', False)
|
|
_assert_volume_in_mapping('sdb', True)
|
|
_assert_volume_in_mapping('sdc1', True)
|
|
_assert_volume_in_mapping('sdd', True)
|
|
_assert_volume_in_mapping('sde', True)
|
|
_assert_volume_in_mapping('sdf', True)
|
|
_assert_volume_in_mapping('sdg', False)
|
|
_assert_volume_in_mapping('sdh1', False)
|
|
|
|
def test_get_root_bdm(self):
|
|
root_bdm = {'device_name': 'vda', 'boot_index': 0}
|
|
bdms = [root_bdm,
|
|
{'device_name': 'vdb', 'boot_index': 1},
|
|
{'device_name': 'vdc', 'boot_index': -1},
|
|
{'device_name': 'vdd'}]
|
|
self.assertEqual(root_bdm, block_device.get_root_bdm(bdms))
|
|
self.assertEqual(root_bdm, block_device.get_root_bdm([bdms[0]]))
|
|
self.assertIsNone(block_device.get_root_bdm(bdms[1:]))
|
|
self.assertIsNone(block_device.get_root_bdm(bdms[2:]))
|
|
self.assertIsNone(block_device.get_root_bdm(bdms[3:]))
|
|
self.assertIsNone(block_device.get_root_bdm([]))
|
|
|
|
def test_get_bdm_ephemeral_disk_size(self):
|
|
size = block_device.get_bdm_ephemeral_disk_size(self.new_mapping)
|
|
self.assertEqual(10, size)
|
|
|
|
def test_get_bdm_swap_list(self):
|
|
swap_list = block_device.get_bdm_swap_list(self.new_mapping)
|
|
self.assertEqual(1, len(swap_list))
|
|
self.assertEqual(1, swap_list[0].get('id'))
|
|
|
|
def test_get_bdm_local_disk_num(self):
|
|
size = block_device.get_bdm_local_disk_num(self.new_mapping)
|
|
self.assertEqual(2, size)
|
|
|
|
def test_new_format_is_swap(self):
|
|
expected_results = [True, False, False, False, False]
|
|
for expected, bdm in zip(expected_results, self.new_mapping):
|
|
res = block_device.new_format_is_swap(bdm)
|
|
self.assertEqual(expected, res)
|
|
|
|
def test_new_format_is_ephemeral(self):
|
|
expected_results = [False, True, False, False, False]
|
|
for expected, bdm in zip(expected_results, self.new_mapping):
|
|
res = block_device.new_format_is_ephemeral(bdm)
|
|
self.assertEqual(expected, res)
|
|
|
|
def test_validate_device_name(self):
|
|
for value in [' ', 10, None, 'a' * 260]:
|
|
self.assertRaises(exception.InvalidBDMFormat,
|
|
block_device.validate_device_name,
|
|
value)
|
|
|
|
def test_validate_and_default_volume_size(self):
|
|
bdm = {}
|
|
for value in [-1, 'a', 2.5]:
|
|
bdm['volume_size'] = value
|
|
self.assertRaises(exception.InvalidBDMFormat,
|
|
block_device.validate_and_default_volume_size,
|
|
bdm)
|
|
|
|
def test_get_bdms_to_connect(self):
|
|
root_bdm = {'device_name': 'vda', 'boot_index': 0}
|
|
bdms = [root_bdm,
|
|
{'device_name': 'vdb', 'boot_index': 1},
|
|
{'device_name': 'vdc', 'boot_index': -1},
|
|
{'device_name': 'vde', 'boot_index': None},
|
|
{'device_name': 'vdd'}]
|
|
self.assertNotIn(root_bdm, block_device.get_bdms_to_connect(bdms,
|
|
exclude_root_mapping=True))
|
|
self.assertIn(root_bdm, block_device.get_bdms_to_connect(bdms))
|
|
|
|
|
|
class TestBlockDeviceDict(test.NoDBTestCase):
|
|
def setUp(self):
|
|
super(TestBlockDeviceDict, self).setUp()
|
|
|
|
BDM = block_device.BlockDeviceDict
|
|
|
|
self.api_mapping = [
|
|
{'id': 1, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sdb1',
|
|
'source_type': 'blank',
|
|
'destination_type': 'local',
|
|
'delete_on_termination': True,
|
|
'guest_format': 'swap',
|
|
'boot_index': -1},
|
|
{'id': 2, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sdc1',
|
|
'source_type': 'blank',
|
|
'destination_type': 'local',
|
|
'delete_on_termination': True,
|
|
'boot_index': -1},
|
|
{'id': 3, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sda1',
|
|
'source_type': 'volume',
|
|
'destination_type': 'volume',
|
|
'uuid': 'fake-volume-id-1',
|
|
'boot_index': 0},
|
|
{'id': 4, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sda2',
|
|
'source_type': 'snapshot',
|
|
'destination_type': 'volume',
|
|
'uuid': 'fake-snapshot-id-1',
|
|
'boot_index': -1},
|
|
{'id': 5, 'instance_uuid': uuids.instance,
|
|
'no_device': True,
|
|
'device_name': '/dev/vdc'},
|
|
]
|
|
|
|
self.new_mapping = [
|
|
BDM({'id': 1, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sdb1',
|
|
'source_type': 'blank',
|
|
'destination_type': 'local',
|
|
'delete_on_termination': True,
|
|
'guest_format': 'swap',
|
|
'boot_index': -1}),
|
|
BDM({'id': 2, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sdc1',
|
|
'source_type': 'blank',
|
|
'destination_type': 'local',
|
|
'delete_on_termination': True,
|
|
'boot_index': -1}),
|
|
BDM({'id': 3, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sda1',
|
|
'source_type': 'volume',
|
|
'destination_type': 'volume',
|
|
'volume_id': 'fake-volume-id-1',
|
|
'connection_info': "{'fake': 'connection_info'}",
|
|
'boot_index': 0}),
|
|
BDM({'id': 4, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sda2',
|
|
'source_type': 'snapshot',
|
|
'destination_type': 'volume',
|
|
'connection_info': "{'fake': 'connection_info'}",
|
|
'snapshot_id': 'fake-snapshot-id-1',
|
|
'volume_id': 'fake-volume-id-2',
|
|
'boot_index': -1}),
|
|
BDM({'id': 5, 'instance_uuid': uuids.instance,
|
|
'no_device': True,
|
|
'device_name': '/dev/vdc'}),
|
|
]
|
|
|
|
self.legacy_mapping = [
|
|
{'id': 1, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sdb1',
|
|
'delete_on_termination': True,
|
|
'virtual_name': 'swap'},
|
|
{'id': 2, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sdc1',
|
|
'delete_on_termination': True,
|
|
'virtual_name': 'ephemeral0'},
|
|
{'id': 3, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sda1',
|
|
'volume_id': 'fake-volume-id-1',
|
|
'connection_info': "{'fake': 'connection_info'}"},
|
|
{'id': 4, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sda2',
|
|
'connection_info': "{'fake': 'connection_info'}",
|
|
'snapshot_id': 'fake-snapshot-id-1',
|
|
'volume_id': 'fake-volume-id-2'},
|
|
{'id': 5, 'instance_uuid': uuids.instance,
|
|
'no_device': True,
|
|
'device_name': '/dev/vdc'},
|
|
]
|
|
|
|
self.new_mapping_source_image = [
|
|
BDM({'id': 6, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sda3',
|
|
'source_type': 'image',
|
|
'destination_type': 'volume',
|
|
'connection_info': "{'fake': 'connection_info'}",
|
|
'volume_id': 'fake-volume-id-3',
|
|
'boot_index': -1}),
|
|
BDM({'id': 7, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sda4',
|
|
'source_type': 'image',
|
|
'destination_type': 'local',
|
|
'connection_info': "{'fake': 'connection_info'}",
|
|
'image_id': 'fake-image-id-2',
|
|
'boot_index': -1}),
|
|
]
|
|
|
|
self.legacy_mapping_source_image = [
|
|
{'id': 6, 'instance_uuid': uuids.instance,
|
|
'device_name': '/dev/sda3',
|
|
'connection_info': "{'fake': 'connection_info'}",
|
|
'volume_id': 'fake-volume-id-3'},
|
|
]
|
|
|
|
def test_init(self):
|
|
def fake_validate(obj, dct):
|
|
pass
|
|
|
|
self.stub_out('nova.block_device.BlockDeviceDict._fields',
|
|
set(['field1', 'field2']))
|
|
self.stub_out('nova.block_device.BlockDeviceDict._db_only_fields',
|
|
set(['db_field1', 'db_field2']))
|
|
self.stub_out('nova.block_device.BlockDeviceDict._validate',
|
|
fake_validate)
|
|
|
|
# Make sure db fields are not picked up if they are not
|
|
# in the original dict
|
|
dev_dict = block_device.BlockDeviceDict({'field1': 'foo',
|
|
'field2': 'bar',
|
|
'db_field1': 'baz'})
|
|
self.assertIn('field1', dev_dict)
|
|
self.assertIn('field2', dev_dict)
|
|
self.assertIn('db_field1', dev_dict)
|
|
self.assertNotIn('db_field2', dev_dict)
|
|
|
|
# Make sure all expected fields are defaulted
|
|
dev_dict = block_device.BlockDeviceDict({'field1': 'foo'})
|
|
self.assertIn('field1', dev_dict)
|
|
self.assertIn('field2', dev_dict)
|
|
self.assertIsNone(dev_dict['field2'])
|
|
self.assertNotIn('db_field1', dev_dict)
|
|
self.assertNotIn('db_field2', dev_dict)
|
|
|
|
# Unless they are not meant to be
|
|
dev_dict = block_device.BlockDeviceDict({'field1': 'foo'},
|
|
do_not_default=set(['field2']))
|
|
self.assertIn('field1', dev_dict)
|
|
self.assertNotIn('field2', dev_dict)
|
|
self.assertNotIn('db_field1', dev_dict)
|
|
self.assertNotIn('db_field2', dev_dict)
|
|
|
|
# Passing kwargs to constructor works
|
|
dev_dict = block_device.BlockDeviceDict(field1='foo')
|
|
self.assertIn('field1', dev_dict)
|
|
self.assertIn('field2', dev_dict)
|
|
self.assertIsNone(dev_dict['field2'])
|
|
dev_dict = block_device.BlockDeviceDict(
|
|
{'field1': 'foo'}, field2='bar')
|
|
self.assertEqual('foo', dev_dict['field1'])
|
|
self.assertEqual('bar', dev_dict['field2'])
|
|
|
|
def test_init_prepend_dev_to_device_name(self):
|
|
bdm = {'id': 3, 'instance_uuid': uuids.instance,
|
|
'device_name': 'vda',
|
|
'source_type': 'volume',
|
|
'destination_type': 'volume',
|
|
'volume_id': 'fake-volume-id-1',
|
|
'boot_index': 0}
|
|
bdm_dict = block_device.BlockDeviceDict(bdm)
|
|
self.assertEqual('/dev/vda', bdm_dict['device_name'])
|
|
|
|
bdm['device_name'] = '/dev/vdb'
|
|
bdm_dict = block_device.BlockDeviceDict(bdm)
|
|
self.assertEqual('/dev/vdb', bdm_dict['device_name'])
|
|
|
|
bdm['device_name'] = None
|
|
bdm_dict = block_device.BlockDeviceDict(bdm)
|
|
self.assertIsNone(bdm_dict['device_name'])
|
|
|
|
def test_init_boolify_delete_on_termination(self):
|
|
# Make sure that when delete_on_termination is not passed it's
|
|
# still set to False and not None
|
|
bdm = {'id': 3, 'instance_uuid': uuids.instance,
|
|
'device_name': 'vda',
|
|
'source_type': 'volume',
|
|
'destination_type': 'volume',
|
|
'volume_id': 'fake-volume-id-1',
|
|
'boot_index': 0}
|
|
bdm_dict = block_device.BlockDeviceDict(bdm)
|
|
self.assertFalse(bdm_dict['delete_on_termination'])
|
|
|
|
def test_validate(self):
|
|
self.assertRaises(exception.InvalidBDMFormat,
|
|
block_device.BlockDeviceDict,
|
|
{'bogus_field': 'lame_val'})
|
|
|
|
lame_bdm = dict(self.new_mapping[2])
|
|
del lame_bdm['source_type']
|
|
self.assertRaises(exception.InvalidBDMFormat,
|
|
block_device.BlockDeviceDict,
|
|
lame_bdm)
|
|
|
|
lame_bdm['no_device'] = True
|
|
block_device.BlockDeviceDict(lame_bdm)
|
|
|
|
lame_dev_bdm = dict(self.new_mapping[2])
|
|
lame_dev_bdm['device_name'] = "not a valid name"
|
|
self.assertRaises(exception.InvalidBDMFormat,
|
|
block_device.BlockDeviceDict,
|
|
lame_dev_bdm)
|
|
|
|
lame_dev_bdm['device_name'] = ""
|
|
self.assertRaises(exception.InvalidBDMFormat,
|
|
block_device.BlockDeviceDict,
|
|
lame_dev_bdm)
|
|
|
|
cool_volume_size_bdm = dict(self.new_mapping[2])
|
|
cool_volume_size_bdm['volume_size'] = '42'
|
|
cool_volume_size_bdm = block_device.BlockDeviceDict(
|
|
cool_volume_size_bdm)
|
|
self.assertEqual(42, cool_volume_size_bdm['volume_size'])
|
|
|
|
lame_volume_size_bdm = dict(self.new_mapping[2])
|
|
lame_volume_size_bdm['volume_size'] = 'some_non_int_string'
|
|
self.assertRaises(exception.InvalidBDMFormat,
|
|
block_device.BlockDeviceDict,
|
|
lame_volume_size_bdm)
|
|
|
|
truthy_bdm = dict(self.new_mapping[2])
|
|
truthy_bdm['delete_on_termination'] = '1'
|
|
truthy_bdm = block_device.BlockDeviceDict(truthy_bdm)
|
|
self.assertTrue(truthy_bdm['delete_on_termination'])
|
|
|
|
verbose_bdm = dict(self.new_mapping[2])
|
|
verbose_bdm['boot_index'] = 'first'
|
|
self.assertRaises(exception.InvalidBDMFormat,
|
|
block_device.BlockDeviceDict,
|
|
verbose_bdm)
|
|
|
|
def test_from_legacy(self):
|
|
for legacy, new in zip(self.legacy_mapping, self.new_mapping):
|
|
self.assertThat(
|
|
block_device.BlockDeviceDict.from_legacy(legacy),
|
|
matchers.IsSubDictOf(new))
|
|
|
|
def test_from_legacy_mapping(self):
|
|
def _get_image_bdms(bdms):
|
|
return [bdm for bdm in bdms if bdm['source_type'] == 'image']
|
|
|
|
def _get_bootable_bdms(bdms):
|
|
return [bdm for bdm in bdms
|
|
if (bdm['boot_index'] is not None and
|
|
bdm['boot_index'] >= 0)]
|
|
|
|
new_no_img = block_device.from_legacy_mapping(self.legacy_mapping)
|
|
self.assertEqual(0, len(_get_image_bdms(new_no_img)))
|
|
|
|
for new, expected in zip(new_no_img, self.new_mapping):
|
|
self.assertThat(new, matchers.IsSubDictOf(expected))
|
|
|
|
new_with_img = block_device.from_legacy_mapping(
|
|
self.legacy_mapping, 'fake_image_ref')
|
|
image_bdms = _get_image_bdms(new_with_img)
|
|
boot_bdms = _get_bootable_bdms(new_with_img)
|
|
self.assertEqual(1, len(image_bdms))
|
|
self.assertEqual(1, len(boot_bdms))
|
|
self.assertEqual(0, image_bdms[0]['boot_index'])
|
|
self.assertEqual('image', boot_bdms[0]['source_type'])
|
|
|
|
new_with_img_and_root = block_device.from_legacy_mapping(
|
|
self.legacy_mapping, 'fake_image_ref', 'sda1')
|
|
image_bdms = _get_image_bdms(new_with_img_and_root)
|
|
boot_bdms = _get_bootable_bdms(new_with_img_and_root)
|
|
self.assertEqual(0, len(image_bdms))
|
|
self.assertEqual(1, len(boot_bdms))
|
|
self.assertEqual(0, boot_bdms[0]['boot_index'])
|
|
self.assertEqual('volume', boot_bdms[0]['source_type'])
|
|
|
|
new_no_root = block_device.from_legacy_mapping(
|
|
self.legacy_mapping, 'fake_image_ref', 'sda1', no_root=True)
|
|
self.assertEqual(0, len(_get_image_bdms(new_no_root)))
|
|
self.assertEqual(0, len(_get_bootable_bdms(new_no_root)))
|
|
|
|
def test_from_api(self):
|
|
for api, new in zip(self.api_mapping, self.new_mapping):
|
|
new['connection_info'] = None
|
|
if new['snapshot_id']:
|
|
new['volume_id'] = None
|
|
self.assertThat(
|
|
block_device.BlockDeviceDict.from_api(api, False),
|
|
matchers.IsSubDictOf(new))
|
|
|
|
def test_from_api_invalid_blank_id(self):
|
|
api_dict = {'id': 1,
|
|
'source_type': 'blank',
|
|
'destination_type': 'volume',
|
|
'uuid': 'fake-volume-id-1',
|
|
'delete_on_termination': True,
|
|
'boot_index': -1}
|
|
self.assertRaises(exception.InvalidBDMFormat,
|
|
block_device.BlockDeviceDict.from_api, api_dict,
|
|
False)
|
|
|
|
def test_from_api_invalid_source_to_local_mapping(self):
|
|
api_dict = {'id': 1,
|
|
'source_type': 'image',
|
|
'destination_type': 'local',
|
|
'uuid': 'fake-volume-id-1'}
|
|
self.assertRaises(exception.InvalidBDMFormat,
|
|
block_device.BlockDeviceDict.from_api, api_dict,
|
|
False)
|
|
|
|
def test_from_api_valid_source_to_local_mapping(self):
|
|
api_dict = {'id': 1,
|
|
'source_type': 'image',
|
|
'destination_type': 'local',
|
|
'volume_id': 'fake-volume-id-1',
|
|
'uuid': 1,
|
|
'boot_index': 0}
|
|
|
|
retexp = block_device.BlockDeviceDict(
|
|
{'id': 1,
|
|
'source_type': 'image',
|
|
'image_id': 1,
|
|
'destination_type': 'local',
|
|
'volume_id': 'fake-volume-id-1',
|
|
'boot_index': 0})
|
|
self.assertEqual(retexp,
|
|
block_device.BlockDeviceDict.from_api(api_dict, True))
|
|
|
|
def test_from_api_valid_source_to_local_mapping_with_string_bi(self):
|
|
api_dict = {'id': 1,
|
|
'source_type': 'image',
|
|
'destination_type': 'local',
|
|
'volume_id': 'fake-volume-id-1',
|
|
'uuid': 1,
|
|
'boot_index': '0'}
|
|
|
|
retexp = block_device.BlockDeviceDict(
|
|
{'id': 1,
|
|
'source_type': 'image',
|
|
'image_id': 1,
|
|
'destination_type': 'local',
|
|
'volume_id': 'fake-volume-id-1',
|
|
'boot_index': 0})
|
|
self.assertEqual(retexp,
|
|
block_device.BlockDeviceDict.from_api(api_dict, True))
|
|
|
|
def test_from_api_invalid_image_to_destination_local_mapping(self):
|
|
api_dict = {'id': 1,
|
|
'source_type': 'image',
|
|
'destination_type': 'local',
|
|
'uuid': 'fake-volume-id-1',
|
|
'volume_type': 'fake-lvm-1',
|
|
'boot_index': 1}
|
|
ex = self.assertRaises(exception.InvalidBDMFormat,
|
|
block_device.BlockDeviceDict.from_api,
|
|
api_dict, False)
|
|
self.assertIn('Mapping image to local is not supported',
|
|
six.text_type(ex))
|
|
|
|
def test_from_api_invalid_volume_type_to_destination_local_mapping(self):
|
|
api_dict = {'id': 1,
|
|
'source_type': 'volume',
|
|
'destination_type': 'local',
|
|
'uuid': 'fake-volume-id-1',
|
|
'volume_type': 'fake-lvm-1'}
|
|
ex = self.assertRaises(exception.InvalidBDMFormat,
|
|
block_device.BlockDeviceDict.from_api,
|
|
api_dict, False)
|
|
self.assertIn('Specifying a volume_type with destination_type=local '
|
|
'is not supported', six.text_type(ex))
|
|
|
|
def test_from_api_invalid_specify_volume_type_with_source_volume_mapping(
|
|
self):
|
|
api_dict = {'id': 1,
|
|
'source_type': 'volume',
|
|
'destination_type': 'volume',
|
|
'uuid': 'fake-volume-id-1',
|
|
'volume_type': 'fake-lvm-1'}
|
|
ex = self.assertRaises(exception.InvalidBDMFormat,
|
|
block_device.BlockDeviceDict.from_api,
|
|
api_dict, False)
|
|
self.assertIn('Specifying volume type to existing volume is '
|
|
'not supported', six.text_type(ex))
|
|
|
|
def test_legacy(self):
|
|
for legacy, new in zip(self.legacy_mapping, self.new_mapping):
|
|
self.assertThat(
|
|
legacy,
|
|
matchers.IsSubDictOf(new.legacy()))
|
|
|
|
def test_legacy_mapping(self):
|
|
got_legacy = block_device.legacy_mapping(self.new_mapping)
|
|
|
|
for legacy, expected in zip(got_legacy, self.legacy_mapping):
|
|
self.assertThat(expected, matchers.IsSubDictOf(legacy))
|
|
|
|
def test_legacy_source_image(self):
|
|
for legacy, new in zip(self.legacy_mapping_source_image,
|
|
self.new_mapping_source_image):
|
|
if new['destination_type'] == 'volume':
|
|
self.assertThat(legacy, matchers.IsSubDictOf(new.legacy()))
|
|
else:
|
|
self.assertRaises(exception.InvalidBDMForLegacy, new.legacy)
|
|
|
|
def test_legacy_mapping_source_image(self):
|
|
got_legacy = block_device.legacy_mapping(self.new_mapping)
|
|
|
|
for legacy, expected in zip(got_legacy, self.legacy_mapping):
|
|
self.assertThat(expected, matchers.IsSubDictOf(legacy))
|
|
|
|
def test_legacy_mapping_from_object_list(self):
|
|
bdm1 = objects.BlockDeviceMapping()
|
|
bdm1 = objects.BlockDeviceMapping._from_db_object(
|
|
None, bdm1, fake_block_device.FakeDbBlockDeviceDict(
|
|
self.new_mapping[0]))
|
|
bdm2 = objects.BlockDeviceMapping()
|
|
bdm2 = objects.BlockDeviceMapping._from_db_object(
|
|
None, bdm2, fake_block_device.FakeDbBlockDeviceDict(
|
|
self.new_mapping[1]))
|
|
bdmlist = objects.BlockDeviceMappingList()
|
|
bdmlist.objects = [bdm1, bdm2]
|
|
block_device.legacy_mapping(bdmlist)
|
|
|
|
def test_image_mapping(self):
|
|
removed_fields = ['id', 'instance_uuid', 'connection_info',
|
|
'created_at', 'updated_at', 'deleted_at', 'deleted']
|
|
for bdm in self.new_mapping:
|
|
mapping_bdm = fake_block_device.FakeDbBlockDeviceDict(
|
|
bdm).get_image_mapping()
|
|
for fld in removed_fields:
|
|
self.assertNotIn(fld, mapping_bdm)
|
|
|
|
def _test_snapshot_from_bdm(self, template):
|
|
snapshot = block_device.snapshot_from_bdm('new-snapshot-id', template)
|
|
self.assertEqual('new-snapshot-id', snapshot['snapshot_id'])
|
|
self.assertEqual('snapshot', snapshot['source_type'])
|
|
self.assertEqual('volume', snapshot['destination_type'])
|
|
self.assertEqual(template.volume_size, snapshot['volume_size'])
|
|
self.assertEqual(template.delete_on_termination,
|
|
snapshot['delete_on_termination'])
|
|
self.assertEqual(template.device_name, snapshot['device_name'])
|
|
for key in ['disk_bus', 'device_type', 'boot_index']:
|
|
self.assertEqual(template[key], snapshot[key])
|
|
|
|
def test_snapshot_from_bdm(self):
|
|
for bdm in self.new_mapping:
|
|
self._test_snapshot_from_bdm(objects.BlockDeviceMapping(**bdm))
|
|
|
|
def test_snapshot_from_object(self):
|
|
for bdm in self.new_mapping[:-1]:
|
|
obj = objects.BlockDeviceMapping()
|
|
obj = objects.BlockDeviceMapping._from_db_object(
|
|
None, obj, fake_block_device.FakeDbBlockDeviceDict(
|
|
bdm))
|
|
self._test_snapshot_from_bdm(obj)
|