diff options
Diffstat (limited to 'functest/core')
| -rw-r--r-- | functest/core/__init__.py | 0 | ||||
| -rw-r--r-- | functest/core/feature.py | 133 | ||||
| -rw-r--r-- | functest/core/robotframework.py | 126 | ||||
| -rw-r--r-- | functest/core/testcase.py | 227 | ||||
| -rw-r--r-- | functest/core/unit.py | 92 | ||||
| -rw-r--r-- | functest/core/vnf.py | 205 |
6 files changed, 0 insertions, 783 deletions
diff --git a/functest/core/__init__.py b/functest/core/__init__.py deleted file mode 100644 index e69de29b..00000000 --- a/functest/core/__init__.py +++ /dev/null diff --git a/functest/core/feature.py b/functest/core/feature.py deleted file mode 100644 index 65fd5a08..00000000 --- a/functest/core/feature.py +++ /dev/null @@ -1,133 +0,0 @@ -#!/usr/bin/env python - -# Copyright (c) 2016 ZTE Corp 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 - -"""Define the parent classes of all Functest Features. - -Feature is considered as TestCase offered by Third-party. It offers -helpers to run any python method or any bash command. -""" - -import logging -import subprocess -import time - -import functest.core.testcase as base - -__author__ = ("Serena Feng <feng.xiaowei@zte.com.cn>, " - "Cedric Ollivier <cedric.ollivier@orange.com>") - - -class Feature(base.TestCase): - """Base model for single feature.""" - - __logger = logging.getLogger(__name__) - dir_results = "/home/opnfv/functest/results" - - def __init__(self, **kwargs): - super(Feature, self).__init__(**kwargs) - self.result_file = "{}/{}.log".format(self.dir_results, self.case_name) - try: - module = kwargs['run']['module'] - self.logger = logging.getLogger(module) - except KeyError: - self.__logger.warning( - "Cannot get module name %s. Using %s as fallback", - kwargs, self.case_name) - self.logger = logging.getLogger(self.case_name) - handler = logging.StreamHandler() - handler.setLevel(logging.WARN) - self.logger.addHandler(handler) - handler = logging.FileHandler(self.result_file) - handler.setLevel(logging.DEBUG) - self.logger.addHandler(handler) - formatter = logging.Formatter( - '%(asctime)s - %(name)s - %(levelname)s - %(message)s') - handler.setFormatter(formatter) - self.logger.addHandler(handler) - - def execute(self, **kwargs): - """Execute the Python method. - - The subclasses must override the default implementation which - is false on purpose. - - The new implementation must return 0 if success or anything - else if failure. - - Args: - kwargs: Arbitrary keyword arguments. - - Returns: - -1. - """ - # pylint: disable=unused-argument,no-self-use - return -1 - - def run(self, **kwargs): - """Run the feature. - - It allows executing any Python method by calling execute(). - - It sets the following attributes required to push the results - to DB: - - * result, - * start_time, - * stop_time. - - It doesn't fulfill details when pushing the results to the DB. - - Args: - kwargs: Arbitrary keyword arguments. - - Returns: - TestCase.EX_OK if execute() returns 0, - TestCase.EX_RUN_ERROR otherwise. - """ - self.start_time = time.time() - exit_code = base.TestCase.EX_RUN_ERROR - self.result = 0 - try: - if self.execute(**kwargs) == 0: - exit_code = base.TestCase.EX_OK - self.result = 100 - except Exception: # pylint: disable=broad-except - self.__logger.exception("%s FAILED", self.project_name) - self.__logger.info("Test result is stored in '%s'", self.result_file) - self.stop_time = time.time() - return exit_code - - -class BashFeature(Feature): - """Class designed to run any bash command.""" - - __logger = logging.getLogger(__name__) - - def execute(self, **kwargs): - """Execute the cmd passed as arg - - Args: - kwargs: Arbitrary keyword arguments. - - Returns: - 0 if cmd returns 0, - -1 otherwise. - """ - ret = -1 - try: - cmd = kwargs["cmd"] - with open(self.result_file, 'w+') as f_stdout: - proc = subprocess.Popen(cmd.split(), stdout=f_stdout, - stderr=subprocess.STDOUT) - ret = proc.wait() - if ret != 0: - self.__logger.error("Execute command: %s failed", cmd) - except KeyError: - self.__logger.error("Please give cmd as arg. kwargs: %s", kwargs) - return ret diff --git a/functest/core/robotframework.py b/functest/core/robotframework.py deleted file mode 100644 index 54574a68..00000000 --- a/functest/core/robotframework.py +++ /dev/null @@ -1,126 +0,0 @@ -#!/usr/bin/env python - -# Copyright (c) 2017 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 - -"""Define classes required to run any Robot suites.""" - -from __future__ import division - -import errno -import logging -import os - -import robot.api -from robot.errors import RobotError -import robot.run -from robot.utils.robottime import timestamp_to_secs -from six import StringIO - -from functest.core import testcase - -__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>" - - -class ResultVisitor(robot.api.ResultVisitor): - """Visitor to get result details.""" - - def __init__(self): - self._data = [] - - def visit_test(self, test): - output = {} - output['name'] = test.name - output['parent'] = test.parent.name - output['status'] = test.status - output['starttime'] = test.starttime - output['endtime'] = test.endtime - output['critical'] = test.critical - output['text'] = test.message - output['elapsedtime'] = test.elapsedtime - self._data.append(output) - - def get_data(self): - """Get the details of the result.""" - return self._data - - -class RobotFramework(testcase.TestCase): - """RobotFramework runner.""" - - __logger = logging.getLogger(__name__) - dir_results = "/home/opnfv/functest/results" - - def __init__(self, **kwargs): - self.res_dir = os.path.join(self.dir_results, 'robot') - self.xml_file = os.path.join(self.res_dir, 'output.xml') - super(RobotFramework, self).__init__(**kwargs) - - def parse_results(self): - """Parse output.xml and get the details in it.""" - result = robot.api.ExecutionResult(self.xml_file) - visitor = ResultVisitor() - result.visit(visitor) - try: - self.result = 100 * ( - result.suite.statistics.critical.passed / - result.suite.statistics.critical.total) - except ZeroDivisionError: - self.__logger.error("No test has been run") - self.start_time = timestamp_to_secs(result.suite.starttime) - self.stop_time = timestamp_to_secs(result.suite.endtime) - self.details = {} - self.details['description'] = result.suite.name - self.details['tests'] = visitor.get_data() - - def run(self, **kwargs): - """Run the RobotFramework suites - - Here are the steps: - * create the output directories if required, - * get the results in output.xml, - * delete temporary files. - - Args: - kwargs: Arbitrary keyword arguments. - - Returns: - EX_OK if all suites ran well. - EX_RUN_ERROR otherwise. - """ - try: - suites = kwargs["suites"] - variable = kwargs.get("variable", []) - variablefile = kwargs.get("variablefile", []) - except KeyError: - self.__logger.exception("Mandatory args were not passed") - return self.EX_RUN_ERROR - try: - os.makedirs(self.res_dir) - except OSError as ex: - if ex.errno != errno.EEXIST: - self.__logger.exception("Cannot create %s", self.res_dir) - return self.EX_RUN_ERROR - except Exception: # pylint: disable=broad-except - self.__logger.exception("Cannot create %s", self.res_dir) - return self.EX_RUN_ERROR - stream = StringIO() - robot.run(*suites, variable=variable, variablefile=variablefile, - output=self.xml_file, log='NONE', - report='NONE', stdout=stream) - self.__logger.info("\n" + stream.getvalue()) - self.__logger.info("Results were successfully generated") - try: - self.parse_results() - self.__logger.info("Results were successfully parsed") - except RobotError as ex: - self.__logger.error("Run suites before publishing: %s", ex.message) - return self.EX_RUN_ERROR - except Exception: # pylint: disable=broad-except - self.__logger.exception("Cannot parse results") - return self.EX_RUN_ERROR - return self.EX_OK diff --git a/functest/core/testcase.py b/functest/core/testcase.py deleted file mode 100644 index e8bb1409..00000000 --- a/functest/core/testcase.py +++ /dev/null @@ -1,227 +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 - -"""Define the parent class of all Functest TestCases.""" - -from datetime import datetime -import json -import logging -import os -import re -import requests - -from functest.utils import decorators -from functest.utils import env - - -import prettytable - - -__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>" - - -class TestCase(object): - """Base model for single test case.""" - - EX_OK = os.EX_OK - """everything is OK""" - - EX_RUN_ERROR = os.EX_SOFTWARE - """run() failed""" - - EX_PUSH_TO_DB_ERROR = os.EX_SOFTWARE - 1 - """push_to_db() failed""" - - EX_TESTCASE_FAILED = os.EX_SOFTWARE - 2 - """results are false""" - - _job_name_rule = "(dai|week)ly-(.+?)-[0-9]*" - _headers = {'Content-Type': 'application/json'} - __logger = logging.getLogger(__name__) - - def __init__(self, **kwargs): - self.details = {} - self.project_name = kwargs.get('project_name', 'functest') - self.case_name = kwargs.get('case_name', '') - self.criteria = kwargs.get('criteria', 100) - self.result = 0 - self.start_time = 0 - self.stop_time = 0 - - def __str__(self): - try: - assert self.project_name - assert self.case_name - result = 'PASS' if(self.is_successful( - ) == TestCase.EX_OK) else 'FAIL' - msg = prettytable.PrettyTable( - header_style='upper', padding_width=5, - field_names=['test case', 'project', 'duration', - 'result']) - msg.add_row([self.case_name, self.project_name, - self.get_duration(), result]) - return msg.get_string() - except AssertionError: - self.__logger.error("We cannot print invalid objects") - return super(TestCase, self).__str__() - - def get_duration(self): - """Return the duration of the test case. - - Returns: - duration if start_time and stop_time are set - "XX:XX" otherwise. - """ - try: - assert self.start_time - assert self.stop_time - if self.stop_time < self.start_time: - return "XX:XX" - return "{0[0]:02.0f}:{0[1]:02.0f}".format(divmod( - self.stop_time - self.start_time, 60)) - except Exception: # pylint: disable=broad-except - self.__logger.error("Please run test before getting the duration") - return "XX:XX" - - def is_successful(self): - """Interpret the result of the test case. - - It allows getting the result of TestCase. It completes run() - which only returns the execution status. - - It can be overriden if checking result is not suitable. - - Returns: - TestCase.EX_OK if result is 'PASS'. - TestCase.EX_TESTCASE_FAILED otherwise. - """ - try: - assert self.criteria - assert self.result is not None - if (not isinstance(self.result, str) and - not isinstance(self.criteria, str)): - if self.result >= self.criteria: - return TestCase.EX_OK - else: - # Backward compatibility - # It must be removed as soon as TestCase subclasses - # stop setting result = 'PASS' or 'FAIL'. - # In this case criteria is unread. - self.__logger.warning( - "Please update result which must be an int!") - if self.result == 'PASS': - return TestCase.EX_OK - except AssertionError: - self.__logger.error("Please run test before checking the results") - return TestCase.EX_TESTCASE_FAILED - - def run(self, **kwargs): - """Run the test case. - - It allows running TestCase and getting its execution - status. - - The subclasses must override the default implementation which - is false on purpose. - - The new implementation must set the following attributes to - push the results to DB: - - * result, - * start_time, - * stop_time. - - Args: - kwargs: Arbitrary keyword arguments. - - Returns: - TestCase.EX_RUN_ERROR. - """ - # pylint: disable=unused-argument - self.__logger.error("Run must be implemented") - return TestCase.EX_RUN_ERROR - - @decorators.can_dump_request_to_file - def push_to_db(self): - """Push the results of the test case to the DB. - - It allows publishing the results and checking the status. - - It could be overriden if the common implementation is not - suitable. - - The following attributes must be set before pushing the results to DB: - - * project_name, - * case_name, - * result, - * start_time, - * stop_time. - - The next vars must be set in env: - - * TEST_DB_URL, - * INSTALLER_TYPE, - * DEPLOY_SCENARIO, - * NODE_NAME, - * BUILD_TAG. - - Returns: - TestCase.EX_OK if results were pushed to DB. - TestCase.EX_PUSH_TO_DB_ERROR otherwise. - """ - try: - assert self.project_name - assert self.case_name - assert self.start_time - assert self.stop_time - url = env.get('TEST_DB_URL') - data = {"project_name": self.project_name, - "case_name": self.case_name, - "details": self.details} - data["installer"] = env.get('INSTALLER_TYPE') - data["scenario"] = env.get('DEPLOY_SCENARIO') - data["pod_name"] = env.get('NODE_NAME') - data["build_tag"] = env.get('BUILD_TAG') - data["criteria"] = 'PASS' if self.is_successful( - ) == TestCase.EX_OK else 'FAIL' - data["start_date"] = datetime.fromtimestamp( - self.start_time).strftime('%Y-%m-%d %H:%M:%S') - data["stop_date"] = datetime.fromtimestamp( - self.stop_time).strftime('%Y-%m-%d %H:%M:%S') - try: - data["version"] = re.search( - TestCase._job_name_rule, - env.get('BUILD_TAG')).group(2) - except Exception: # pylint: disable=broad-except - data["version"] = "unknown" - req = requests.post( - url, data=json.dumps(data, sort_keys=True), - headers=self._headers) - req.raise_for_status() - self.__logger.info( - "The results were successfully pushed to DB %s", url) - except AssertionError: - self.__logger.exception( - "Please run test before publishing the results") - return TestCase.EX_PUSH_TO_DB_ERROR - except requests.exceptions.HTTPError: - self.__logger.exception("The HTTP request raises issues") - return TestCase.EX_PUSH_TO_DB_ERROR - except Exception: # pylint: disable=broad-except - self.__logger.exception("The results cannot be pushed to DB") - return TestCase.EX_PUSH_TO_DB_ERROR - return TestCase.EX_OK - - def clean(self): - """Clean the resources. - - It can be overriden if resources must be deleted after - running the test case. - """ diff --git a/functest/core/unit.py b/functest/core/unit.py deleted file mode 100644 index 61b5a58d..00000000 --- a/functest/core/unit.py +++ /dev/null @@ -1,92 +0,0 @@ -#!/usr/bin/env python - -# Copyright (c) 2016 Cable Television Laboratories, Inc. 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 - -"""Define the parent class to run unittest.TestSuite as TestCase.""" - -from __future__ import division - -import logging -import time -import unittest - -import six - -from functest.core import testcase - -__author__ = ("Steven Pisarski <s.pisarski@cablelabs.com>, " - "Cedric Ollivier <cedric.ollivier@orange.com>") - - -class Suite(testcase.TestCase): - """Base model for running unittest.TestSuite.""" - - __logger = logging.getLogger(__name__) - - def __init__(self, **kwargs): - super(Suite, self).__init__(**kwargs) - self.suite = None - - def run(self, **kwargs): - """Run the test suite. - - It allows running any unittest.TestSuite and getting its - execution status. - - By default, it runs the suite defined as instance attribute. - It can be overriden by passing name as arg. It must - conform with TestLoader.loadTestsFromName(). - - It sets the following attributes required to push the results - to DB: - - * result, - * start_time, - * stop_time, - * details. - - Args: - kwargs: Arbitrary keyword arguments. - - Returns: - TestCase.EX_OK if any TestSuite has been run, - TestCase.EX_RUN_ERROR otherwise. - """ - try: - name = kwargs["name"] - try: - self.suite = unittest.TestLoader().loadTestsFromName(name) - except ImportError: - self.__logger.error("Can not import %s", name) - return testcase.TestCase.EX_RUN_ERROR - except KeyError: - pass - try: - assert self.suite - self.start_time = time.time() - stream = six.StringIO() - result = unittest.TextTestRunner( - stream=stream, verbosity=2).run(self.suite) - self.__logger.debug("\n\n%s", stream.getvalue()) - self.stop_time = time.time() - self.details = { - "testsRun": result.testsRun, - "failures": len(result.failures), - "errors": len(result.errors), - "stream": stream.getvalue()} - self.result = 100 * ( - (result.testsRun - (len(result.failures) + - len(result.errors))) / - result.testsRun) - return testcase.TestCase.EX_OK - except AssertionError: - self.__logger.error("No suite is defined") - return testcase.TestCase.EX_RUN_ERROR - except ZeroDivisionError: - self.__logger.error("No test has been run") - return testcase.TestCase.EX_RUN_ERROR diff --git a/functest/core/vnf.py b/functest/core/vnf.py deleted file mode 100644 index cf20492b..00000000 --- a/functest/core/vnf.py +++ /dev/null @@ -1,205 +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 - -"""Define the parent class of all VNF TestCases.""" - -import logging -import time -import uuid - -from snaps.config.user import UserConfig -from snaps.config.project import ProjectConfig -from snaps.openstack.create_user import OpenStackUser -from snaps.openstack.create_project import OpenStackProject -from snaps.openstack.tests import openstack_tests - -from functest.core import testcase -from functest.utils import constants - -__author__ = ("Morgan Richomme <morgan.richomme@orange.com>, " - "Valentin Boucher <valentin.boucher@orange.com>") - - -class VnfPreparationException(Exception): - """Raise when VNF preparation cannot be executed.""" - - -class OrchestratorDeploymentException(Exception): - """Raise when orchestrator cannot be deployed.""" - - -class VnfDeploymentException(Exception): - """Raise when VNF cannot be deployed.""" - - -class VnfTestException(Exception): - """Raise when VNF cannot be tested.""" - - -class VnfOnBoarding(testcase.TestCase): - # pylint: disable=too-many-instance-attributes - """Base model for VNF test cases.""" - - __logger = logging.getLogger(__name__) - - def __init__(self, **kwargs): - super(VnfOnBoarding, self).__init__(**kwargs) - self.uuid = uuid.uuid4() - self.user_name = "{}-{}".format(self.case_name, self.uuid) - self.tenant_name = "{}-{}".format(self.case_name, self.uuid) - self.snaps_creds = {} - self.created_object = [] - self.os_project = None - self.tenant_description = "Created by OPNFV Functest: {}".format( - self.case_name) - - def run(self, **kwargs): - """ - Run of the VNF test case: - - * Deploy an orchestrator if needed (e.g. heat, cloudify, ONAP,...), - * Deploy the VNF, - * Perform tests on the VNF - - A VNF test case is successfull when the 3 steps are PASS - If one of the step is FAIL, the test case is FAIL - - Returns: - TestCase.EX_OK if result is 'PASS'. - TestCase.EX_TESTCASE_FAILED otherwise. - """ - self.start_time = time.time() - - try: - self.prepare() - if (self.deploy_orchestrator() and - self.deploy_vnf() and - self.test_vnf()): - self.stop_time = time.time() - # Calculation with different weight depending on the steps TODO - self.result = 100 - return testcase.TestCase.EX_OK - self.result = 0 - self.stop_time = time.time() - return testcase.TestCase.EX_TESTCASE_FAILED - except Exception: # pylint: disable=broad-except - self.stop_time = time.time() - self.__logger.exception("Exception on VNF testing") - return testcase.TestCase.EX_TESTCASE_FAILED - - def prepare(self): - """ - Prepare the environment for VNF testing: - - * Creation of a user, - * Creation of a tenant, - * Allocation admin role to the user on this tenant - - Returns base.TestCase.EX_OK if preparation is successfull - - Raise VnfPreparationException in case of problem - """ - try: - self.__logger.info( - "Prepare VNF: %s, description: %s", self.case_name, - self.tenant_description) - snaps_creds = openstack_tests.get_credentials( - os_env_file=constants.ENV_FILE) - - self.os_project = OpenStackProject( - snaps_creds, - ProjectConfig( - name=self.tenant_name, - description=self.tenant_description - )) - self.os_project.create() - self.created_object.append(self.os_project) - user_creator = OpenStackUser( - snaps_creds, - UserConfig( - name=self.user_name, - password=str(uuid.uuid4()), - roles={'admin': self.tenant_name})) - user_creator.create() - self.created_object.append(user_creator) - self.snaps_creds = user_creator.get_os_creds(self.tenant_name) - - return testcase.TestCase.EX_OK - except Exception: # pylint: disable=broad-except - self.__logger.exception("Exception raised during VNF preparation") - raise VnfPreparationException - - def deploy_orchestrator(self): - """ - Deploy an orchestrator (optional). - - If this method is overriden then raise orchestratorDeploymentException - if error during orchestrator deployment - """ - self.__logger.info("Deploy orchestrator (if necessary)") - return True - - def deploy_vnf(self): - """ - Deploy the VNF - - This function MUST be implemented by vnf test cases. - The details section MAY be updated in the vnf test cases. - - The deployment can be executed via a specific orchestrator - or using build-in orchestrators such as heat, OpenBaton, cloudify, - juju, onap, ... - - Returns: - True if the VNF is properly deployed - False if the VNF is not deployed - - Raise VnfDeploymentException if error during VNF deployment - """ - self.__logger.error("VNF must be deployed") - raise VnfDeploymentException - - def test_vnf(self): - """ - Test the VNF - - This function MUST be implemented by vnf test cases. - The details section MAY be updated in the vnf test cases. - - Once a VNF is deployed, it is assumed that specific test suite can be - run to validate the VNF. - Please note that the same test suite can be used on several test case - (e.g. clearwater test suite can be used whatever the orchestrator used - for the deployment) - - Returns: - True if VNF tests are PASS - False if test suite is FAIL - - Raise VnfTestException if error during VNF test - """ - self.__logger.error("VNF must be tested") - raise VnfTestException - - def clean(self): - """ - Clean VNF test case. - - It is up to the test providers to delete resources used for the tests. - By default we clean: - - * the user, - * the tenant - """ - self.__logger.info('Removing the VNF resources ..') - for creator in reversed(self.created_object): - try: - creator.clean() - except Exception as exc: # pylint: disable=broad-except - self.__logger.error('Unexpected error cleaning - %s', exc) |