aboutsummaryrefslogtreecommitdiffstats
path: root/functest/core
diff options
context:
space:
mode:
Diffstat (limited to 'functest/core')
-rw-r--r--functest/core/__init__.py0
-rw-r--r--functest/core/feature.py133
-rw-r--r--functest/core/robotframework.py126
-rw-r--r--functest/core/testcase.py227
-rw-r--r--functest/core/unit.py92
-rw-r--r--functest/core/vnf.py205
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)