rwlaunchpad/ra/pytest/ns/pingpong/test_floating_ip.py

   1 #!/usr/bin/env python3
   2 """
   3 #
   4 #   Copyright 2017 RIFT.IO Inc
   5 #
   6 #   Licensed under the Apache License, Version 2.0 (the "License");
   7 #   you may not use this file except in compliance with the License.
   8 #   You may obtain a copy of the License at
   9 #
  10 #       http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 #   Unless required by applicable law or agreed to in writing, software
  13 #   distributed under the License is distributed on an "AS IS" BASIS,
  14 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15 #   See the License for the specific language governing permissions and
  16 #   limitations under the License.
  17 #
  18 """
  19
  20 import gi
  21 import os
  22
  23 import rift.auto.descriptor
  24 import rift.auto.mano as mano
  25
  26 gi.require_version('RwNsrYang', '1.0')
  27 gi.require_version('RwProjectNsdYang', '1.0')
  28 gi.require_version('RwProjectVnfdYang', '1.0')
  29 gi.require_version('RwVnfrYang', '1.0')
  30 gi.require_version('RwCloudYang', '1.0')
  31
  32 from gi.repository import (
  33     RwProjectNsdYang,
  34     RwNsrYang,
  35     RwVnfrYang,
  36     RwProjectVnfdYang,
  37     RwCloudYang
  38 )
  39 gi.require_version('RwKeyspec', '1.0')
  40 from gi.repository.RwKeyspec import quoted_key
  41
  42
  43 class TestFloatingIP(object):
  44     """TestFloatingIP."""
  45
  46     # After RIFTIT-909 is completed this test will be set to working
  47     valid_pool_names = ['FIP_SYSTEST_POOL_LARGE', 'public']
  48     invalid_pool_names = ['', 'FIP_SYSTEST_POOL_EMPTY', 'invalid']
  49
  50     def create_cloud_account(
  51             self, cloud_host, cloud_user, cloud_tenants, vim_ssl_enabled,
  52             idx, mgmt_session):
  53         """create_cloud_account."""
  54         for cloud_tenant in cloud_tenants:
  55             floating_ip_pool_names = (
  56                 self.valid_pool_names + self.invalid_pool_names)
  57             project_name = 'float_project_{}'.format(idx)
  58             password = 'mypasswd'
  59             auth_url = 'http://{host}:5000/v3/'.format(host=cloud_host)
  60             if vim_ssl_enabled is True:
  61                 auth_url = 'https://{host}:5000/v3/'.format(host=cloud_host)
  62             mgmt_network = os.getenv('MGMT_NETWORK', 'private')
  63             cloud_acc_name = 'cloud_account'
  64             pool_name = floating_ip_pool_names[idx - 1]
  65             cloud_account = (
  66                 RwCloudYang.
  67                 YangData_RwProject_Project_Cloud_Account.from_dict({
  68                     'name': cloud_acc_name,
  69                     'account_type': 'openstack',
  70                     'openstack': {
  71                         'admin': True,
  72                         'key': cloud_user,
  73                         'secret': password,
  74                         'auth_url': auth_url,
  75                         'tenant': cloud_tenant,
  76                         'mgmt_network': mgmt_network,
  77                         'floating_ip_pool': pool_name,
  78                     }
  79                 }))
  80             mano.create_cloud_account(
  81                 mgmt_session, cloud_account, project_name=project_name)
  82
  83     def yield_vnfd_vnfr_pairs(self, proxy, nsr=None):
  84         """
  85         Yield tuples of vnfd & vnfr entries.
  86
  87         Args:
  88             proxy (callable): Launchpad proxy
  89             nsr (optional): If specified, only the vnfr & vnfd records of the
  90                 NSR are returned
  91
  92         Yields:
  93             Tuple: VNFD and its corresponding VNFR entry
  94         """
  95         def get_vnfd(vnfd_id):
  96             xpath = (
  97                 "/rw-project:project[rw-project:name='default']/" +
  98                 "vnfd-catalog/vnfd[id={}]".format(quoted_key(vnfd_id)))
  99             return proxy(RwProjectVnfdYang).get(xpath)
 100
 101         vnfr = (
 102             "/rw-project:project[rw-project:name='default']/vnfr-catalog/vnfr")
 103         vnfrs = proxy(RwVnfrYang).get(vnfr, list_obj=True)
 104         for vnfr in vnfrs.vnfr:
 105
 106             if nsr:
 107                 const_vnfr_ids = [const_vnfr.vnfr_id for const_vnfr in nsr.constituent_vnfr_ref]
 108                 if vnfr.id not in const_vnfr_ids:
 109                     continue
 110
 111             vnfd = get_vnfd(vnfr.vnfd.id)
 112             yield vnfd, vnfr
 113
 114     def test_floating_ip(
 115             self, rw_user_proxy, rbac_user_passwd, user_domain, logger,
 116             rw_project_proxy, rw_rbac_int_proxy, descriptors, mgmt_session,
 117             cloud_user, cloud_tenants, vim_ssl_enabled, cloud_host,
 118             fmt_nsd_catalog_xpath):
 119         """test_floating_ip."""
 120         proxy = mgmt_session.proxy
 121         no_of_pool_name_cases = (
 122             len(self.valid_pool_names + self.invalid_pool_names) + 1)
 123         for idx in range(1, no_of_pool_name_cases):
 124             project_name = 'float_project_{}'.format(idx)
 125             user_name = 'float_user_{}'.format(idx)
 126             project_role = 'rw-project:project-admin'
 127             cloud_acc_name = 'cloud_account'
 128             mano.create_user(
 129                 rw_user_proxy, user_name, rbac_user_passwd, user_domain)
 130             mano.assign_project_role_to_user(
 131                 rw_project_proxy, project_role, user_name, project_name,
 132                 user_domain, rw_rbac_int_proxy)
 133
 134             self.create_cloud_account(
 135                 cloud_host, cloud_user, cloud_tenants,
 136                 vim_ssl_enabled, idx, mgmt_session)
 137
 138             for descriptor in descriptors:
 139                 rift.auto.descriptor.onboard(
 140                     mgmt_session, descriptor, project=project_name)
 141
 142             nsd_pxy = mgmt_session.proxy(RwProjectNsdYang)
 143             nsd_catalog = nsd_pxy.get_config(
 144                 fmt_nsd_catalog_xpath.format(project=quoted_key(project_name)))
 145             assert nsd_catalog
 146             nsd = nsd_catalog.nsd[0]
 147             nsr = rift.auto.descriptor.create_nsr(
 148                 cloud_acc_name, nsd.name, nsd)
 149             rwnsr_pxy = mgmt_session.proxy(RwNsrYang)
 150
 151             try:
 152                 rift.auto.descriptor.instantiate_nsr(
 153                     nsr, rwnsr_pxy, logger, project=project_name)
 154             except(Exception):
 155                 continue
 156             for vnfd, vnfr in self.yield_vnfd_vnfr_pairs(proxy):
 157                 if idx > len(self.valid_pool_names):
 158                     assert vnfr.vdur[0].management_ip is None
 159                 else:
 160                     vnfr.vdur[0].management_ip is not None