# 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. from unittest import mock from senlin.common import exception as exc from senlin.profiles.os.nova import server from senlin.tests.unit.common import base from senlin.tests.unit.common import utils spec = { 'type': 'os.nova.server', 'version': '1.0', 'properties': { 'context': {}, 'auto_disk_config': True, 'availability_zone': 'FAKE_AZ', 'block_device_mapping': [{ 'device_name': 'FAKE_NAME', 'volume_size': 1000, }], 'flavor': 'FLAV', 'image': 'FAKE_IMAGE', 'key_name': 'FAKE_KEYNAME', "metadata": {"meta var": "meta val"}, 'name': 'FAKE_SERVER_NAME', 'networks': [{ 'floating_ip': 'FAKE_FLOATING_IP', 'floating_network': 'FAKE_FLOATING_NET', 'security_groups': ['FAKE_SECURITY_GROUP'], 'port': 'FAKE_PORT', 'vnic_type': 'direct', 'fixed_ip': 'FAKE_IP', 'network': 'FAKE_NET', }], 'scheduler_hints': { 'same_host': 'HOST_ID', }, } } class TestAvailabilityZoneValidation(base.SenlinTestCase): scenarios = [ ('validate:success', dict( reason=None, success=True, validate_result=[['FAKE_AZ']], result='FAKE_AZ', exception=None, message='')), ('validate:driver_failure', dict( reason=None, success=False, validate_result=exc.InternalError(message='BANG.'), result='FAKE_AZ', exception=exc.InternalError, message='BANG.')), ('validate:not_found', dict( reason=None, success=False, validate_result=[[]], result='FAKE_AZ', exception=exc.InvalidSpec, message=("The specified availability_zone 'FAKE_AZ' could " "not be found"))), ('create:success', dict( reason='create', success=True, validate_result=[['FAKE_AZ']], result='FAKE_AZ', exception=None, message='')), ('create:driver_failure', dict( reason='create', success=False, validate_result=exc.InternalError(message='BANG'), result='FAKE_AZ', exception=exc.EResourceCreation, message='Failed in creating server: BANG.')), ('create:not_found', dict( reason='create', success=False, validate_result=[[]], result='FAKE_AZ', exception=exc.EResourceCreation, message=("Failed in creating server: The specified " "availability_zone 'FAKE_AZ' could not be found."))) ] def setUp(self): super(TestAvailabilityZoneValidation, self).setUp() self.cc = mock.Mock() prof = server.ServerProfile('t', spec) prof._computeclient = self.cc self.profile = prof def test_validation(self): self.cc.validate_azs.side_effect = self.validate_result node = mock.Mock(id='NODE_ID') if self.success: res = self.profile._validate_az(node, 'FAKE_AZ', self.reason) self.assertEqual(self.result, res) else: ex = self.assertRaises(self.exception, self.profile._validate_az, node, 'FAKE_AZ', self.reason) self.assertEqual(self.message, str(ex)) self.cc.validate_azs.assert_called_once_with(['FAKE_AZ']) class TestFlavorValidation(base.SenlinTestCase): scenarios = [ ('validate:success', dict( reason=None, success=True, validate_result=[mock.Mock(id='FID', is_disabled=False)], result='FID', exception=None, message='')), ('validate:driver_failure', dict( reason=None, success=False, validate_result=exc.InternalError(message='BANG.'), result='FID', exception=exc.InternalError, message='BANG.')), ('validate:not_found', dict( reason=None, success=False, validate_result=exc.InternalError(code=404, message='BANG.'), result='FID', exception=exc.InvalidSpec, message="The specified flavor 'FLAVOR' could not be found.")), ('validate:disabled', dict( reason=None, success=False, validate_result=[mock.Mock(id='FID', is_disabled=True)], result='FID', exception=exc.InvalidSpec, message="The specified flavor 'FLAVOR' is disabled")), ('create:success', dict( reason='create', success=True, validate_result=[mock.Mock(id='FID', is_disabled=False)], result='FID', exception=None, message='')), ('create:driver_failure', dict( reason='create', success=False, validate_result=exc.InternalError(message='BANG'), result='FID', exception=exc.EResourceCreation, message='Failed in creating server: BANG.')), ('create:not_found', dict( reason='create', success=False, validate_result=exc.InternalError(code=404, message='BANG'), result='FID', exception=exc.EResourceCreation, message="Failed in creating server: BANG.")), ('create:disabled', dict( reason='create', success=False, validate_result=[mock.Mock(id='FID', is_disabled=True)], result='FID', exception=exc.EResourceCreation, message=("Failed in creating server: The specified flavor " "'FLAVOR' is disabled."))), ('update:success', dict( reason='update', success=True, validate_result=[mock.Mock(id='FID', is_disabled=False)], result='FID', exception=None, message='')), ('update:driver_failure', dict( reason='update', success=False, validate_result=exc.InternalError(message='BANG'), result='FID', exception=exc.EResourceUpdate, message="Failed in updating server 'NOVA_ID': BANG.")), ('update:not_found', dict( reason='update', success=False, validate_result=exc.InternalError(code=404, message='BANG'), result='FID', exception=exc.EResourceUpdate, message="Failed in updating server 'NOVA_ID': BANG.")), ('update:disabled', dict( reason='update', success=False, validate_result=[mock.Mock(id='FID', is_disabled=True)], result='FID', exception=exc.EResourceUpdate, message=("Failed in updating server 'NOVA_ID': The specified " "flavor 'FLAVOR' is disabled."))) ] def setUp(self): super(TestFlavorValidation, self).setUp() self.cc = mock.Mock() self.profile = server.ServerProfile('t', spec) self.profile._computeclient = self.cc def test_validation(self): self.cc.flavor_find.side_effect = self.validate_result node = mock.Mock(id='NODE_ID', physical_id='NOVA_ID') flavor = 'FLAVOR' if self.success: res = self.profile._validate_flavor(node, flavor, self.reason) self.assertIsNotNone(res) self.assertEqual(self.result, res.id) else: ex = self.assertRaises(self.exception, self.profile._validate_flavor, node, flavor, self.reason) self.assertEqual(self.message, str(ex)) self.cc.flavor_find.assert_called_once_with(flavor, False) class TestImageValidation(base.SenlinTestCase): scenarios = [ ('validate:success', dict( reason=None, success=True, validate_result=[mock.Mock(id='IMAGE_ID')], result='IMAGE_ID', exception=None, message='')), ('validate:driver_failure', dict( reason=None, success=False, validate_result=exc.InternalError(message='BANG.'), result='FID', exception=exc.InternalError, message='BANG.')), ('validate:not_found', dict( reason=None, success=False, validate_result=exc.InternalError(code=404, message='BANG.'), result='FID', exception=exc.InvalidSpec, message="The specified image 'IMAGE' could not be found.")), ('create:success', dict( reason='create', success=True, validate_result=[mock.Mock(id='IMAGE_ID')], result='IMAGE_ID', exception=None, message='')), ('create:driver_failure', dict( reason='create', success=False, validate_result=exc.InternalError(message='BANG'), result='FID', exception=exc.EResourceCreation, message='Failed in creating server: BANG.')), ('create:not_found', dict( reason='create', success=False, validate_result=exc.InternalError(code=404, message='BANG'), result='FID', exception=exc.EResourceCreation, message="Failed in creating server: BANG.")), ('update:success', dict( reason='update', success=True, validate_result=[mock.Mock(id='IMAGE_ID')], result='IMAGE_ID', exception=None, message='')), ('update:driver_failure', dict( reason='update', success=False, validate_result=exc.InternalError(message='BANG'), result='FID', exception=exc.EResourceUpdate, message="Failed in updating server 'NOVA_ID': BANG.")), ('update:not_found', dict( reason='update', success=False, validate_result=exc.InternalError(code=404, message='BANG'), result='FID', exception=exc.EResourceUpdate, message="Failed in updating server 'NOVA_ID': BANG.")), ] def setUp(self): super(TestImageValidation, self).setUp() self.cc = mock.Mock() self.gc = mock.Mock() self.profile = server.ServerProfile('t', spec) self.profile._computeclient = self.cc self.profile._glanceclient = self.gc def test_validation(self): self.gc.image_find.side_effect = self.validate_result node = mock.Mock(id='NODE_ID', physical_id='NOVA_ID') image = 'IMAGE' if self.success: res = self.profile._validate_image(node, image, self.reason) self.assertIsNotNone(res) self.assertEqual(self.result, res.id) else: ex = self.assertRaises(self.exception, self.profile._validate_image, node, image, self.reason) self.assertEqual(self.message, str(ex)) self.gc.image_find.assert_called_once_with(image, False) class TestVolumeValidation(base.SenlinTestCase): scenarios = [ ('validate:success', dict( reason=None, success=True, validate_result=[mock.Mock(id='VOLUME_ID', status='available')], result='VOLUME_ID', exception=None, message='')), ('validate:failure', dict( reason=None, success=False, validate_result=[mock.Mock(id='VOLUME_ID', status='in-use')], result='VOLUME_ID', exception=exc.InvalidSpec, message="The volume VOLUME should be in 'available' " "status but is in 'in-use' status.")), ('validate:driver_failure', dict( reason=None, success=False, validate_result=exc.InternalError(message='BANG.'), result='FID', exception=exc.InternalError, message='BANG.')), ('validate:not_found', dict( reason=None, success=False, validate_result=exc.InternalError(code=404, message='BANG.'), result='FID', exception=exc.InvalidSpec, message="The specified volume 'VOLUME' could not be found.")), ('create:success', dict( reason='create', success=True, validate_result=[mock.Mock(id='VOLUME_ID', status='available')], result='VOLUME_ID', exception=None, message='')), ('create:driver_failure', dict( reason='create', success=False, validate_result=exc.InternalError(message='BANG'), result='FID', exception=exc.EResourceCreation, message='Failed in creating server: BANG.')), ('create:not_found', dict( reason='create', success=False, validate_result=exc.InternalError(code=404, message='BANG'), result='FID', exception=exc.EResourceCreation, message="Failed in creating server: BANG.")), ] def setUp(self): super(TestVolumeValidation, self).setUp() bdm_v2 = [ { 'volume_size': 1, 'uuid': '6ce0be68', 'source_type': 'volume', 'destination_type': 'volume', 'boot_index': 0, }, ] volume_spec = { 'type': 'os.nova.server', 'version': '1.0', 'properties': { 'flavor': 'FLAV', 'name': 'FAKE_SERVER_NAME', 'security_groups': ['HIGH_SECURITY_GROUP'], 'block_device_mapping_v2': bdm_v2, } } self.vc = mock.Mock() self.profile = server.ServerProfile('t', volume_spec) self.profile._block_storageclient = self.vc def test_validation(self): self.vc.volume_get.side_effect = self.validate_result node = mock.Mock(id='NODE_ID', physical_id='NOVA_ID') volume = 'VOLUME' if self.success: res = self.profile._validate_volume(node, volume, self.reason) self.assertIsNotNone(res) self.assertEqual(self.result, res.id) else: ex = self.assertRaises(self.exception, self.profile._validate_volume, node, volume, self.reason) self.assertEqual(self.message, str(ex)) self.vc.volume_get.assert_called_once_with(volume) class TestKeypairValidation(base.SenlinTestCase): scenarios = [ ('validate:success', dict( reason=None, success=True, validate_result=[mock.Mock(id='KEY_ID')], result='KEY_ID', exception=None, message='')), ('validate:driver_failure', dict( reason=None, success=False, validate_result=exc.InternalError(message='BANG.'), result='FID', exception=exc.InternalError, message='BANG.')), ('validate:not_found', dict( reason=None, success=False, validate_result=exc.InternalError(code=404, message='BANG.'), result='FID', exception=exc.InvalidSpec, message="The specified key_name 'KEY' could not be found.")), ('create:success', dict( reason='create', success=True, validate_result=[mock.Mock(id='IMAGE_ID')], result='IMAGE_ID', exception=None, message='')), ('create:driver_failure', dict( reason='create', success=False, validate_result=exc.InternalError(message='BANG'), result='FID', exception=exc.EResourceCreation, message='Failed in creating server: BANG.')), ('create:not_found', dict( reason='create', success=False, validate_result=exc.InternalError(code=404, message='BANG'), result='FID', exception=exc.EResourceCreation, message="Failed in creating server: BANG.")), ('update:success', dict( reason='update', success=True, validate_result=[mock.Mock(id='KEY_ID')], result='KEY_ID', exception=None, message='')), ('update:driver_failure', dict( reason='update', success=False, validate_result=exc.InternalError(message='BANG'), result='FID', exception=exc.EResourceUpdate, message="Failed in updating server 'NOVA_ID': BANG.")), ('update:not_found', dict( reason='update', success=False, validate_result=exc.InternalError(code=404, message='BANG'), result='FID', exception=exc.EResourceUpdate, message="Failed in updating server 'NOVA_ID': BANG.")), ] def setUp(self): super(TestKeypairValidation, self).setUp() self.cc = mock.Mock() self.profile = server.ServerProfile('t', spec) self.profile._computeclient = self.cc def test_validation(self): self.cc.keypair_find.side_effect = self.validate_result node = mock.Mock(id='NODE_ID', physical_id='NOVA_ID') key = 'KEY' if self.success: res = self.profile._validate_keypair(node, key, self.reason) self.assertIsNotNone(res) self.assertEqual(self.result, res.id) else: ex = self.assertRaises(self.exception, self.profile._validate_keypair, node, key, self.reason) self.assertEqual(self.message, str(ex)) self.cc.keypair_find.assert_called_once_with(key, False) class TestNetworkValidation(base.SenlinTestCase): scenarios = [ ('validate:net-n:port-n:fixed_ip-n:sgroups-n', dict( reason=None, success=True, inputs={'port': 'PORT'}, net_result=[], port_result=[mock.Mock(id='PORT_ID', status='DOWN')], sg_result=[], floating_result=[], result={'port': 'PORT_ID'}, exception=None, message='')), ('validate:net-y:port-n:fixed_ip-n:sgroups-y', dict( reason=None, success=True, inputs={'network': 'NET', 'security_groups': ['default']}, net_result=[mock.Mock(id='NET_ID')], port_result=[], sg_result=[mock.Mock(id='SG_ID')], floating_result=[], result={'network': 'NET_ID', 'security_groups': ['SG_ID']}, exception=None, message='')), ('validate:net-y:port-n:fixed_ip-n:sgroups-n:floating_net-y', dict( reason=None, success=True, inputs={'network': 'NET', 'floating_network': 'NET'}, net_result=[mock.Mock(id='NET_ID'), mock.Mock(id='NET_ID')], port_result=[], sg_result=[], floating_result=[], result={'network': 'NET_ID', 'floating_network': 'NET_ID'}, exception=None, message='')), ('validate:net-y:port-n:fixed_ip-n:floating_net-y:floating_ip-y', dict( reason=None, success=True, inputs={'network': 'NET', 'floating_network': 'NET', 'floating_ip': 'FLOATINGIP'}, net_result=[mock.Mock(id='NET_ID'), mock.Mock(id='NET_ID')], port_result=[], sg_result=[], floating_result=[mock.Mock(id='FLOATINGIP_ID', status='INACTIVE')], result={'network': 'NET_ID', 'floating_network': 'NET_ID', 'floating_ip_id': 'FLOATINGIP_ID', 'floating_ip': 'FLOATINGIP'}, exception=None, message='')), ('validate:net-y:port-n:fixed_ip-y:sgroups-n', dict( reason=None, success=True, inputs={'network': 'NET', 'fixed_ip': 'FIXED_IP'}, net_result=[mock.Mock(id='NET_ID')], port_result=[], sg_result=[], floating_result=[], result={'network': 'NET_ID', 'fixed_ip': 'FIXED_IP'}, exception=None, message='')), ('validate:net-f:port-y:fixed_ip-n:sgroups-n', dict( reason=None, success=False, inputs={'network': 'NET', 'port': 'PORT'}, net_result=[exc.InternalError(message='NET Failure')], port_result=[], sg_result=[], floating_result=[], result={}, exception=exc.InvalidSpec, message='NET Failure')), ('validate:net-n:port-f:fixed_ip-n', dict( reason=None, success=False, inputs={'port': 'PORT'}, net_result=[], port_result=[exc.InternalError(message='PORT Failure')], sg_result=[], floating_result=[], result={}, exception=exc.InvalidSpec, message='PORT Failure')), ('validate:net-n:port-active:fixed_ip-n', dict( reason=None, success=False, inputs={'port': 'PORT'}, net_result=[], port_result=[mock.Mock(id='PORT_ID', status='ACTIVE')], sg_result=[], floating_result=[], result={}, exception=exc.InvalidSpec, message='The status of the port PORT must be DOWN')), ('validate:net-n:port-y:fixed_ip-n:floating_net-n:floating_ip-y', dict( reason=None, success=False, inputs={'port': 'PORT', 'floating_ip': 'FLOATINGIP'}, net_result=[], port_result=[mock.Mock(id='PORT_ID', status='DOWN')], sg_result=[], floating_result=[mock.Mock(id='FLOATINGIP_ID', status='INACTIVE')], result={}, exception=exc.InvalidSpec, message='Must specify a network to create floating IP')), ('validate:net-n:port-y:fixed_ip-n:floating_ip-active', dict( reason=None, success=False, inputs={'port': 'PORT', 'floating_network': 'NET', 'floating_ip': 'FLOATINGIP'}, net_result=[mock.Mock(id='NET_ID')], port_result=[mock.Mock(id='PORT_ID', status='DOWN')], sg_result=[], floating_result=[mock.Mock(id='FLOATINGIP_ID', status='ACTIVE')], result={}, exception=exc.InvalidSpec, message='the floating IP FLOATINGIP has been used.')), ('validate:net-n:port-n:fixed_ip-n', dict( reason=None, success=False, inputs={'fixed_ip': 'FIXED_IP'}, net_result=[], port_result=[], sg_result=[], floating_result=[], result={}, exception=exc.InvalidSpec, message="One of 'port' and 'network' must be provided")), ('validate:net-n:port-y:fixed_ip-y', dict( reason=None, success=False, inputs={'port': 'PORT', 'fixed_ip': 'FIXED_IP'}, net_result=[], port_result=[mock.Mock(id='PORT_ID', status='DOWN')], sg_result=[], floating_result=[], result={}, exception=exc.InvalidSpec, message=("The 'port' property and the 'fixed_ip' property cannot " "be specified at the same time"))), ('create:net-y:port-y:fixed_ip-n', dict( reason='create', success=True, inputs={'network': 'NET', 'port': 'PORT'}, net_result=[mock.Mock(id='NET_ID')], port_result=[mock.Mock(id='PORT_ID', status='DOWN')], sg_result=[], floating_result=[], result={'network': 'NET_ID', 'port': 'PORT_ID'}, exception=None, message='')), ('create:net-y:port-n:fixed_ip-y', dict( reason='create', success=True, inputs={'network': 'NET', 'fixed_ip': 'FIXED_IP'}, net_result=[mock.Mock(id='NET_ID')], port_result=[], sg_result=[], floating_result=[], result={'network': 'NET_ID', 'fixed_ip': 'FIXED_IP'}, exception=None, message='')), ('create:net-y:port-n:fixed_ip-n:sgroups-y', dict( reason='create', success=True, inputs={'network': 'NET', 'security_groups': ['default']}, net_result=[mock.Mock(id='NET_ID')], port_result=[], sg_result=[mock.Mock(id='SG_ID')], floating_result=[], result={'network': 'NET_ID', 'security_groups': ['SG_ID']}, exception=None, message='')), ('create:net-y:port-n:fixed_ip-n:sgroups-n:floating_net-y', dict( reason=None, success=True, inputs={'network': 'NET', 'floating_network': 'NET'}, net_result=[mock.Mock(id='NET_ID'), mock.Mock(id='NET_ID')], port_result=[], sg_result=[], floating_result=[], result={'network': 'NET_ID', 'floating_network': 'NET_ID'}, exception=None, message='')), ('create:net-f:port-y:fixed_ip-n', dict( reason='create', success=False, inputs={'network': 'NET', 'port': 'PORT'}, net_result=[exc.InternalError(message='NET Failure')], port_result=[], sg_result=[], floating_result=[], result={}, exception=exc.EResourceCreation, message='Failed in creating server: NET Failure.')), ('create:net-n:port-f:fixed_ip-n', dict( reason='create', success=False, inputs={'port': 'PORT'}, net_result=[], port_result=[exc.InternalError(message='PORT Failure')], sg_result=[], floating_result=[], result={}, exception=exc.EResourceCreation, message='Failed in creating server: PORT Failure.')), ('create:net-n:port-active:fixed_ip-n', dict( reason='create', success=False, inputs={'port': 'PORT'}, net_result=[], port_result=[mock.Mock(id='PORT_ID', status='ACTIVE')], sg_result=[], floating_result=[], result={}, exception=exc.EResourceCreation, message=('Failed in creating server: The status of the port PORT ' 'must be DOWN.'))), ('create:net-n:port-n:fixed_ip-n', dict( reason='create', success=False, inputs={'fixed_ip': 'FIXED_IP'}, net_result=[], port_result=[], sg_result=[], floating_result=[], result={}, exception=exc.EResourceCreation, message=("Failed in creating server: One of 'port' " "and 'network' must be provided."))), ('create:net-n:port-y:fixed_ip-y', dict( reason='create', success=False, inputs={'port': 'PORT', 'fixed_ip': 'FIXED_IP'}, net_result=[], port_result=[mock.Mock(id='PORT_ID', status='DOWN')], sg_result=[], floating_result=[], result={}, exception=exc.EResourceCreation, message=("Failed in creating server: The 'port' property and the " "'fixed_ip' property cannot be specified at the same " "time."))), ('update:net-y:port-y:fixed_ip-n', dict( reason='update', success=True, inputs={'network': 'NET', 'port': 'PORT'}, net_result=[mock.Mock(id='NET_ID')], port_result=[mock.Mock(id='PORT_ID', status='DOWN')], sg_result=[], floating_result=[], result={'network': 'NET_ID', 'port': 'PORT_ID'}, exception=None, message='')), ('update:net-y:port-n:fixed_ip-y', dict( reason='update', success=True, inputs={'network': 'NET', 'fixed_ip': 'FIXED_IP'}, net_result=[mock.Mock(id='NET_ID')], port_result=[], sg_result=[], floating_result=[], result={'network': 'NET_ID', 'fixed_ip': 'FIXED_IP'}, exception=None, message='')), ('update:net-y:port-n:fixed_ip-n:sgroups-y', dict( reason='create', success=True, inputs={'network': 'NET', 'security_groups': ['default']}, net_result=[mock.Mock(id='NET_ID')], port_result=[], sg_result=[mock.Mock(id='SG_ID')], floating_result=[], result={'network': 'NET_ID', 'security_groups': ['SG_ID']}, exception=None, message='')), ('update:net-y:port-n:fixed_ip-n:sgroups-n:floating_net-y', dict( reason=None, success=True, inputs={'network': 'NET', 'floating_network': 'NET'}, net_result=[mock.Mock(id='NET_ID'), mock.Mock(id='NET_ID')], port_result=[], sg_result=[], floating_result=[], result={'network': 'NET_ID', 'floating_network': 'NET_ID'}, exception=None, message='')), ('update:net-f:port-y:fixed_ip-n', dict( reason='update', success=False, inputs={'network': 'NET', 'port': 'PORT'}, net_result=[exc.InternalError(message='NET Failure')], port_result=[], sg_result=[], floating_result=[], result={}, exception=exc.EResourceUpdate, message="Failed in updating server 'NOVA_ID': NET Failure.")), ('update:net-n:port-f:fixed_ip-n', dict( reason='update', success=False, inputs={'port': 'PORT'}, net_result=[], port_result=[exc.InternalError(message='PORT Failure')], sg_result=[], floating_result=[], result={}, exception=exc.EResourceUpdate, message="Failed in updating server 'NOVA_ID': PORT Failure.")), ('update:net-n:port-active:fixed_ip-n', dict( reason='update', success=False, inputs={'port': 'PORT'}, net_result=[], port_result=[mock.Mock(id='PORT_ID', status='ACTIVE')], sg_result=[], floating_result=[], result={}, exception=exc.EResourceUpdate, message=("Failed in updating server 'NOVA_ID': The status of the " "port PORT must be DOWN."))), ('update:net-n:port-n:fixed_ip-n', dict( reason='update', success=False, inputs={'fixed_ip': 'FIXED_IP'}, net_result=[], port_result=[], sg_result=[], floating_result=[], result={}, exception=exc.EResourceUpdate, message=("Failed in updating server 'NOVA_ID': One of 'port' " "and 'network' must be provided."))), ('update:net-n:port-y:fixed_ip-y', dict( reason='update', success=False, inputs={'port': 'PORT', 'fixed_ip': 'FIXED_IP'}, net_result=[], port_result=[mock.Mock(id='PORT_ID', status='DOWN')], sg_result=[], floating_result=[], result={}, exception=exc.EResourceUpdate, message=("Failed in updating server 'NOVA_ID': The 'port' " "property and the 'fixed_ip' property cannot be " "specified at the same time."))), ] def setUp(self): super(TestNetworkValidation, self).setUp() self.nc = mock.Mock() self.profile = server.ServerProfile('t', spec) self.profile._networkclient = self.nc def test_validation(self): self.nc.network_get.side_effect = self.net_result self.nc.port_find.side_effect = self.port_result self.nc.security_group_find.side_effect = self.sg_result self.nc.floatingip_find.side_effect = self.floating_result obj = mock.Mock(physical_id='NOVA_ID') if self.success: res = self.profile._validate_network(obj, self.inputs, self.reason) self.assertEqual(self.result, res) else: ex = self.assertRaises(self.exception, self.profile._validate_network, obj, self.inputs, self.reason) self.assertEqual(self.message, str(ex)) if self.net_result: self.nc.network_get.assert_called_with('NET') if self.port_result: self.nc.port_find.assert_called_once_with('PORT') if self.sg_result: self.nc.security_group_find.assert_called_once_with( 'default', project_id=None) if self.floating_result: self.nc.floatingip_find.assert_called_once_with('FLOATINGIP') def test_validation_with_project_scope(self): self.nc.network_get.side_effect = self.net_result self.nc.port_find.side_effect = self.port_result self.nc.security_group_find.side_effect = self.sg_result self.nc.floatingip_find.side_effect = self.floating_result obj = mock.Mock(physical_id='NOVA_ID') self.profile.project_scope = True self.profile.project = 'FAKE_PROJECT_ID' if self.success: res = self.profile._validate_network(obj, self.inputs, self.reason) self.assertEqual(self.result, res) else: ex = self.assertRaises(self.exception, self.profile._validate_network, obj, self.inputs, self.reason) self.assertEqual(self.message, str(ex)) if self.net_result: self.nc.network_get.assert_called_with('NET') if self.port_result: self.nc.port_find.assert_called_once_with('PORT') if self.sg_result: self.nc.security_group_find.assert_called_once_with( 'default', project_id='FAKE_PROJECT_ID') if self.floating_result: self.nc.floatingip_find.assert_called_once_with('FLOATINGIP') class TestNovaServerValidate(base.SenlinTestCase): def setUp(self): super(TestNovaServerValidate, self).setUp() self.context = utils.dummy_context() def test_do_validate_all_passed(self): profile = server.ServerProfile('t', spec) mock_az = self.patchobject(profile, '_validate_az') mock_flavor = self.patchobject(profile, '_validate_flavor') mock_image = self.patchobject(profile, '_validate_image') mock_keypair = self.patchobject(profile, '_validate_keypair') mock_network = self.patchobject(profile, '_validate_network') obj = mock.Mock() res = profile.do_validate(obj) properties = spec['properties'] self.assertTrue(res) mock_az.assert_called_once_with(obj, properties['availability_zone']) mock_flavor.assert_called_once_with(obj, properties['flavor']) mock_image.assert_called_once_with(obj, properties['image']) mock_keypair.assert_called_once_with(obj, properties['key_name']) mock_network.assert_called_once_with(obj, properties['networks'][0])