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