diff options
Diffstat (limited to 'xtesting/core')
| -rw-r--r-- | xtesting/core/__init__.py | 0 | ||||
| -rw-r--r-- | xtesting/core/feature.py | 133 | ||||
| -rw-r--r-- | xtesting/core/robotframework.py | 126 | ||||
| -rw-r--r-- | xtesting/core/testcase.py | 227 | ||||
| -rw-r--r-- | xtesting/core/unit.py | 92 | ||||
| -rw-r--r-- | xtesting/core/vnf.py | 205 |
6 files changed, 783 insertions, 0 deletions
diff --git a/xtesting/core/__init__.py b/xtesting/core/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/xtesting/core/__init__.py diff --git a/xtesting/core/feature.py b/xtesting/core/feature.py new file mode 100644 index 00000000..d3f86c02 --- /dev/null +++ b/xtesting/core/feature.py @@ -0,0 +1,133 @@ +#!/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 Xtesting 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 + +from xtesting.core import testcase + +__author__ = ("Serena Feng <feng.xiaowei@zte.com.cn>, " + "Cedric Ollivier <cedric.ollivier@orange.com>") + + +class Feature(testcase.TestCase): + """Base model for single feature.""" + + __logger = logging.getLogger(__name__) + dir_results = "/home/opnfv/xtesting/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 = testcase.TestCase.EX_RUN_ERROR + self.result = 0 + try: + if self.execute(**kwargs) == 0: + exit_code = testcase.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/xtesting/core/robotframework.py b/xtesting/core/robotframework.py new file mode 100644 index 00000000..4d3746aa --- /dev/null +++ b/xtesting/core/robotframework.py @@ -0,0 +1,126 @@ +#!/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 xtesting.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/xtesting/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/xtesting/core/testcase.py b/xtesting/core/testcase.py new file mode 100644 index 00000000..4effa932 --- /dev/null +++ b/xtesting/core/testcase.py @@ -0,0 +1,227 @@ +#!/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 Xtesting TestCases.""" + +from datetime import datetime +import json +import logging +import os +import re +import requests + +from xtesting.utils import decorators +from xtesting.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', 'xtesting') + 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/xtesting/core/unit.py b/xtesting/core/unit.py new file mode 100644 index 00000000..27773679 --- /dev/null +++ b/xtesting/core/unit.py @@ -0,0 +1,92 @@ +#!/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 xtesting.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/xtesting/core/vnf.py b/xtesting/core/vnf.py new file mode 100644 index 00000000..95ebde04 --- /dev/null +++ b/xtesting/core/vnf.py @@ -0,0 +1,205 @@ +#!/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 xtesting.core import testcase +from xtesting.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) |