"Fossies" - the Fresh Open Source Software Archive

Member "tacker-2.0.0/tacker/tests/functional/vnfm/test_tosca_vnf_floatingip.py" (16 Oct 2019, 4309 Bytes) of package /linux/misc/openstack/tacker-2.0.0.tar.gz:

As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the last Fossies "Diffs" side-by-side code changes report for "test_tosca_vnf_floatingip.py": 0.9.0_vs_1.0.0.

    1 # Copyright 2017 OpenStack Foundation
    2 #
    3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    4 #    not use this file except in compliance with the License. You may obtain
    5 #    a copy of the License at
    6 #
    7 #         http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 #    Unless required by applicable law or agreed to in writing, software
   10 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   12 #    License for the specific language governing permissions and limitations
   13 #    under the License.
   14 #
   15 
   16 import yaml
   17 
   18 from tacker.tests import constants
   19 from tacker.tests.functional import base
   20 from tacker.tests.utils import read_file
   21 
   22 
   23 class VnfTestToscaFloatingIp(base.BaseTackerTest):
   24 
   25     def get_heat_stack_resource(self, stack_id, resource_name):
   26         resource_types = self.h_client.resources
   27         resource_details = resource_types.get(stack_id=stack_id,
   28                                               resource_name=resource_name)
   29         resource_dict = resource_details.to_dict()
   30         return resource_dict
   31 
   32     def connect_public_and_private_nw_with_router(self):
   33         public_nw = 'public'
   34         private_nw = 'net_mgmt'
   35         private_nw_subnet = 'subnet_mgmt'
   36         public_nw_id = None
   37         private_nw_id = None
   38         private_nw_subnet_id = None
   39         neutronclient = self.neutronclient()
   40         networks = neutronclient.list_networks()['networks']
   41         for nw in networks:
   42             if nw['name'] == public_nw:
   43                 public_nw_id = nw['id']
   44             if nw['name'] == private_nw:
   45                 private_nw_id = nw['id']
   46             if public_nw_id and private_nw_id:
   47                 break
   48         self.assertIsNotNone(public_nw_id)
   49         self.assertIsNotNone(private_nw_id)
   50         subnets = neutronclient.list_subnets()['subnets']
   51         for subnet in subnets:
   52             if subnet['network_id'] == private_nw_id\
   53                     and subnet['name'] == private_nw_subnet:
   54                 private_nw_subnet_id = subnet['id']
   55                 break
   56         self.assertIsNotNone(private_nw_subnet_id)
   57         router_id = neutronclient.create_router(
   58             {'router': {'name': 'fip_test_router'}})['router']['id']
   59         self.assertIsNotNone(router_id)
   60         self.addCleanup(self.neutronclient().delete_router, router_id)
   61         rt_gw_id = neutronclient.add_gateway_router(
   62             router_id, {'network_id': public_nw_id})['router']['id']
   63         self.assertIsNotNone(rt_gw_id)
   64         self.addCleanup(self.neutronclient().remove_gateway_router,
   65             router_id)
   66         rt_int = neutronclient.add_interface_router(
   67             router_id, {'subnet_id': private_nw_subnet_id})['id']
   68         self.assertIsNotNone(rt_int)
   69         self.addCleanup(self.neutronclient().remove_interface_router,
   70             router_id, {'subnet_id': private_nw_subnet_id})
   71 
   72     def test_assign_floatingip_to_vdu(self):
   73         vnfd_file = 'sample_tosca_assign_floatingip_to_vdu.yaml'
   74         vnf_name = 'Assign Floating IP to VDU'
   75         values_str = read_file(vnfd_file)
   76         template = yaml.safe_load(values_str)
   77         vnf_arg = {'vnf': {'vnfd_template': template, 'name': vnf_name}}
   78         self.connect_public_and_private_nw_with_router()
   79         vnf_instance = self.client.create_vnf(body=vnf_arg)
   80         vnf_id = vnf_instance['vnf']['id']
   81         self.addCleanup(self.wait_until_vnf_delete, vnf_id,
   82                         constants.VNF_CIRROS_DELETE_TIMEOUT)
   83         self.addCleanup(self.client.delete_vnf, vnf_id)
   84         self.wait_until_vnf_active(
   85             vnf_id,
   86             constants.VNF_CIRROS_CREATE_TIMEOUT,
   87             constants.ACTIVE_SLEEP_TIME)
   88         vnf_show_out = self.client.show_vnf(vnf_id)['vnf']
   89         self.assertIsNotNone(vnf_show_out['mgmt_ip_address'])
   90 
   91         stack_id = vnf_show_out['instance_id']
   92         fip_res = self.get_heat_stack_resource(stack_id, 'FIP1')
   93         floating_ip_address = fip_res['attributes']['floating_ip_address']
   94         self.assertIsNotNone(floating_ip_address)
   95         fip_port_id = fip_res['attributes']['port_id']
   96         port_res = self.get_heat_stack_resource(stack_id, 'CP1')
   97         port_id = port_res['attributes']['id']
   98         self.assertEqual(fip_port_id, port_id)