X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=test%2Ftest_cirros.py;fp=test%2Ftest_cirros.py;h=0000000000000000000000000000000000000000;hb=e5245336823411b71968e72c32b3bf33cc07638f;hp=1451d1139332d5d8496aa896b05059925f75902f;hpb=84fe31f632166d2fbf566968bd25aa9545e96fdc;p=osm%2Fdevops.git diff --git a/test/test_cirros.py b/test/test_cirros.py deleted file mode 100644 index 1451d113..00000000 --- a/test/test_cirros.py +++ /dev/null @@ -1,185 +0,0 @@ -#!/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