test/test_pingpong_osm.py

   1 #   Copyright 2016 RIFT.IO Inc
   2 #   Copyright 2016 Telefónica Investigación y Desarrollo S.A.U.
   3 #
   4 #   Licensed under the Apache License, Version 2.0 (the "License");
   5 #   you may not use this file except in compliance with the License.
   6 #   You may obtain a copy of the License at
   7 #
   8 #       http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #   Unless required by applicable law or agreed to in writing, software
  11 #   distributed under the License is distributed on an "AS IS" BASIS,
  12 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 #   See the License for the specific language governing permissions and
  14 #   limitations under the License.
  15
  16 #!/usr/bin/env python3
  17
  18 import json
  19 import logging
  20 import os
  21 import pytest
  22 import requests
  23 import requests_toolbelt
  24 import subprocess
  25 import time
  26
  27 import certs
  28
  29 logger = logging.getLogger(__name__)
  30
  31 def chomp_json(string):
  32     return ''.join([line.strip() for line in string.splitlines()])
  33
  34 @pytest.fixture(scope='session')
  35 def session(request, so_user, so_pass):
  36     client_session = requests.session()
  37     client_session.auth = (so_user, so_pass)
  38     client_session.headers = {
  39         "Accept" : "application/vnd.yang.data+json",
  40         "Content-Type" : "application/vnd.yang.data+json",
  41     }
  42     client_session.verify = False
  43     _, cert, key = certs.get_bootstrap_cert_and_key()
  44     client_session.cert = (cert, key)
  45     return client_session
  46
  47 @pytest.fixture(scope='session')
  48 def rest_endpoint(so_host, so_port):
  49     return 'https://{host}:{port}'.format(host=so_host, port=so_port)
  50
  51 def test_so_started(session, rest_endpoint):
  52     ''' Get contents of /vcs/info/components and verify all components have started successfully
  53     '''
  54     uri = "{endpoint}/api/operational/vcs/info/components".format(endpoint=rest_endpoint)
  55     response = session.request("GET", uri)
  56     vcs_info = json.loads(response.text)
  57     for component in vcs_info['rw-base:components']['component_info']:
  58         assert component['state'] == 'RUNNING'
  59
  60 def test_configure_vim_account(session, rest_endpoint, vim_type, vim_host, vim_user, vim_tenant, vim_pass):
  61     ''' Configure an openstack vim account
  62     '''
  63     uri = "{endpoint}/api/config/cloud/account".format(
  64         endpoint=rest_endpoint,
  65     )
  66     account = '''
  67     {
  68         "account":{
  69             "name":"vim-0",
  70             "account-type":"openstack",
  71             "openstack":{
  72                 "admin":"True",
  73                 "key": "%s",
  74                 "secret": "%s",
  75                 "auth_url": "http://%s:5000/v3/",
  76                 "tenant": "%s",
  77                 "mgmt-network": "private"
  78             }
  79         }
  80     }
  81     ''' % (vim_user, vim_pass, vim_host, vim_tenant)
  82     response = session.request("POST", uri, data=chomp_json(account))
  83
  84     uri = "{endpoint}/api/operational/cloud/account".format(endpoint=rest_endpoint)
  85     response = session.request("GET", uri)
  86     assert 'error' not in response.text
  87     vim_accounts = json.loads(response.text)
  88     assert vim_accounts
  89
  90     found_account = False
  91     for account in vim_accounts['rw-cloud:account']:
  92         if account['name'] == 'vim-0':
  93             found_account = True
  94             break
  95     assert found_account == True
  96
  97
  98 @pytest.fixture(scope='session')
  99 def ping_pkg(package_location):
 100     '''ping vnfd package location'''
 101     return "%s/ping_vnfd.tar.gz" % (package_location)
 102
 103
 104 @pytest.fixture(scope='session')
 105 def pong_pkg(package_location):
 106     '''pong vnfd package location'''
 107     return "%s/pong_vnfd.tar.gz" % (package_location)
 108
 109
 110 @pytest.fixture(scope='session')
 111 def ping_pong_pkg(package_location):
 112     '''ping pong nsd package location'''
 113     return "%s/ping_pong_nsd.tar.gz" % (package_location)
 114
 115
 116 def test_onboard_pingpong_descriptors(session, so_host, ping_pkg, pong_pkg, ping_pong_pkg, rest_endpoint):
 117     ''' Onboard ping_pong descriptors
 118     '''
 119     onboard_command = 'onboard_pkg -s {host} -u {ping_pkg}'.format(
 120         host=so_host,
 121         ping_pkg=ping_pkg,
 122     )
 123     subprocess.check_call(onboard_command, shell=True)
 124
 125     onboard_command = 'onboard_pkg -s {host} -u {pong_pkg}'.format(
 126         host=so_host,
 127         pong_pkg=pong_pkg,
 128     )
 129     subprocess.check_call(onboard_command, shell=True)
 130
 131     onboard_command = 'onboard_pkg -s {host} -u {ping_pong_pkg}'.format(
 132         host=so_host,
 133         ping_pong_pkg=ping_pong_pkg,
 134     )
 135     subprocess.check_call(onboard_command, shell=True)
 136
 137
 138 def test_instantiate_ping_pong(session, so_host, rest_endpoint):
 139     ''' Instantiate an instance of ping pong from descriptors
 140     '''
 141     uri = "{endpoint}/api/operational/nsd-catalog".format(endpoint=rest_endpoint)
 142     response = session.request("GET", uri)
 143     catalog = json.loads(response.text)
 144     ping_pong_nsd = None
 145     for nsd in catalog['nsd:nsd-catalog']['nsd']:
 146         if nsd['name'] == 'ping_pong_nsd':
 147             ping_pong_nsd = nsd
 148             break
 149     assert ping_pong_nsd is not None
 150
 151     instantiate_command = 'onboard_pkg -s {host} -i instance-0 -d {nsd_id} -c vim-0'.format(
 152         host=so_host,
 153         nsd_id=ping_pong_nsd['id']
 154     )
 155     subprocess.check_call(instantiate_command, shell=True)
 156
 157     def wait_for_ping_pong(instance_name, timeout=600, retry_interval=5):
 158         start_time = time.time()
 159         while True:
 160             uri = "{endpoint}/api/operational/ns-instance-opdata".format(endpoint=rest_endpoint)
 161             response = session.request("GET", uri)
 162             print(response, response.text)
 163             opdata = json.loads(response.text)
 164
 165             nsr = None
 166             for instance in opdata['nsr:ns-instance-opdata']['nsr']:
 167                 if instance['name-ref'] == instance_name:
 168                     nsr = instance
 169
 170             assert nsr is not None, response.text
 171             assert nsr['operational-status'] not in ['failed']
 172             assert nsr['config-status'] not in ['failed']
 173
 174             if nsr['operational-status'] in ['running'] and nsr['config-status'] in ['configured']:
 175                 break
 176
 177             time_elapsed = time.time() - start_time
 178             time_remaining = timeout - time_elapsed
 179             assert time_remaining > 0
 180             time.sleep(min(time_remaining, retry_interval))
 181
 182     wait_for_ping_pong('instance-0')