From b9cac1d63b0c919c961671a1405cbb3ced06942b Mon Sep 17 00:00:00 2001 From: Cédric Ollivier Date: Tue, 9 May 2017 09:50:02 +0200 Subject: Rename vnf_base to vnf MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Change-Id: I0b8b9b2e85717fc92a233e4f7344b3419421778d Signed-off-by: Cédric Ollivier --- functest/core/vnf.py | 226 +++++++++++++++++++++ functest/core/vnf_base.py | 226 --------------------- functest/opnfv_tests/vnf/aaa/aaa.py | 4 +- .../opnfv_tests/vnf/ims/clearwater_ims_base.py | 4 +- functest/opnfv_tests/vnf/ims/orchestra_ims.py | 6 +- functest/tests/unit/core/test_vnf.py | 52 +++++ functest/tests/unit/core/test_vnf_base.py | 52 ----- 7 files changed, 285 insertions(+), 285 deletions(-) create mode 100644 functest/core/vnf.py delete mode 100644 functest/core/vnf_base.py create mode 100644 functest/tests/unit/core/test_vnf.py delete mode 100644 functest/tests/unit/core/test_vnf_base.py (limited to 'functest') diff --git a/functest/core/vnf.py b/functest/core/vnf.py new file mode 100644 index 00000000..bc5bf9a7 --- /dev/null +++ b/functest/core/vnf.py @@ -0,0 +1,226 @@ +#!/usr/bin/env python + +# Copyright (c) 2016 Orange and others. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +import inspect +import logging +import time + +import functest.core.testcase as base +from functest.utils.constants import CONST +import functest.utils.functest_utils as ft_utils +import functest.utils.openstack_utils as os_utils + + +class VnfOnBoardingBase(base.TestCase): + + __logger = logging.getLogger(__name__) + + def __init__(self, **kwargs): + super(VnfOnBoardingBase, self).__init__(**kwargs) + self.repo = kwargs.get('repo', '') + self.cmd = kwargs.get('cmd', '') + self.details = {} + self.result_dir = CONST.__getattribute__('dir_results') + self.details_step_mapping = dict( + deploy_orchestrator='orchestrator', + deploy_vnf='vnf', + test_vnf='test_vnf', + prepare='prepare_env') + self.details['prepare_env'] = {} + self.details['orchestrator'] = {} + self.details['vnf'] = {} + self.details['test_vnf'] = {} + self.images = {} + try: + self.tenant_name = CONST.__getattribute__( + 'vnf_{}_tenant_name'.format(self.case_name)) + self.tenant_description = CONST.__getattribute__( + 'vnf_{}_tenant_description'.format(self.case_name)) + except Exception: + # raise Exception("Unknown VNF case=" + self.case_name) + self.__logger.error("Unknown VNF case={}".format(self.case_name)) + + try: + self.images = CONST.__getattribute__( + 'vnf_{}_tenant_images'.format(self.case_name)) + except Exception: + self.__logger.warn("No tenant image defined for this VNF") + + def execute(self): + self.start_time = time.time() + # Prepare the test (Create Tenant, User, ...) + try: + self.__logger.info("Create VNF Onboarding environment") + self.prepare() + except Exception: + self.__logger.error("Error during VNF Onboarding environment" + "creation", exc_info=True) + return base.TestCase.EX_TESTCASE_FAILED + + # Deploy orchestrator + try: + self.__logger.info("Deploy orchestrator (if necessary)") + orchestrator_ready_time = time.time() + res_orchestrator = self.deploy_orchestrator() + # orchestrator is not mandatory + if res_orchestrator is not None: + self.details['orchestrator']['status'] = ( + res_orchestrator['status']) + self.details['orchestrator']['result'] = ( + res_orchestrator['result']) + self.details['orchestrator']['duration'] = round( + orchestrator_ready_time - self.start_time, 1) + except Exception: + self.__logger.warn("Problem with the Orchestrator", exc_info=True) + + # Deploy VNF + try: + self.__logger.info("Deploy VNF " + self.case_name) + res_deploy_vnf = self.deploy_vnf() + vnf_ready_time = time.time() + self.details['vnf']['status'] = res_deploy_vnf['status'] + self.details['vnf']['result'] = res_deploy_vnf['result'] + self.details['vnf']['duration'] = round( + vnf_ready_time - orchestrator_ready_time, 1) + except Exception: + self.__logger.error("Error during VNF deployment", exc_info=True) + return base.TestCase.EX_TESTCASE_FAILED + + # Test VNF + try: + self.__logger.info("Test VNF") + res_test_vnf = self.test_vnf() + test_vnf_done_time = time.time() + self.details['test_vnf']['status'] = res_test_vnf['status'] + self.details['test_vnf']['result'] = res_test_vnf['result'] + self.details['test_vnf']['duration'] = round( + test_vnf_done_time - vnf_ready_time, 1) + except Exception: + self.__logger.error("Error when running VNF tests", exc_info=True) + return base.TestCase.EX_TESTCASE_FAILED + + # Clean the system + self.clean() + self.stop_time = time.time() + + exit_code = self.parse_results() + self.log_results() + return exit_code + + # prepare state could consist in the creation of the resources + # a dedicated user + # a dedicated tenant + # dedicated images + def prepare(self): + self.creds = os_utils.get_credentials() + self.keystone_client = os_utils.get_keystone_client() + + self.__logger.info( + "Prepare OpenStack plateform(create tenant and user)") + admin_user_id = os_utils.get_user_id(self.keystone_client, + self.creds['username']) + if not admin_user_id: + self.step_failure("Failed to get id of {0}".format( + self.creds['username'])) + + tenant_id = os_utils.get_tenant_id(self.keystone_client, + self.tenant_name) + if not tenant_id: + tenant_id = os_utils.create_tenant(self.keystone_client, + self.tenant_name, + self.tenant_description) + if not tenant_id: + self.step_failure("Failed to get or create {0} tenant".format( + self.tenant_name)) + roles_name = ["admin", "Admin"] + role_id = '' + for role_name in roles_name: + if not role_id: + role_id = os_utils.get_role_id(self.keystone_client, + role_name) + + if not role_id: + self.step_failure("Failed to get id for {0} role".format( + role_name)) + + if not os_utils.add_role_user(self.keystone_client, admin_user_id, + role_id, tenant_id): + self.step_failure("Failed to add {0} on tenant".format( + self.creds['username'])) + + user_id = os_utils.get_or_create_user(self.keystone_client, + self.tenant_name, + self.tenant_name, + tenant_id) + if not user_id: + self.step_failure("Failed to get or create {0} user".format( + self.tenant_name)) + + os_utils.add_role_user(self.keystone_client, user_id, + role_id, tenant_id) + + self.__logger.info("Update OpenStack creds informations") + self.admin_creds = self.creds.copy() + self.admin_creds.update({ + "tenant": self.tenant_name + }) + self.neutron_client = os_utils.get_neutron_client(self.admin_creds) + self.nova_client = os_utils.get_nova_client(self.admin_creds) + self.creds.update({ + "tenant": self.tenant_name, + "username": self.tenant_name, + "password": self.tenant_name, + }) + + # orchestrator is not mandatory to deploy and test VNF + def deploy_orchestrator(self, **kwargs): + pass + + # TODO see how to use built-in exception from releng module + def deploy_vnf(self): + self.__logger.error("VNF must be deployed") + raise Exception("VNF not deployed") + + def test_vnf(self): + self.__logger.error("VNF must be tested") + raise Exception("VNF not tested") + + # clean before openstack clean run + def clean(self): + self.__logger.info("test cleaning") + + def parse_results(self): + exit_code = self.EX_OK + self.result = "PASS" + self.__logger.info(self.details) + # The 2 VNF steps must be OK to get a PASS result + if (self.details['vnf']['status'] is not "PASS" or + self.details['test_vnf']['status'] is not "PASS"): + exit_code = self.EX_RUN_ERROR + self.result = "FAIL" + return exit_code + + def log_results(self): + ft_utils.logger_test_results(self.project_name, + self.case_name, + self.result, + self.details) + + def step_failure(self, error_msg): + part = inspect.stack()[1][3] + self.__logger.error("Step {0} failed: {1}".format(part, error_msg)) + try: + step_name = self.details_step_mapping[part] + part_info = self.details[step_name] + except KeyError: + self.details[part] = {} + part_info = self.details[part] + part_info['status'] = 'FAIL' + part_info['result'] = error_msg + raise Exception(error_msg) diff --git a/functest/core/vnf_base.py b/functest/core/vnf_base.py deleted file mode 100644 index bc5bf9a7..00000000 --- a/functest/core/vnf_base.py +++ /dev/null @@ -1,226 +0,0 @@ -#!/usr/bin/env python - -# Copyright (c) 2016 Orange and others. -# -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 - -import inspect -import logging -import time - -import functest.core.testcase as base -from functest.utils.constants import CONST -import functest.utils.functest_utils as ft_utils -import functest.utils.openstack_utils as os_utils - - -class VnfOnBoardingBase(base.TestCase): - - __logger = logging.getLogger(__name__) - - def __init__(self, **kwargs): - super(VnfOnBoardingBase, self).__init__(**kwargs) - self.repo = kwargs.get('repo', '') - self.cmd = kwargs.get('cmd', '') - self.details = {} - self.result_dir = CONST.__getattribute__('dir_results') - self.details_step_mapping = dict( - deploy_orchestrator='orchestrator', - deploy_vnf='vnf', - test_vnf='test_vnf', - prepare='prepare_env') - self.details['prepare_env'] = {} - self.details['orchestrator'] = {} - self.details['vnf'] = {} - self.details['test_vnf'] = {} - self.images = {} - try: - self.tenant_name = CONST.__getattribute__( - 'vnf_{}_tenant_name'.format(self.case_name)) - self.tenant_description = CONST.__getattribute__( - 'vnf_{}_tenant_description'.format(self.case_name)) - except Exception: - # raise Exception("Unknown VNF case=" + self.case_name) - self.__logger.error("Unknown VNF case={}".format(self.case_name)) - - try: - self.images = CONST.__getattribute__( - 'vnf_{}_tenant_images'.format(self.case_name)) - except Exception: - self.__logger.warn("No tenant image defined for this VNF") - - def execute(self): - self.start_time = time.time() - # Prepare the test (Create Tenant, User, ...) - try: - self.__logger.info("Create VNF Onboarding environment") - self.prepare() - except Exception: - self.__logger.error("Error during VNF Onboarding environment" - "creation", exc_info=True) - return base.TestCase.EX_TESTCASE_FAILED - - # Deploy orchestrator - try: - self.__logger.info("Deploy orchestrator (if necessary)") - orchestrator_ready_time = time.time() - res_orchestrator = self.deploy_orchestrator() - # orchestrator is not mandatory - if res_orchestrator is not None: - self.details['orchestrator']['status'] = ( - res_orchestrator['status']) - self.details['orchestrator']['result'] = ( - res_orchestrator['result']) - self.details['orchestrator']['duration'] = round( - orchestrator_ready_time - self.start_time, 1) - except Exception: - self.__logger.warn("Problem with the Orchestrator", exc_info=True) - - # Deploy VNF - try: - self.__logger.info("Deploy VNF " + self.case_name) - res_deploy_vnf = self.deploy_vnf() - vnf_ready_time = time.time() - self.details['vnf']['status'] = res_deploy_vnf['status'] - self.details['vnf']['result'] = res_deploy_vnf['result'] - self.details['vnf']['duration'] = round( - vnf_ready_time - orchestrator_ready_time, 1) - except Exception: - self.__logger.error("Error during VNF deployment", exc_info=True) - return base.TestCase.EX_TESTCASE_FAILED - - # Test VNF - try: - self.__logger.info("Test VNF") - res_test_vnf = self.test_vnf() - test_vnf_done_time = time.time() - self.details['test_vnf']['status'] = res_test_vnf['status'] - self.details['test_vnf']['result'] = res_test_vnf['result'] - self.details['test_vnf']['duration'] = round( - test_vnf_done_time - vnf_ready_time, 1) - except Exception: - self.__logger.error("Error when running VNF tests", exc_info=True) - return base.TestCase.EX_TESTCASE_FAILED - - # Clean the system - self.clean() - self.stop_time = time.time() - - exit_code = self.parse_results() - self.log_results() - return exit_code - - # prepare state could consist in the creation of the resources - # a dedicated user - # a dedicated tenant - # dedicated images - def prepare(self): - self.creds = os_utils.get_credentials() - self.keystone_client = os_utils.get_keystone_client() - - self.__logger.info( - "Prepare OpenStack plateform(create tenant and user)") - admin_user_id = os_utils.get_user_id(self.keystone_client, - self.creds['username']) - if not admin_user_id: - self.step_failure("Failed to get id of {0}".format( - self.creds['username'])) - - tenant_id = os_utils.get_tenant_id(self.keystone_client, - self.tenant_name) - if not tenant_id: - tenant_id = os_utils.create_tenant(self.keystone_client, - self.tenant_name, - self.tenant_description) - if not tenant_id: - self.step_failure("Failed to get or create {0} tenant".format( - self.tenant_name)) - roles_name = ["admin", "Admin"] - role_id = '' - for role_name in roles_name: - if not role_id: - role_id = os_utils.get_role_id(self.keystone_client, - role_name) - - if not role_id: - self.step_failure("Failed to get id for {0} role".format( - role_name)) - - if not os_utils.add_role_user(self.keystone_client, admin_user_id, - role_id, tenant_id): - self.step_failure("Failed to add {0} on tenant".format( - self.creds['username'])) - - user_id = os_utils.get_or_create_user(self.keystone_client, - self.tenant_name, - self.tenant_name, - tenant_id) - if not user_id: - self.step_failure("Failed to get or create {0} user".format( - self.tenant_name)) - - os_utils.add_role_user(self.keystone_client, user_id, - role_id, tenant_id) - - self.__logger.info("Update OpenStack creds informations") - self.admin_creds = self.creds.copy() - self.admin_creds.update({ - "tenant": self.tenant_name - }) - self.neutron_client = os_utils.get_neutron_client(self.admin_creds) - self.nova_client = os_utils.get_nova_client(self.admin_creds) - self.creds.update({ - "tenant": self.tenant_name, - "username": self.tenant_name, - "password": self.tenant_name, - }) - - # orchestrator is not mandatory to deploy and test VNF - def deploy_orchestrator(self, **kwargs): - pass - - # TODO see how to use built-in exception from releng module - def deploy_vnf(self): - self.__logger.error("VNF must be deployed") - raise Exception("VNF not deployed") - - def test_vnf(self): - self.__logger.error("VNF must be tested") - raise Exception("VNF not tested") - - # clean before openstack clean run - def clean(self): - self.__logger.info("test cleaning") - - def parse_results(self): - exit_code = self.EX_OK - self.result = "PASS" - self.__logger.info(self.details) - # The 2 VNF steps must be OK to get a PASS result - if (self.details['vnf']['status'] is not "PASS" or - self.details['test_vnf']['status'] is not "PASS"): - exit_code = self.EX_RUN_ERROR - self.result = "FAIL" - return exit_code - - def log_results(self): - ft_utils.logger_test_results(self.project_name, - self.case_name, - self.result, - self.details) - - def step_failure(self, error_msg): - part = inspect.stack()[1][3] - self.__logger.error("Step {0} failed: {1}".format(part, error_msg)) - try: - step_name = self.details_step_mapping[part] - part_info = self.details[step_name] - except KeyError: - self.details[part] = {} - part_info = self.details[part] - part_info['status'] = 'FAIL' - part_info['result'] = error_msg - raise Exception(error_msg) diff --git a/functest/opnfv_tests/vnf/aaa/aaa.py b/functest/opnfv_tests/vnf/aaa/aaa.py index 1a484ddf..f23adb5d 100755 --- a/functest/opnfv_tests/vnf/aaa/aaa.py +++ b/functest/opnfv_tests/vnf/aaa/aaa.py @@ -13,10 +13,10 @@ import sys import argparse import functest.core.testcase as testcase -import functest.core.vnf_base as vnf_base +import functest.core.vnf as vnf -class AaaVnf(vnf_base.VnfOnBoardingBase): +class AaaVnf(vnf.VnfOnBoardingBase): logger = logging.getLogger(__name__) diff --git a/functest/opnfv_tests/vnf/ims/clearwater_ims_base.py b/functest/opnfv_tests/vnf/ims/clearwater_ims_base.py index f3bb3012..4bd3dc4f 100644 --- a/functest/opnfv_tests/vnf/ims/clearwater_ims_base.py +++ b/functest/opnfv_tests/vnf/ims/clearwater_ims_base.py @@ -13,12 +13,12 @@ import shutil import requests -import functest.core.vnf_base as vnf_base +import functest.core.vnf as vnf from functest.utils.constants import CONST import functest.utils.functest_utils as ft_utils -class ClearwaterOnBoardingBase(vnf_base.VnfOnBoardingBase): +class ClearwaterOnBoardingBase(vnf.VnfOnBoardingBase): def __init__(self, **kwargs): self.logger = logging.getLogger(__name__) diff --git a/functest/opnfv_tests/vnf/ims/orchestra_ims.py b/functest/opnfv_tests/vnf/ims/orchestra_ims.py index 95751d47..128ea366 100755 --- a/functest/opnfv_tests/vnf/ims/orchestra_ims.py +++ b/functest/opnfv_tests/vnf/ims/orchestra_ims.py @@ -9,15 +9,15 @@ import json import logging +import os import socket import sys import time import yaml -import functest.core.vnf_base as vnf_base +import functest.core.vnf as vnf import functest.utils.functest_utils as ft_utils import functest.utils.openstack_utils as os_utils -import os from functest.utils.constants import CONST from org.openbaton.cli.agents.agents import MainAgent @@ -76,7 +76,7 @@ def servertest(host, port): return True -class ImsVnf(vnf_base.VnfOnBoardingBase): +class ImsVnf(vnf.VnfOnBoardingBase): def __init__(self, project='functest', case_name='orchestra_ims', repo='', cmd=''): diff --git a/functest/tests/unit/core/test_vnf.py b/functest/tests/unit/core/test_vnf.py new file mode 100644 index 00000000..a0e73308 --- /dev/null +++ b/functest/tests/unit/core/test_vnf.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python + +# Copyright (c) 2016 Orange and others. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +import logging +import unittest + +from functest.core import vnf + + +class VnfBaseTesting(unittest.TestCase): + + logging.disable(logging.CRITICAL) + + def setUp(self): + self.test = vnf.VnfOnBoardingBase(project='functest', + case_name='aaa') + self.test.project = "functest" + self.test.start_time = "1" + self.test.stop_time = "5" + self.test.result = "" + self.test.details = {"orchestrator": {"status": "PASS", + "result": "", + "duration": 20}, + "vnf": {"status": "PASS", + "result": "", + "duration": 15}, + "test_vnf": {"status": "FAIL", + "result": "", + "duration": 5}} + + def test_deploy_vnf_unimplemented(self): + with self.assertRaises(Exception) as context: + self.test.deploy_vnf() + self.assertTrue('VNF not deployed' in context.exception) + + def test_test_vnf_unimplemented(self): + with self.assertRaises(Exception) as context: + self.test.test_vnf()() + self.assertTrue('VNF not tested' in context.exception) + + def test_parse_results(self): + self.assertNotEqual(self.test.parse_results(), 0) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/functest/tests/unit/core/test_vnf_base.py b/functest/tests/unit/core/test_vnf_base.py deleted file mode 100644 index 540cf610..00000000 --- a/functest/tests/unit/core/test_vnf_base.py +++ /dev/null @@ -1,52 +0,0 @@ -#!/usr/bin/env python - -# Copyright (c) 2016 Orange and others. -# -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 - -import logging -import unittest - -from functest.core import vnf_base - - -class VnfBaseTesting(unittest.TestCase): - - logging.disable(logging.CRITICAL) - - def setUp(self): - self.test = vnf_base.VnfOnBoardingBase(project='functest', - case_name='aaa') - self.test.project = "functest" - self.test.start_time = "1" - self.test.stop_time = "5" - self.test.result = "" - self.test.details = {"orchestrator": {"status": "PASS", - "result": "", - "duration": 20}, - "vnf": {"status": "PASS", - "result": "", - "duration": 15}, - "test_vnf": {"status": "FAIL", - "result": "", - "duration": 5}} - - def test_deploy_vnf_unimplemented(self): - with self.assertRaises(Exception) as context: - self.test.deploy_vnf() - self.assertTrue('VNF not deployed' in context.exception) - - def test_test_vnf_unimplemented(self): - with self.assertRaises(Exception) as context: - self.test.test_vnf()() - self.assertTrue('VNF not tested' in context.exception) - - def test_parse_results(self): - self.assertNotEqual(self.test.parse_results(), 0) - - -if __name__ == "__main__": - unittest.main(verbosity=2) -- cgit 1.2.3-korg