Add more tests for server metadata processing

Make sure that each of the things we do in get_hostvars_for_server is
checked as working. In walking through it, found a bug in image ids from
boot-from-volume servers.

Change-Id: If71d2cd114c5b5406a39dbb04c96b61fc7193399
This commit is contained in:
Monty Taylor 2015-04-26 10:47:57 -04:00 committed by greghaynes
parent 2c4d909111
commit 59848071e4
2 changed files with 111 additions and 20 deletions

View File

@ -117,8 +117,9 @@ def get_hostvars_from_server(cloud, server, mounts=None):
server_vars['flavor'].pop('links', None)
# OpenStack can return image as a string when you've booted from volume
if unicode(server.image) == server.image:
if str(server.image) == server.image:
image_id = server.image
server_vars['image'] = dict(id=image_id)
else:
image_id = server.image.get('id', None)
if image_id:

View File

@ -12,10 +12,43 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import mock
import testtools
from shade import exc
from shade import meta
PRIVATE_V4 = '198.51.100.3'
PUBLIC_V4 = '192.0.2.99'
class FakeCloud(object):
region_name = 'test-region'
name = 'test-name'
private = False
def get_flavor_name(self, id):
return 'test-flavor-name'
def get_image_name(self, id):
return 'test-image-name'
def get_volumes(self, server):
return []
class FakeServer(object):
id = 'test-id-0'
metadata = {'group': 'test-group'}
addresses = {'private': [{'OS-EXT-IPS:type': 'fixed',
'addr': PRIVATE_V4,
'version': 4}],
'public': [{'OS-EXT-IPS:type': 'floating',
'addr': PUBLIC_V4,
'version': 4}]}
flavor = {'id': '101'}
image = {'id': '471c2475-da2f-47ac-aba5-cb4aa3d546f5'}
class TestMeta(testtools.TestCase):
def test_find_nova_addresses_key_name(self):
@ -36,26 +69,11 @@ class TestMeta(testtools.TestCase):
self.assertEqual([], meta.find_nova_addresses(addrs, ext_tag='foo'))
def test_get_server_ip(self):
class Server(object):
addresses = {'private': [{'OS-EXT-IPS:type': 'fixed',
'addr': '198.51.100.3',
'version': 4}],
'public': [{'OS-EXT-IPS:type': 'floating',
'addr': '192.0.2.99',
'version': 4}]}
srv = Server()
self.assertEqual('198.51.100.3', meta.get_server_private_ip(srv))
self.assertEqual('192.0.2.99', meta.get_server_public_ip(srv))
srv = FakeServer()
self.assertEqual(PRIVATE_V4, meta.get_server_private_ip(srv))
self.assertEqual(PUBLIC_V4, meta.get_server_public_ip(srv))
def test_get_groups_from_server(self):
class Cloud(object):
region_name = 'test-region'
name = 'test-name'
class Server(object):
id = 'test-id-0'
metadata = {'group': 'test-group'}
server_vars = {'flavor': 'test-flavor',
'image': 'test-image',
'az': 'test-az'}
@ -69,7 +87,8 @@ class TestMeta(testtools.TestCase):
'test-az',
'test-region_test-az',
'test-name_test-region_test-az'],
meta.get_groups_from_server(Cloud(), Server(), server_vars))
meta.get_groups_from_server(
FakeCloud(), FakeServer(), server_vars))
def test_obj_list_to_dict(self):
"""Test conversion of a list of objects to a list of dictonaries"""
@ -83,3 +102,74 @@ class TestMeta(testtools.TestCase):
new_list = meta.obj_list_to_dict(list)
self.assertEqual(new_list[0]['value'], 0)
self.assertEqual(new_list[1]['value'], 1)
def test_basic_hostvars(self):
hostvars = meta.get_hostvars_from_server(FakeCloud(), FakeServer())
self.assertNotIn('links', hostvars)
self.assertEqual(PRIVATE_V4, hostvars['private_v4'])
self.assertEqual(PUBLIC_V4, hostvars['public_v4'])
self.assertEqual(PUBLIC_V4, hostvars['interface_ip'])
self.assertEquals(FakeCloud.region_name, hostvars['region'])
self.assertEquals(FakeCloud.name, hostvars['cloud'])
self.assertEquals("test-image-name", hostvars['image']['name'])
self.assertEquals(FakeServer.image['id'], hostvars['image']['id'])
self.assertNotIn('links', hostvars['image'])
self.assertEquals(FakeServer.flavor['id'], hostvars['flavor']['id'])
self.assertEquals("test-flavor-name", hostvars['flavor']['name'])
self.assertNotIn('links', hostvars['flavor'])
# test having volumes
# test volume exception
self.assertEquals([], hostvars['volumes'])
def test_private_interface_ip(self):
cloud = FakeCloud()
cloud.private = True
hostvars = meta.get_hostvars_from_server(cloud, FakeServer())
self.assertEqual(PRIVATE_V4, hostvars['interface_ip'])
def test_image_string(self):
server = FakeServer()
server.image = 'fake-image-id'
hostvars = meta.get_hostvars_from_server(FakeCloud(), server)
self.assertEquals('fake-image-id', hostvars['image']['id'])
def test_az(self):
server = FakeServer()
server.__dict__['OS-EXT-AZ:availability_zone'] = 'az1'
hostvars = meta.get_hostvars_from_server(FakeCloud(), server)
self.assertEquals('az1', hostvars['az'])
def test_has_volume(self):
mock_cloud = mock.MagicMock()
mock_volume = mock.MagicMock()
mock_volume.id = 'volume1'
mock_volume.status = 'available'
mock_volume.display_name = 'Volume 1 Display Name'
mock_volume.attachments = [{'device': '/dev/sda0'}]
mock_volume_dict = meta.obj_to_dict(mock_volume)
mock_cloud.get_volumes.return_value = [mock_volume_dict]
hostvars = meta.get_hostvars_from_server(mock_cloud, FakeServer())
self.assertEquals('volume1', hostvars['volumes'][0]['id'])
self.assertEquals('/dev/sda0', hostvars['volumes'][0]['device'])
def test_has_no_volume_service(self):
mock_cloud = mock.MagicMock()
def side_effect(*args):
raise exc.OpenStackCloudException("No Volumes")
mock_cloud.get_volumes.side_effect = side_effect
hostvars = meta.get_hostvars_from_server(mock_cloud, FakeServer())
self.assertEquals([], hostvars['volumes'])
def test_unknown_volume_exception(self):
mock_cloud = mock.MagicMock()
class FakeException(Exception):
pass
def side_effect(*args):
raise FakeException("No Volumes")
mock_cloud.get_volumes.side_effect = side_effect
self.assertRaises(
FakeException,
meta.get_hostvars_from_server, mock_cloud, FakeServer())