Source code for heat_integrationtests.functional.test_create_update_neutron_trunk

# Copyright (c) 2017 Ericsson.
#
#    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 yaml

from heat_integrationtests.functional import functional_base


test_template = '''
heat_template_version: pike
description: Test template to create, update, delete trunk.
resources:
  parent_net:
    type: OS::Neutron::Net
  trunk_net_one:
    type: OS::Neutron::Net
  trunk_net_two:
    type: OS::Neutron::Net
  parent_subnet:
    type: OS::Neutron::Subnet
    properties:
      network: { get_resource: parent_net }
      cidr: 10.0.0.0/16
  trunk_subnet_one:
    type: OS::Neutron::Subnet
    properties:
      network: { get_resource: trunk_net_one }
      cidr: 10.10.0.0/16
  trunk_subnet_two:
    type: OS::Neutron::Subnet
    properties:
      network: { get_resource: trunk_net_two }
      cidr: 10.20.0.0/16
  parent_port:
    type: OS::Neutron::Port
    properties:
      network: { get_resource: parent_net }
      name: trunk_parent_port
  sub_port_one:
    type: OS::Neutron::Port
    properties:
      network: { get_resource: trunk_net_one }
      name: trunk_sub_port_one
  sub_port_two:
    type: OS::Neutron::Port
    properties:
      network: { get_resource: trunk_net_two }
      name: trunk_sub_port_two
  trunk:
    type: OS::Neutron::Trunk
    properties:
      name: test_trunk
      port: { get_resource: parent_port }
      sub_ports:
outputs:
  trunk_parent_port:
    value: { get_attr: [trunk, port_id] }
'''


[docs]class UpdateTrunkTest(functional_base.FunctionalTestsBase):

    @staticmethod
    def _sub_ports_dict_to_set(sub_ports):
        new_sub_ports = copy.deepcopy(sub_ports)

        # NOTE(lajos katona): In the template we have to give the sub port as
        # port, but from trunk_details we receive back them with port_id.
        # As an extra trunk_details contains the mac_address as well which is
        # useless here.
        # So here we have to make sure that the dictionary (input from
        # template or output from trunk_details) have the same keys:
        if any('mac_address' in d for d in new_sub_ports):
            for sp in new_sub_ports:
                sp['port'] = sp['port_id']
                del sp['port_id']
                del sp['mac_address']

        # NOTE(lajos katona): We receive lists (trunk_details['sub_ports'] and
        # the input to the template) and we can't be sure that the order is the
        # same, so by using sets we can compare them.
        sub_ports_set = {frozenset(d.items()) for d in new_sub_ports}
        return sub_ports_set

[docs]    def test_add_first_sub_port(self):
        stack_identifier = self.stack_create(template=test_template)

        parsed_template = yaml.safe_load(test_template)
        new_sub_port = [{'port': {'get_resource': 'sub_port_one'},
                         'segmentation_id': 10,
                         'segmentation_type': 'vlan'}]
        parsed_template['resources']['trunk']['properties'][
            'sub_ports'] = new_sub_port
        updated_template = yaml.safe_dump(parsed_template)
        self.update_stack(stack_identifier, updated_template)

        # Fix the port_id in the template for assertion
        new_sub_port[0]['port'] = self.get_physical_resource_id(
            stack_identifier, 'sub_port_one')
        parent_id = self.get_stack_output(
            stack_identifier, 'trunk_parent_port')
        parent_port = self.network_client.show_port(parent_id)['port']
        trunk_sub_port = parent_port['trunk_details']['sub_ports']

        self.assertEqual(self._sub_ports_dict_to_set(new_sub_port),
                         self._sub_ports_dict_to_set(trunk_sub_port))

[docs]    def test_add_a_second_sub_port(self):
        parsed_template = yaml.safe_load(test_template)
        sub_ports = [{'port': {'get_resource': 'sub_port_one'},
                      'segmentation_type': 'vlan',
                      'segmentation_id': 10}, ]
        parsed_template['resources']['trunk']['properties'][
            'sub_ports'] = sub_ports
        template_with_sub_ports = yaml.safe_dump(parsed_template)

        stack_identifier = self.stack_create(template=template_with_sub_ports)

        new_sub_port = {'port': {'get_resource': 'sub_port_two'},
                        'segmentation_id': 20,
                        'segmentation_type': 'vlan'}
        parsed_template['resources']['trunk']['properties'][
            'sub_ports'].append(new_sub_port)

        updated_template = yaml.safe_dump(parsed_template)

        self.update_stack(stack_identifier, updated_template)

        # Fix the port_ids in the templates for assertion
        sub_ports[0]['port'] = self.get_physical_resource_id(
            stack_identifier, 'sub_port_one')
        new_sub_port['port'] = self.get_physical_resource_id(
            stack_identifier, 'sub_port_two')
        expected_sub_ports = [sub_ports[0], new_sub_port]

        parent_id = self.get_stack_output(
            stack_identifier, 'trunk_parent_port')
        parent_port = self.network_client.show_port(parent_id)['port']
        trunk_sub_ports = parent_port['trunk_details']['sub_ports']

        self.assertEqual(self._sub_ports_dict_to_set(expected_sub_ports),
                         self._sub_ports_dict_to_set(trunk_sub_ports))

