#!/usr/bin/env python3 # Copyright 2016 RIFT.IO Inc # Copyright 2016 Telefónica Investigación y Desarrollo S.A.U. # # 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 json import logging import pytest import requests import subprocess import time import yaml import certs logger = logging.getLogger(__name__) def chomp_json(string): return ''.join([line.strip() for line in string.splitlines()]) @pytest.fixture(scope='session') def session(request, so_user, so_pass): client_session = requests.session() client_session.auth = (so_user, so_pass) client_session.headers = { "Accept": "application/vnd.yang.data+json", "Content-Type": "application/vnd.yang.data+json", } client_session.verify = False _, cert, key = certs.get_bootstrap_cert_and_key() client_session.cert = (cert, key) return client_session @pytest.fixture(scope='session') def fetch_packages(): """ Fetch NSD/VNFD packages""" wget_command = 'wget --no-parent -r https://osm-download.etsi.org/ftp/osm-1.0-one/vnf-packages' subprocess.check_call(wget_command, shell=True) @pytest.fixture(scope='session') def rest_endpoint(so_host, so_port): return 'https://{host}:{port}'.format(host=so_host, port=so_port) def test_so_started(session, rest_endpoint): ''' Get contents of /vcs/info/components and verify all components have started successfully ''' uri = "{endpoint}/api/operational/vcs/info/components".format(endpoint=rest_endpoint) response = session.request("GET", uri) vcs_info = json.loads(response.text) for component in vcs_info['rw-base:components']['component_info']: assert component['state'] == 'RUNNING' @pytest.fixture(scope='session') def cirros_vnfd_pkg(package_location): '''cirros vnfd package location''' return "%s/cirros_vnf.tar.gz" % (package_location) @pytest.fixture(scope='session') def cirros_nsd_pkg(package_location): '''cirros nsd package location''' return "%s/cirros_2vnf_ns.tar.gz" % (package_location) def test_onboard_cirros_descriptors(session, so_host, cirros_vnfd_pkg, cirros_nsd_pkg, rest_endpoint): ''' Onboard Cirros NSD/VNFD descriptors ''' onboard_command = 'onboard_pkg -s {host} -u {cirros_vnfd_pkg}'.format( host=so_host, cirros_vnfd_pkg=cirros_vnfd_pkg, ) subprocess.check_call(onboard_command, shell=True) onboard_command = 'onboard_pkg -s {host} -u {cirros_nsd_pkg}'.format( host=so_host, cirros_nsd_pkg=cirros_nsd_pkg, ) subprocess.check_call(onboard_command, shell=True) def test_instantiate_cirros(session, so_host, data_center_id, rest_endpoint): ''' Instantiate an instance of cirros from descriptors ''' uri = "{endpoint}/api/operational/nsd-catalog".format(endpoint=rest_endpoint) response = session.request("GET", uri) catalog = json.loads(response.text) cirros_nsd = None for nsd in catalog['nsd:nsd-catalog']['nsd']: if nsd['name'] == 'cirros_2vnf_nsd': cirros_nsd = nsd break assert cirros_nsd is not None instantiate_command = 'onboard_pkg -s {host} -i instance-0 -d {nsd_id} -D {data_center_id}'.format( host=so_host, nsd_id=cirros_nsd['id'], data_center_id=data_center_id ) subprocess.check_call(instantiate_command, shell=True) def wait_for_cirros_ns(instance_name, timeout=600, retry_interval=5): start_time = time.time() while True: uri = "{endpoint}/api/operational/ns-instance-opdata".format(endpoint=rest_endpoint) response = session.request("GET", uri) print(response, response.text) opdata = json.loads(response.text) nsr = None for instance in opdata['nsr:ns-instance-opdata']['nsr']: if instance['name-ref'] == instance_name: nsr = instance assert nsr is not None, response.text assert nsr['operational-status'] not in ['failed'] assert nsr['config-status'] not in ['failed'] if nsr['operational-status'] in ['running'] and nsr['config-status'] in ['configured']: break time_elapsed = time.time() - start_time time_remaining = timeout - time_elapsed assert time_remaining > 0 time.sleep(min(time_remaining, retry_interval)) wait_for_cirros_ns('instance-0') @pytest.fixture(scope='session') def test_add_datacenter(name, url, vim_type, tenant_id=None, tenant_name=None, user=None, password=None, description=None, config=None): ''' Add a datacenter to RO ''' onboard_command = \ 'lxc exec RO --env OPENMANO_TENANT=osm -- openmano datacenter-create "{name}" "{url}" --type={vimtype}'.format( name=name, url=url, vimtype=vim_type) if description: onboard_command += ' --description="{}"'.format(description) out = subprocess.check_output(onboard_command, shell=True) assert out datacenter_id = out.split()[0] onboard_command = 'lxc exec RO --env OPENMANO_TENANT=osm -- openmano datacenter-attach {id}'.format( id=datacenter_id) if tenant_id: onboard_command += " --vim_tenant_id=" + tenant_id if tenant_name: onboard_command += " --vim_tenant_name=" + tenant_name if user: onboard_command += " --user=" + user if password: onboard_command += " --password=" + password if config: onboard_command += " --config=" + yaml.safe_dump(config) subprocess.check_call(onboard_command, shell=True) return datacenter_id @pytest.fixture(scope='session') def get_datacenter_id(name): ''' Get the id of a datacenter ''' onboard_command = \ 'lxc exec RO --env OPENMANO_TENANT=osm -- openmano datacenter-list {name}'.format(name=name) out = subprocess.check_output(onboard_command, shell=True) assert(out) datacenter_id = out.split()[0] return datacenter_id