openstack/deb-heat
Code Issues Proposed changes
ec0a6720ab
deb-heat/heat/tests/test_instance_network.py
Unmesh Gurjar 2f640158f9 Added PublicDnsName verification in unit tests 
Changed duplicate validation of attribute PrivateDnsName to verify
PublicDnsName instead in unit tests.

Change-Id: If9282f61a421ee6b79234c75d60b6b2b3dbc1ee1
2014-09-01 20:42:42 +05:30

334 lines
12 KiB
Python
Raw Blame History

#
# 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 copy
import uuid
from heat.common import template_format
from heat.engine.clients.os import glance
from heat.engine.clients.os import neutron
from heat.engine.clients.os import nova
from heat.engine import environment
from heat.engine import parser
from heat.engine.resources import instance as instances
from heat.engine.resources import network_interface as network_interfaces
from heat.engine import scheduler
from heat.tests.common import HeatTestCase
from heat.tests import utils
from heat.tests.v1_1 import fakes
wp_template = '''
{
"AWSTemplateFormatVersion" : "2010-09-09",
"Description" : "WordPress",
"Parameters" : {
"KeyName" : {
"Description" : "KeyName",
"Type" : "String",
"Default" : "test"
},
"InstanceType": {
"Type": "String",
"Description": "EC2 instance type",
"Default": "m1.small",
"AllowedValues": [ "m1.small", "m1.large" ]
},
"SubnetId": {
"Type" : "String",
"Description" : "SubnetId of an existing subnet in your VPC"
},
},
"Resources" : {
"WebServer": {
"Type": "AWS::EC2::Instance",
"Properties": {
"ImageId" : "F17-x86_64-gold",
"InstanceType" : { "Ref" : "InstanceType" },
"SubnetId" : { "Ref" : "SubnetId" },
"KeyName" : { "Ref" : "KeyName" },
"UserData" : "wordpress"
}
}
}
}
'''
wp_template_with_nic = '''
{
"AWSTemplateFormatVersion" : "2010-09-09",
"Description" : "WordPress",
"Parameters" : {
"KeyName" : {
"Description" : "KeyName",
"Type" : "String",
"Default" : "test"
},
"InstanceType": {
"Type": "String",
"Description": "EC2 instance type",
"Default": "m1.small",
"AllowedValues": [ "m1.small", "m1.large" ]
},
"SubnetId": {
"Type" : "String",
"Description" : "SubnetId of an existing subnet in your VPC"
},
},
"Resources" : {
"nic1": {
"Type": "AWS::EC2::NetworkInterface",
"Properties": {
"SubnetId": { "Ref": "SubnetId" }
}
},
"WebServer": {
"Type": "AWS::EC2::Instance",
"Properties": {
"ImageId" : "F17-x86_64-gold",
"InstanceType" : { "Ref" : "InstanceType" },
"NetworkInterfaces": [ { "NetworkInterfaceId" : {"Ref": "nic1"},
"DeviceIndex" : "0" } ],
"KeyName" : { "Ref" : "KeyName" },
"UserData" : "wordpress"
}
}
}
}
'''
class FakeNeutron(object):
def show_subnet(self, subnet, **_params):
return {
'subnet': {
'name': 'name',
'network_id': 'fc68ea2c-b60b-4b4f-bd82-94ec81110766',
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'allocation_pools': [{'start': '10.10.0.2',
'end': '10.10.0.254'}],
'gateway_ip': '10.10.0.1',
'ip_version': 4,
'cidr': '10.10.0.0/24',
'id': '4156c7a5-e8c4-4aff-a6e1-8f3c7bc83861',
'enable_dhcp': False,
}}
def create_port(self, body=None):
return {
'port': {
'admin_state_up': True,
'device_id': '',
'device_owner': '',
'fixed_ips': [{
'ip_address': '10.0.3.3',
'subnet_id': '4156c7a5-e8c4-4aff-a6e1-8f3c7bc83861'}],
'id': '64d913c1-bcb1-42d2-8f0a-9593dbcaf251',
'mac_address': 'fa:16:3e:25:32:5d',
'name': '',
'network_id': 'fc68ea2c-b60b-4b4f-bd82-94ec81110766',
'status': 'ACTIVE',
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f'
}}
def delete_port(self, port_id):
return None
class instancesTest(HeatTestCase):
def setUp(self):
super(instancesTest, self).setUp()
self.fc = fakes.FakeClient()
def _mock_get_image_id_success(self, imageId_input, imageId):
self.m.StubOutWithMock(glance.GlanceClientPlugin, 'get_image_id')
glance.GlanceClientPlugin.get_image_id(
imageId_input).MultipleTimes().AndReturn(imageId)
def _test_instance_create_delete(self, vm_status='ACTIVE',
vm_delete_status='NotFound'):
return_server = self.fc.servers.list()[1]
instance = self._create_test_instance(return_server, 'in_create')
instance.resource_id = '1234'
instance.status = vm_status
# this makes sure the auto increment worked on instance creation
self.assertTrue(instance.id > 0)
expected_ip = return_server.networks['public'][0]
self.assertEqual(expected_ip, instance.FnGetAtt('PublicIp'))
self.assertEqual(expected_ip, instance.FnGetAtt('PrivateIp'))
self.assertEqual(expected_ip, instance.FnGetAtt('PrivateDnsName'))
self.assertEqual(expected_ip, instance.FnGetAtt('PublicDnsName'))
d1 = {'server': self.fc.client.get_servers_detail()[1]['servers'][0]}
d1['server']['status'] = vm_status
self.m.StubOutWithMock(self.fc.client, 'get_servers_1234')
get = self.fc.client.get_servers_1234
get().AndReturn((200, d1))
d2 = copy.deepcopy(d1)
if vm_delete_status == 'DELETED':
d2['server']['status'] = vm_delete_status
get().AndReturn((200, d2))
else:
get().AndRaise(fakes.fake_exception())
self.m.ReplayAll()
scheduler.TaskRunner(instance.delete)()
self.assertEqual((instance.DELETE, instance.COMPLETE), instance.state)
self.m.VerifyAll()
def _create_test_instance(self, return_server, name):
stack_name = '%s_s' % name
t = template_format.parse(wp_template)
template = parser.Template(t)
kwargs = {'KeyName': 'test',
'InstanceType': 'm1.large',
'SubnetId': '4156c7a5-e8c4-4aff-a6e1-8f3c7bc83861'}
stack = parser.Stack(utils.dummy_context(), stack_name, template,
environment.Environment(kwargs),
stack_id=str(uuid.uuid4()))
image_id = 'CentOS 5.2'
t['Resources']['WebServer']['Properties']['ImageId'] = image_id
resource_defns = stack.t.resource_definitions(stack)
instance = instances.Instance('%s_name' % name,
resource_defns['WebServer'], stack)
metadata = instance.metadata_get()
self.m.StubOutWithMock(nova.NovaClientPlugin, '_create')
nova.NovaClientPlugin._create().AndReturn(self.fc)
self._mock_get_image_id_success(image_id, 1)
self.m.StubOutWithMock(instance, 'neutron')
instance.neutron().MultipleTimes().AndReturn(FakeNeutron())
self.m.StubOutWithMock(neutron.NeutronClientPlugin, '_create')
neutron.NeutronClientPlugin._create().MultipleTimes().AndReturn(
FakeNeutron())
# need to resolve the template functions
server_userdata = instance.client_plugin().build_userdata(
metadata,
instance.t['Properties']['UserData'],
'ec2-user')
self.m.StubOutWithMock(nova.NovaClientPlugin, 'build_userdata')
nova.NovaClientPlugin.build_userdata(
metadata,
instance.t['Properties']['UserData'],
'ec2-user').AndReturn(server_userdata)
self.m.StubOutWithMock(self.fc.servers, 'create')
self.fc.servers.create(
image=1, flavor=3, key_name='test',
name=utils.PhysName(stack_name, instance.name),
security_groups=None,
userdata=server_userdata, scheduler_hints=None, meta=None,
nics=[{'port-id': '64d913c1-bcb1-42d2-8f0a-9593dbcaf251'}],
availability_zone=None,
block_device_mapping=None).AndReturn(
return_server)
self.m.ReplayAll()
scheduler.TaskRunner(instance.create)()
return instance
def _create_test_instance_with_nic(self, return_server, name):
stack_name = '%s_s' % name
t = template_format.parse(wp_template_with_nic)
template = parser.Template(t)
kwargs = {'KeyName': 'test',
'InstanceType': 'm1.large',
'SubnetId': '4156c7a5-e8c4-4aff-a6e1-8f3c7bc83861'}
stack = parser.Stack(utils.dummy_context(), stack_name, template,
environment.Environment(kwargs),
stack_id=str(uuid.uuid4()))
image_id = 'CentOS 5.2'
t['Resources']['WebServer']['Properties']['ImageId'] = image_id
resource_defns = stack.t.resource_definitions(stack)
nic = network_interfaces.NetworkInterface('%s_nic' % name,
resource_defns['nic1'],
stack)
instance = instances.Instance('%s_name' % name,
resource_defns['WebServer'], stack)
metadata = instance.metadata_get()
self._mock_get_image_id_success(image_id, 1)
self.m.StubOutWithMock(nic, 'neutron')
nic.neutron().MultipleTimes().AndReturn(FakeNeutron())
self.m.StubOutWithMock(neutron.NeutronClientPlugin, '_create')
neutron.NeutronClientPlugin._create().MultipleTimes().AndReturn(
FakeNeutron())
self.m.StubOutWithMock(nova.NovaClientPlugin, '_create')
nova.NovaClientPlugin._create().AndReturn(self.fc)
# need to resolve the template functions
server_userdata = instance.client_plugin().build_userdata(
metadata,
instance.t['Properties']['UserData'],
'ec2-user')
self.m.StubOutWithMock(nova.NovaClientPlugin, 'build_userdata')
nova.NovaClientPlugin.build_userdata(
metadata,
instance.t['Properties']['UserData'],
'ec2-user').AndReturn(server_userdata)
self.m.StubOutWithMock(self.fc.servers, 'create')
self.fc.servers.create(
image=1, flavor=3, key_name='test',
name=utils.PhysName(stack_name, instance.name),
security_groups=None,
userdata=server_userdata, scheduler_hints=None, meta=None,
nics=[{'port-id': '64d913c1-bcb1-42d2-8f0a-9593dbcaf251'}],
availability_zone=None,
block_device_mapping=None).AndReturn(
return_server)
self.m.ReplayAll()
# create network interface
scheduler.TaskRunner(nic.create)()
stack.resources["nic1"] = nic
scheduler.TaskRunner(instance.create)()
return instance
def test_instance_create_delete_with_SubnetId(self):
self._test_instance_create_delete(vm_delete_status='DELETED')
def test_instance_create_with_nic(self):
return_server = self.fc.servers.list()[1]
instance = self._create_test_instance_with_nic(
return_server, 'in_create_wnic')
# this makes sure the auto increment worked on instance creation
self.assertTrue(instance.id > 0)
expected_ip = return_server.networks['public'][0]
self.assertEqual(expected_ip, instance.FnGetAtt('PublicIp'))
self.assertEqual(expected_ip, instance.FnGetAtt('PrivateIp'))
self.assertEqual(expected_ip, instance.FnGetAtt('PrivateDnsName'))
self.assertEqual(expected_ip, instance.FnGetAtt('PublicDnsName'))
self.m.VerifyAll()
View Git Blame Copy Permalink