[docs]    def test_remove_sub_port_from_trunk(self):
        sub_ports = [{'port': {'get_resource': 'sub_port_one'},
                      'segmentation_type': 'vlan',
                      'segmentation_id': 10},
                     {'port': {'get_resource': 'sub_port_two'},
                      'segmentation_type': 'vlan',
                      'segmentation_id': 20}]
        parsed_template = yaml.safe_load(test_template)
        parsed_template['resources']['trunk']['properties'][
            'sub_ports'] = sub_ports
        template_with_sub_ports = yaml.safe_dump(parsed_template)

        stack_identifier = self.stack_create(template=template_with_sub_ports)

        sub_port_to_be_removed = {'port': {'get_resource': 'sub_port_two'},
                                  'segmentation_type': 'vlan',
                                  'segmentation_id': 20}
        parsed_template['resources']['trunk'][
            'properties']['sub_ports'].remove(sub_port_to_be_removed)
        updated_template = yaml.safe_dump(parsed_template)

        self.update_stack(stack_identifier, updated_template)

        # Fix the port_ids in the templates for assertion
        sub_ports[0]['port'] = self.get_physical_resource_id(
            stack_identifier, 'sub_port_one')
        expected_sub_ports = [sub_ports[0]]

        parent_id = self.get_stack_output(
            stack_identifier, 'trunk_parent_port')
        parent_port = self.network_client.show_port(parent_id)['port']
        trunk_sub_ports = parent_port['trunk_details']['sub_ports']

        self.assertEqual(self._sub_ports_dict_to_set(expected_sub_ports),
                         self._sub_ports_dict_to_set(trunk_sub_ports))

[docs]    def test_remove_last_sub_port_from_trunk(self):
        sub_ports = [{'port': {'get_resource': 'sub_port_one'},
                      'segmentation_type': 'vlan',
                      'segmentation_id': 10}]
        parsed_template = yaml.safe_load(test_template)
        parsed_template['resources']['trunk']['properties'][
            'sub_ports'] = sub_ports

        template_with_sub_ports = yaml.safe_dump(parsed_template)
        stack_identifier = self.stack_create(template=template_with_sub_ports)

        sub_port_to_be_removed = {'port': {'get_resource': 'sub_port_one'},
                                  'segmentation_type': 'vlan',
                                  'segmentation_id': 10}

        parsed_template['resources']['trunk'][
            'properties']['sub_ports'] = []
        updated_template = yaml.safe_dump(parsed_template)

        self.update_stack(stack_identifier, updated_template)

        sub_port_to_be_removed['port'] = self.get_physical_resource_id(
            stack_identifier, 'sub_port_one')
        parent_id = self.get_stack_output(
            stack_identifier, 'trunk_parent_port')
        parent_port = self.network_client.show_port(parent_id)['port']
        trunk_sub_ports = parent_port['trunk_details']['sub_ports']

        self.assertNotEqual(
            self._sub_ports_dict_to_set([sub_port_to_be_removed]),
            self._sub_ports_dict_to_set(trunk_sub_ports))
        self.assertFalse(trunk_sub_ports,
                         'The returned sub ports (%s) in trunk_details is '
                         'not empty!' % trunk_sub_ports)

[docs]    def test_update_existing_sub_port_on_trunk(self):
        sub_ports = [{'port': {'get_resource': 'sub_port_one'},
                      'segmentation_type': 'vlan',
                      'segmentation_id': 10}]
        parsed_template = yaml.safe_load(test_template)
        parsed_template['resources']['trunk']['properties'][
            'sub_ports'] = sub_ports

        template_with_sub_ports = yaml.safe_dump(parsed_template)
        stack_identifier = self.stack_create(template=template_with_sub_ports)

        sub_port_id = self.get_physical_resource_id(
            stack_identifier, 'sub_port_one')
        parsed_template['resources']['trunk']['properties']['sub_ports'][0][
            'segmentation_id'] = 99
        updated_template = yaml.safe_dump(parsed_template)

        self.update_stack(stack_identifier, updated_template)
        updated_sub_port = {'port': sub_port_id,
                            'segmentation_type': 'vlan',
                            'segmentation_id': 99}
        parent_id = self.get_stack_output(
            stack_identifier, 'trunk_parent_port')
        parent_port = self.network_client.show_port(parent_id)['port']
        trunk_sub_ports = parent_port['trunk_details']['sub_ports']

        self.assertEqual(self._sub_ports_dict_to_set([updated_sub_port]),
                         self._sub_ports_dict_to_set(trunk_sub_ports))

[docs]    def test_update_trunk_name_and_description(self):
        new_name = 'pineapple'
        new_description = 'This is a test trunk'

        stack_identifier = self.stack_create(template=test_template)
        parsed_template = yaml.safe_load(test_template)
        parsed_template['resources']['trunk']['properties']['name'] = new_name
        parsed_template['resources']['trunk']['properties'][
            'description'] = new_description
        updated_template = yaml.safe_dump(parsed_template)
        self.update_stack(stack_identifier, template=updated_template)

        parent_id = self.get_stack_output(
            stack_identifier, 'trunk_parent_port')
        parent_port = self.network_client.show_port(parent_id)['port']
        trunk_id = parent_port['trunk_details']['trunk_id']

        trunk = self.network_client.show_trunk(trunk_id)['trunk']
        self.assertEqual(new_name, trunk['name'])
        self.assertEqual(new_description, trunk['description'])