# # 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 mock import mox from neutronclient.common import exceptions as qe from neutronclient.neutron import v2_0 as neutronV20 from neutronclient.v2_0 import client as neutronclient from oslo_serialization import jsonutils import six from heat.common import exception from heat.common import template_format from heat.engine import rsrc_defn from heat.engine import scheduler from heat.tests import common from heat.tests import utils neutron_port_template = ''' heat_template_version: 2015-04-30 description: Template to test port Neutron resource resources: port: type: OS::Neutron::Port properties: network: net1234 fixed_ips: - subnet: sub1234 ip_address: 10.0.3.21 device_owner: network:dhcp ''' neutron_port_with_address_pair_template = ''' heat_template_version: 2015-04-30 description: Template to test port Neutron resource resources: port: type: OS::Neutron::Port properties: network: abcd1234 allowed_address_pairs: - ip_address: 10.0.3.21 mac_address: 00-B0-D0-86-BB-F7 ''' neutron_port_security_template = ''' heat_template_version: 2015-04-30 description: Template to test port Neutron resource resources: port: type: OS::Neutron::Port properties: network: abcd1234 port_security_enabled: False ''' class NeutronPortTest(common.HeatTestCase): def setUp(self): super(NeutronPortTest, self).setUp() self.m.StubOutWithMock(neutronclient.Client, 'create_port') self.m.StubOutWithMock(neutronclient.Client, 'show_port') self.m.StubOutWithMock(neutronclient.Client, 'update_port') self.m.StubOutWithMock(neutronclient.Client, 'show_subnet') self.m.StubOutWithMock(neutronV20, 'find_resourceid_by_name_or_id') def test_missing_subnet_id(self): t = template_format.parse(neutron_port_template) t['resources']['port']['properties']['fixed_ips'][0].pop('subnet') stack = utils.parse_stack(t) neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'net1234', cmd_resource=None, ).MultipleTimes().AndReturn('net1234') neutronclient.Client.create_port({'port': { 'network_id': u'net1234', 'fixed_ips': [ {'ip_address': u'10.0.3.21'} ], 'name': utils.PhysName(stack.name, 'port'), 'admin_state_up': True, 'device_owner': u'network:dhcp'}} ).AndReturn({'port': { "status": "BUILD", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766" }}) neutronclient.Client.show_port( 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' ).AndReturn({'port': { "status": "ACTIVE", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766" }}) self.m.ReplayAll() port = stack['port'] scheduler.TaskRunner(port.create)() self.m.VerifyAll() def test_missing_ip_address(self): t = template_format.parse(neutron_port_template) t['resources']['port']['properties']['fixed_ips'][0].pop('ip_address') stack = utils.parse_stack(t) neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'net1234', cmd_resource=None, ).MultipleTimes().AndReturn('net1234') neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'subnet', 'sub1234', cmd_resource=None, ).MultipleTimes().AndReturn('sub1234') neutronclient.Client.create_port({'port': { 'network_id': u'net1234', 'fixed_ips': [ {'subnet_id': u'sub1234'} ], 'name': utils.PhysName(stack.name, 'port'), 'admin_state_up': True, 'device_owner': u'network:dhcp'}} ).AndReturn({'port': { "status": "BUILD", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766" }}) neutronclient.Client.show_port( 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' ).AndReturn({'port': { "status": "ACTIVE", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766" }}) self.m.ReplayAll() port = stack['port'] scheduler.TaskRunner(port.create)() self.m.VerifyAll() def test_missing_fixed_ips(self): t = template_format.parse(neutron_port_template) t['resources']['port']['properties'].pop('fixed_ips') stack = utils.parse_stack(t) neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'net1234', cmd_resource=None, ).MultipleTimes().AndReturn('net1234') neutronclient.Client.create_port({'port': { 'network_id': u'net1234', 'name': utils.PhysName(stack.name, 'port'), 'admin_state_up': True, 'device_owner': u'network:dhcp'}} ).AndReturn({'port': { "status": "BUILD", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766" }}) neutronclient.Client.show_port( 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' ).AndReturn({'port': { "status": "ACTIVE", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766", "fixed_ips": { "subnet_id": "d0e971a6-a6b4-4f4c-8c88-b75e9c120b7e", "ip_address": "10.0.0.2" } }}) self.m.ReplayAll() port = stack['port'] scheduler.TaskRunner(port.create)() self.m.VerifyAll() def test_allowed_address_pair(self): t = template_format.parse(neutron_port_with_address_pair_template) stack = utils.parse_stack(t) neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'abcd1234', cmd_resource=None, ).MultipleTimes().AndReturn('abcd1234') neutronclient.Client.create_port({'port': { 'network_id': u'abcd1234', 'allowed_address_pairs': [{ 'ip_address': u'10.0.3.21', 'mac_address': u'00-B0-D0-86-BB-F7' }], 'name': utils.PhysName(stack.name, 'port'), 'admin_state_up': True}} ).AndReturn({'port': { "status": "BUILD", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766" }}) neutronclient.Client.show_port( 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' ).AndReturn({'port': { "status": "ACTIVE", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766" }}) self.m.ReplayAll() port = stack['port'] scheduler.TaskRunner(port.create)() self.m.VerifyAll() def test_port_security_enabled(self): t = template_format.parse(neutron_port_security_template) stack = utils.parse_stack(t) neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'abcd1234', cmd_resource=None, ).MultipleTimes().AndReturn('abcd1234') neutronclient.Client.create_port({'port': { 'network_id': u'abcd1234', 'port_security_enabled': False, 'name': utils.PhysName(stack.name, 'port'), 'admin_state_up': True}} ).AndReturn({'port': { "status": "BUILD", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766" }}) neutronclient.Client.show_port( 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' ).AndReturn({'port': { "status": "ACTIVE", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766", }}) self.m.ReplayAll() port = stack['port'] scheduler.TaskRunner(port.create)() self.m.VerifyAll() def test_missing_mac_address(self): t = template_format.parse(neutron_port_with_address_pair_template) t['resources']['port']['properties']['allowed_address_pairs'][0].pop( 'mac_address' ) stack = utils.parse_stack(t) neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'abcd1234', cmd_resource=None, ).MultipleTimes().AndReturn('abcd1234') neutronclient.Client.create_port({'port': { 'network_id': u'abcd1234', 'allowed_address_pairs': [{ 'ip_address': u'10.0.3.21', }], 'name': utils.PhysName(stack.name, 'port'), 'admin_state_up': True}} ).AndReturn({'port': { "status": "BUILD", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766" }}) neutronclient.Client.show_port( 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' ).AndReturn({'port': { "status": "ACTIVE", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766" }}) self.m.ReplayAll() port = stack['port'] scheduler.TaskRunner(port.create)() self.m.VerifyAll() def test_ip_address_is_cidr(self): t = template_format.parse(neutron_port_with_address_pair_template) t['resources']['port']['properties'][ 'allowed_address_pairs'][0]['ip_address'] = '10.0.3.0/24' stack = utils.parse_stack(t) neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'abcd1234', cmd_resource=None, ).MultipleTimes().AndReturn('abcd1234') neutronclient.Client.create_port({'port': { 'network_id': u'abcd1234', 'allowed_address_pairs': [{ 'ip_address': u'10.0.3.0/24', 'mac_address': u'00-B0-D0-86-BB-F7' }], 'name': utils.PhysName(stack.name, 'port'), 'admin_state_up': True}} ).AndReturn({'port': { "status": "BUILD", "id": "2e00180a-ff9d-42c4-b701-a0606b243447" }}) neutronclient.Client.show_port( '2e00180a-ff9d-42c4-b701-a0606b243447' ).AndReturn({'port': { "status": "ACTIVE", "id": "2e00180a-ff9d-42c4-b701-a0606b243447" }}) self.m.ReplayAll() port = stack['port'] scheduler.TaskRunner(port.create)() self.m.VerifyAll() def _mock_create_with_props(self, port_prop): neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'net1234', cmd_resource=None, ).MultipleTimes().AndReturn('net1234') neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'subnet', 'sub1234', cmd_resource=None, ).MultipleTimes().AndReturn('sub1234') neutronclient.Client.create_port({'port': port_prop}).AndReturn( {'port': { "status": "BUILD", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766"}}) neutronclient.Client.show_port( 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' ).MultipleTimes().AndReturn({'port': { "status": "ACTIVE", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766", "dns_assignment": { "hostname": "my-vm", "ip_address": "10.0.0.15", "fqdn": "my-vm.openstack.org."} }}) self.m.ReplayAll() def test_security_groups(self): t = template_format.parse(neutron_port_template) t['resources']['port']['properties']['security_groups'] = [ '8a2f582a-e1cd-480f-b85d-b02631c10656', '024613dc-b489-4478-b46f-ada462738740'] stack = utils.parse_stack(t) port_prop = { 'network_id': u'net1234', 'security_groups': ['8a2f582a-e1cd-480f-b85d-b02631c10656', '024613dc-b489-4478-b46f-ada462738740'], 'fixed_ips': [ {'subnet_id': u'sub1234', 'ip_address': u'10.0.3.21'} ], 'name': utils.PhysName(stack.name, 'port'), 'admin_state_up': True, 'device_owner': u'network:dhcp'} self._mock_create_with_props(port_prop) port = stack['port'] scheduler.TaskRunner(port.create)() self.m.VerifyAll() def test_port_with_dns_name(self): t = template_format.parse(neutron_port_template) t['resources']['port']['properties']['dns_name'] = 'myvm' stack = utils.parse_stack(t) port_prop = { 'network_id': u'net1234', 'dns_name': 'myvm', 'fixed_ips': [ {'subnet_id': u'sub1234', 'ip_address': u'10.0.3.21'} ], 'name': utils.PhysName(stack.name, 'port'), 'admin_state_up': True, 'device_owner': u'network:dhcp'} self._mock_create_with_props(port_prop) port = stack['port'] scheduler.TaskRunner(port.create)() self.assertEqual('my-vm.openstack.org.', port.FnGetAtt('dns_assignment')['fqdn']) self.m.VerifyAll() def test_security_groups_empty_list(self): t = template_format.parse(neutron_port_template) t['resources']['port']['properties']['security_groups'] = [] stack = utils.parse_stack(t) port_prop = { 'network_id': u'net1234', 'security_groups': [], 'fixed_ips': [ {'subnet_id': u'sub1234', 'ip_address': u'10.0.3.21'} ], 'name': utils.PhysName(stack.name, 'port'), 'admin_state_up': True, 'device_owner': u'network:dhcp' } self._mock_create_with_props(port_prop) port = stack['port'] scheduler.TaskRunner(port.create)() self.m.VerifyAll() def test_port_needs_update(self): t = template_format.parse(neutron_port_template) t['resources']['port']['properties'].pop('fixed_ips') stack = utils.parse_stack(t) props = {'network_id': u'net1234', 'name': utils.PhysName(stack.name, 'port'), 'admin_state_up': True, 'device_owner': u'network:dhcp'} neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'net1234', cmd_resource=None, ).MultipleTimes().AndReturn('net1234') neutronclient.Client.create_port( {'port': props} ).AndReturn({'port': { "status": "BUILD", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766" }}) neutronclient.Client.show_port( 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' ).AndReturn({'port': { "status": "ACTIVE", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766", "fixed_ips": { "subnet_id": "d0e971a6-a6b4-4f4c-8c88-b75e9c120b7e", "ip_address": "10.0.0.2" } }}) self.m.ReplayAll() # create port port = stack['port'] scheduler.TaskRunner(port.create)() new_props = props.copy() # test always replace new_props['replacement_policy'] = 'REPLACE_ALWAYS' new_props['network'] = new_props.pop('network_id') update_snippet = rsrc_defn.ResourceDefinition(port.name, port.type(), new_props) self.assertRaises(exception.UpdateReplace, port._needs_update, update_snippet, port.frozen_definition(), new_props, port.properties, None) # test deferring to Resource._needs_update new_props['replacement_policy'] = 'AUTO' update_snippet = rsrc_defn.ResourceDefinition(port.name, port.type(), new_props) self.assertTrue(port._needs_update(update_snippet, port.frozen_definition(), new_props, port.properties, None)) self.m.VerifyAll() def test_port_needs_update_network(self): props = {'network_id': u'net1234', 'name': 'test_port', 'admin_state_up': True, 'device_owner': u'network:dhcp'} neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'net1234', cmd_resource=None, ).MultipleTimes().AndReturn('net1234') neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'old_network', cmd_resource=None, ).MultipleTimes().AndReturn('net1234') neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'new_network', cmd_resource=None, ).MultipleTimes().AndReturn('net5678') create_props = props.copy() neutronclient.Client.create_port( {'port': create_props} ).AndReturn({'port': { "status": "BUILD", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766" }}) neutronclient.Client.show_port( 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' ).AndReturn({'port': { "status": "ACTIVE", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766", "fixed_ips": { "subnet_id": "d0e971a6-a6b4-4f4c-8c88-b75e9c120b7e", "ip_address": "10.0.0.2" } }}) neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'net5678', cmd_resource=None, ).MultipleTimes().AndReturn('net5678') call_dict = copy.deepcopy(props) call_dict['security_groups'] = [ '0389f747-7785-4757-b7bb-2ab07e4b09c3'] del call_dict['network_id'] neutronclient.Client.show_port( 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' ).AndReturn({'port': { "status": "ACTIVE", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766" }}) neutronclient.Client.show_port( 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' ).AndReturn({'port': { "status": "ACTIVE", "id": "fc68ea2c-b60b-4b4f-bd82-94ec81110766" }}) neutronclient.Client.update_port( 'fc68ea2c-b60b-4b4f-bd82-94ec81110766', {'port': {'fixed_ips': []}} ).AndReturn(None) self.m.ReplayAll() # create port t = template_format.parse(neutron_port_template) t['resources']['port']['properties'].pop('fixed_ips') t['resources']['port']['properties']['name'] = 'test_port' stack = utils.parse_stack(t) port = stack['port'] scheduler.TaskRunner(port.create)() # Switch from network_id=ID to network=ID (no replace) new_props = props.copy() new_props['network'] = new_props.pop('network_id') update_snippet = rsrc_defn.ResourceDefinition(port.name, port.type(), new_props) scheduler.TaskRunner(port.update, update_snippet)() self.assertEqual((port.UPDATE, port.COMPLETE), port.state) # Switch from network=ID to network=NAME (no replace) new_props['network'] = 'old_network' update_snippet = rsrc_defn.ResourceDefinition(port.name, port.type(), new_props) scheduler.TaskRunner(port.update, update_snippet)() self.assertEqual((port.UPDATE, port.COMPLETE), port.state) # Switch to a different network (replace) new_props['network'] = 'new_network' update_snippet = rsrc_defn.ResourceDefinition(port.name, port.type(), new_props) updater = scheduler.TaskRunner(port.update, update_snippet) self.assertRaises(exception.UpdateReplace, updater) self.m.VerifyAll() def test_get_port_attributes(self): t = template_format.parse(neutron_port_template) t['resources']['port']['properties'].pop('fixed_ips') stack = utils.parse_stack(t) subnet_dict = {'name': 'test-subnet', 'enable_dhcp': True, 'network_id': 'net1234', 'dns_nameservers': [], 'tenant_id': '58a61fc3992944ce971404a2ece6ff98', 'ipv6_ra_mode': None, 'cidr': '10.0.0.0/24', 'allocation_pools': [{'start': '10.0.0.2', 'end': u'10.0.0.254'}], 'gateway_ip': '10.0.0.1', 'ipv6_address_mode': None, 'ip_version': 4, 'host_routes': [], 'id': '6dd609ad-d52a-4587-b1a0-b335f76062a5'} neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'net1234', cmd_resource=None, ).MultipleTimes().AndReturn('net1234') neutronclient.Client.create_port({'port': { 'network_id': u'net1234', 'name': utils.PhysName(stack.name, 'port'), 'admin_state_up': True, 'device_owner': u'network:dhcp'}} ).AndReturn({'port': { 'status': 'BUILD', 'id': 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' }}) neutronclient.Client.show_subnet( 'd0e971a6-a6b4-4f4c-8c88-b75e9c120b7e' ).AndReturn({'subnet': subnet_dict}) neutronclient.Client.show_port( 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' ).MultipleTimes().AndReturn({'port': { 'status': 'DOWN', 'name': utils.PhysName(stack.name, 'port'), 'allowed_address_pairs': [], 'admin_state_up': True, 'network_id': 'net1234', 'device_id': 'dc68eg2c-b60g-4b3f-bd82-67ec87650532', 'mac_address': 'fa:16:3e:75:67:60', 'tenant_id': '58a61fc3992944ce971404a2ece6ff98', 'security_groups': ['5b15d80c-6b70-4a1c-89c9-253538c5ade6'], 'fixed_ips': [{'subnet_id': 'd0e971a6-a6b4-4f4c-8c88-b75e9c120b7e', 'ip_address': '10.0.0.2'}] }}) self.m.ReplayAll() port = stack['port'] scheduler.TaskRunner(port.create)() self.assertEqual('DOWN', port.FnGetAtt('status')) self.assertEqual([], port.FnGetAtt('allowed_address_pairs')) self.assertTrue(port.FnGetAtt('admin_state_up')) self.assertEqual('net1234', port.FnGetAtt('network_id')) self.assertEqual('fa:16:3e:75:67:60', port.FnGetAtt('mac_address')) self.assertEqual(utils.PhysName(stack.name, 'port'), port.FnGetAtt('name')) self.assertEqual('dc68eg2c-b60g-4b3f-bd82-67ec87650532', port.FnGetAtt('device_id')) self.assertEqual('58a61fc3992944ce971404a2ece6ff98', port.FnGetAtt('tenant_id')) self.assertEqual(['5b15d80c-6b70-4a1c-89c9-253538c5ade6'], port.FnGetAtt('security_groups')) self.assertEqual([{'subnet_id': 'd0e971a6-a6b4-4f4c-8c88-b75e9c120b7e', 'ip_address': '10.0.0.2'}], port.FnGetAtt('fixed_ips')) self.assertEqual([subnet_dict], port.FnGetAtt('subnets')) self.assertRaises(exception.InvalidTemplateAttribute, port.FnGetAtt, 'Foo') self.m.VerifyAll() def test_subnet_attribute_exception(self): t = template_format.parse(neutron_port_template) t['resources']['port']['properties'].pop('fixed_ips') stack = utils.parse_stack(t) neutronV20.find_resourceid_by_name_or_id( mox.IsA(neutronclient.Client), 'network', 'net1234', cmd_resource=None, ).MultipleTimes().AndReturn('net1234') neutronclient.Client.create_port({'port': { 'network_id': u'net1234', 'name': utils.PhysName(stack.name, 'port'), 'admin_state_up': True, 'device_owner': u'network:dhcp'}} ).AndReturn({'port': { 'status': 'BUILD', 'id': 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' }}) neutronclient.Client.show_port( 'fc68ea2c-b60b-4b4f-bd82-94ec81110766' ).MultipleTimes().AndReturn({'port': { 'status': 'DOWN', 'name': utils.PhysName(stack.name, 'port'), 'allowed_address_pairs': [], 'admin_state_up': True, 'network_id': 'net1234', 'device_id': 'dc68eg2c-b60g-4b3f-bd82-67ec87650532', 'mac_address': 'fa:16:3e:75:67:60', 'tenant_id': '58a61fc3992944ce971404a2ece6ff98', 'security_groups': ['5b15d80c-6b70-4a1c-89c9-253538c5ade6'], 'fixed_ips': [{'subnet_id': 'd0e971a6-a6b4-4f4c-8c88-b75e9c120b7e', 'ip_address': '10.0.0.2'}] }}) neutronclient.Client.show_subnet( 'd0e971a6-a6b4-4f4c-8c88-b75e9c120b7e' ).AndRaise(qe.NeutronClientException('ConnectionFailed: Connection ' 'to neutron failed: Maximum ' 'attempts reached')) self.m.ReplayAll() port = stack['port'] scheduler.TaskRunner(port.create)() self.assertIsNone(port.FnGetAtt('subnets')) log_msg = ('Failed to fetch resource attributes: ConnectionFailed: ' 'Connection to neutron failed: Maximum attempts reached') self.assertIn(log_msg, self.LOG.output) self.m.VerifyAll() def test_prepare_for_replace_port_not_created(self): t = template_format.parse(neutron_port_template) stack = utils.parse_stack(t) port = stack['port'] port._show_resource = mock.Mock() port.data_set = mock.Mock() n_client = mock.Mock() port.client = mock.Mock(return_value=n_client) self.assertIsNone(port.resource_id) # execute prepare_for_replace port.prepare_for_replace() # check, if the port is not created, do nothing in # prepare_for_replace() self.assertFalse(port._show_resource.called) self.assertFalse(port.data_set.called) self.assertFalse(n_client.update_port.called) def test_prepare_for_replace_port(self): t = template_format.parse(neutron_port_template) stack = utils.parse_stack(t) port = stack['port'] port.resource_id = 'test_res_id' _value = { 'fixed_ips': { 'subnet_id': 'test_subnet', 'ip_address': '42.42.42.42' } } port._show_resource = mock.Mock(return_value=_value) port.data_set = mock.Mock() n_client = mock.Mock() port.client = mock.Mock(return_value=n_client) # execute prepare_for_replace port.prepare_for_replace() # check, that data was stored port.data_set.assert_called_once_with( 'port_fip', jsonutils.dumps(_value.get('fixed_ips'))) # check, that port was updated and ip was removed expected_props = {'port': {'fixed_ips': []}} n_client.update_port.assert_called_once_with('test_res_id', expected_props) def test_restore_prev_rsrc(self): t = template_format.parse(neutron_port_template) stack = utils.parse_stack(t) new_port = stack['port'] new_port.resource_id = 'new_res_id' # mock backup stack to return only one mocked old_port old_port = mock.Mock() new_port.stack._backup_stack = mock.Mock() new_port.stack._backup_stack().resources.get.return_value = old_port old_port.resource_id = 'old_res_id' _value = { 'subnet_id': 'test_subnet', 'ip_address': '42.42.42.42' } old_port.data = mock.Mock( return_value={'port_fip': jsonutils.dumps(_value)}) n_client = mock.Mock() new_port.client = mock.Mock(return_value=n_client) # execute restore_prev_rsrc new_port.restore_prev_rsrc() # check, that ports were updated: old port get ip and # same ip was removed from old port expected_new_props = {'port': {'fixed_ips': []}} expected_old_props = {'port': {'fixed_ips': _value}} n_client.update_port.assert_has_calls([ mock.call('new_res_id', expected_new_props), mock.call('old_res_id', expected_old_props)]) def test_restore_prev_rsrc_convergence(self): t = template_format.parse(neutron_port_template) stack = utils.parse_stack(t) stack.store() # mock resource from previous template prev_rsrc = stack['port'] prev_rsrc.resource_id = 'prev-rsrc' # store in db prev_rsrc.state_set(prev_rsrc.UPDATE, prev_rsrc.COMPLETE) # mock resource from existing template and store in db existing_rsrc = stack['port'] existing_rsrc.current_template_id = stack.t.id existing_rsrc.resource_id = 'existing-rsrc' existing_rsrc.state_set(existing_rsrc.UPDATE, existing_rsrc.COMPLETE) # mock previous resource was replaced by existing resource prev_rsrc.replaced_by = existing_rsrc.id _value = { 'subnet_id': 'test_subnet', 'ip_address': '42.42.42.42' } prev_rsrc._data = {'port_fip': jsonutils.dumps(_value)} n_client = mock.Mock() prev_rsrc.client = mock.Mock(return_value=n_client) # execute restore_prev_rsrc prev_rsrc.restore_prev_rsrc(convergence=True) expected_existing_props = {'port': {'fixed_ips': []}} expected_prev_props = {'port': {'fixed_ips': _value}} n_client.update_port.assert_has_calls([ mock.call(existing_rsrc.resource_id, expected_existing_props), mock.call(prev_rsrc.resource_id, expected_prev_props)]) class UpdatePortTest(common.HeatTestCase): scenarios = [ ('with_secgrp', dict(secgrp=['8a2f582a-e1cd-480f-b85d-b02631c10656'], name='test', value_specs={}, fixed_ips=None, addr_pair=None, vnic_type=None)), ('with_no_name', dict(secgrp=['8a2f582a-e1cd-480f-b85d-b02631c10656'], name=None, value_specs={}, fixed_ips=None, addr_pair=None, vnic_type=None)), ('with_empty_values', dict(secgrp=[], name='test', value_specs={}, fixed_ips=[], addr_pair=[], vnic_type=None)), ('with_fixed_ips', dict(secgrp=None, value_specs={}, fixed_ips=[ {"subnet_id": "d0e971a6-a6b4-4f4c", "ip_address": "10.0.0.2"}], addr_pair=None, vnic_type=None)), ('with_addr_pair', dict(secgrp=None, value_specs={}, fixed_ips=None, addr_pair=[{'ip_address': '10.0.3.21', 'mac_address': '00-B0-D0-86'}], vnic_type=None)), ('with_value_specs', dict(secgrp=None, value_specs={'binding:vnic_type': 'direct'}, fixed_ips=None, addr_pair=None, vnic_type=None)), ('normal_vnic', dict(secgrp=None, value_specs={}, fixed_ips=None, addr_pair=None, vnic_type='normal')), ('direct_vnic', dict(secgrp=None, value_specs={}, fixed_ips=None, addr_pair=None, vnic_type='direct')), ('with_all', dict(secgrp=['8a2f582a-e1cd-480f-b85d-b02631c10656'], value_specs={}, fixed_ips=[ {"subnet_id": "d0e971a6-a6b4-4f4c", "ip_address": "10.0.0.2"}], addr_pair=[{'ip_address': '10.0.3.21', 'mac_address': '00-B0-D0-86-BB-F7'}], vnic_type='normal')), ] def test_update_port(self): t = template_format.parse(neutron_port_template) t['resources']['port']['properties'].pop('fixed_ips') stack = utils.parse_stack(t) self.patchobject(neutronV20, 'find_resourceid_by_name_or_id', return_value='net1234') create_port = self.patchobject(neutronclient.Client, 'create_port') update_port = self.patchobject(neutronclient.Client, 'update_port') fake_groups_list = { 'security_groups': [ { 'tenant_id': 'dc4b074874244f7693dd65583733a758', 'id': '0389f747-7785-4757-b7bb-2ab07e4b09c3', 'name': 'default', 'security_group_rules': [], 'description': 'no protocol' } ] } self.patchobject(neutronclient.Client, 'list_security_groups', return_value=fake_groups_list) props = {'network_id': u'net1234', 'name': utils.PhysName(stack.name, 'port'), 'admin_state_up': True, 'device_owner': u'network:dhcp'} update_props = props.copy() update_props['security_groups'] = self.secgrp update_props['value_specs'] = self.value_specs if self.fixed_ips: update_props['fixed_ips'] = self.fixed_ips update_props['allowed_address_pairs'] = self.addr_pair update_props['binding:vnic_type'] = self.vnic_type update_dict = update_props.copy() if update_props['security_groups'] is None: update_dict['security_groups'] = ['default'] if update_props['name'] is None: update_dict['name'] = utils.PhysName(stack.name, 'test_subnet') value_specs = update_dict.pop('value_specs') if value_specs: for value_spec in six.iteritems(value_specs): update_dict[value_spec[0]] = value_spec[1] # create port port = stack['port'] self.assertIsNone(scheduler.TaskRunner(port.handle_create)()) create_port.assset_called_once_with(props) # update port update_snippet = rsrc_defn.ResourceDefinition(port.name, port.type(), update_props) self.assertIsNone(scheduler.TaskRunner(port.handle_update, update_snippet, {}, update_props)()) update_port.assset_called_once_with(update_dict) # update with empty prop_diff scheduler.TaskRunner(port.handle_update, update_snippet, {}, {})() self.assertEqual(1, update_port.call_count)