# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
+"""Define classes required to run ODL suites.
+
+It has been designed for any context. But helpers are given for
+running test suites in OPNFV environment.
+
+Example:
+ $ python odl.py
+"""
+
+from __future__ import division
+
import argparse
-import errno
import fileinput
+import logging
import os
import re
import sys
-import urlparse
-import robot.api
-from robot.errors import RobotError
-import robot.run
-from robot.utils.robottime import timestamp_to_secs
+import os_client_config
+from six.moves import urllib
+from xtesting.core import robotframework
-from functest.core import testcase_base
-import functest.utils.functest_logger as ft_logger
-import functest.utils.openstack_utils as op_utils
+from functest.utils import config
+from functest.utils import env
+__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
-class ODLResultVisitor(robot.api.ResultVisitor):
- 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):
- return self._data
-
-
-class ODLTests(testcase_base.TestCase):
-
- repos = "/home/opnfv/repos/"
- odl_test_repo = os.path.join(repos, "odl_test")
- neutron_suite_dir = os.path.join(odl_test_repo,
- "csit/suites/openstack/neutron")
- basic_suite_dir = os.path.join(odl_test_repo,
- "csit/suites/integration/basic")
+class ODLTests(robotframework.RobotFramework):
+ """ODL test runner."""
+
+ odl_test_repo = getattr(config.CONF, 'dir_repo_odl_test')
+ neutron_suite_dir = os.path.join(
+ odl_test_repo, "csit/suites/openstack/neutron")
+ basic_suite_dir = os.path.join(
+ odl_test_repo, "csit/suites/integration/basic")
default_suites = [basic_suite_dir, neutron_suite_dir]
- res_dir = '/home/opnfv/functest/results/odl/'
- logger = ft_logger.Logger("opendaylight").getLogger()
+ odl_variables_file = os.path.join(
+ odl_test_repo, 'csit/variables/Variables.robot')
+ __logger = logging.getLogger(__name__)
- def __init__(self):
- testcase_base.TestCase.__init__(self)
- self.case_name = "odl"
+ def __init__(self, **kwargs):
+ super(ODLTests, self).__init__(**kwargs)
+ self.res_dir = os.path.join(
+ getattr(config.CONF, 'dir_results'), 'odl')
+ self.xml_file = os.path.join(self.res_dir, 'output.xml')
@classmethod
def set_robotframework_vars(cls, odlusername="admin", odlpassword="admin"):
- odl_variables_files = os.path.join(cls.odl_test_repo,
- 'csit/variables/Variables.py')
+ """Set credentials in csit/variables/Variables.robot.
+
+ Returns:
+ True if credentials are set.
+ False otherwise.
+ """
+
try:
- for line in fileinput.input(odl_variables_files,
+ for line in fileinput.input(cls.odl_variables_file,
inplace=True):
- print re.sub("AUTH = .*",
- ("AUTH = [u'" + odlusername + "', u'" +
- odlpassword + "']"),
- line.rstrip())
+ print(re.sub("@{AUTH}.*",
+ "@{{AUTH}} {} {}".format(
+ odlusername, odlpassword),
+ line.rstrip()))
return True
- except Exception as e:
- cls.logger.error("Cannot set ODL creds: %s" % str(e))
+ except Exception: # pylint: disable=broad-except
+ cls.__logger.exception("Cannot set ODL creds:")
return False
- def parse_results(self):
- xml_file = os.path.join(self.res_dir, 'output.xml')
- result = robot.api.ExecutionResult(xml_file)
- visitor = ODLResultVisitor()
- result.visit(visitor)
- self.criteria = result.suite.status
- 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 main(self, suites=default_suites, **kwargs):
+ def run_suites(self, suites=None, **kwargs):
+ """Run the test suites
+
+ It has been designed to be called in any context.
+ It requires the following keyword arguments:
+
+ * odlusername,
+ * odlpassword,
+ * osauthurl,
+ * neutronurl,
+ * osusername,
+ * osprojectname,
+ * ospassword,
+ * odlip,
+ * odlwebport,
+ * odlrestconfport.
+
+ Here are the steps:
+ * set all RobotFramework_variables,
+ * 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:
+ if not suites:
+ suites = self.default_suites
odlusername = kwargs['odlusername']
odlpassword = kwargs['odlpassword']
osauthurl = kwargs['osauthurl']
- keystoneip = urlparse.urlparse(osauthurl).hostname
- variables = ['KEYSTONE:' + keystoneip,
- 'NEUTRON:' + kwargs['neutronip'],
- 'OS_AUTH_URL:"' + osauthurl + '"',
- 'OSUSERNAME:"' + kwargs['osusername'] + '"',
- 'OSTENANTNAME:"' + kwargs['ostenantname'] + '"',
- 'OSPASSWORD:"' + kwargs['ospassword'] + '"',
- 'ODL_SYSTEM_IP:' + kwargs['odlip'],
- 'PORT:' + kwargs['odlwebport'],
- 'RESTCONFPORT:' + kwargs['odlrestconfport']]
- except KeyError as e:
- self.logger.error("Cannot run ODL testcases. Please check "
- "%s" % str(e))
+ keystoneurl = "{}://{}".format(
+ urllib.parse.urlparse(osauthurl).scheme,
+ urllib.parse.urlparse(osauthurl).netloc)
+ variable = ['KEYSTONEURL:' + keystoneurl,
+ 'NEUTRONURL:' + kwargs['neutronurl'],
+ 'OS_AUTH_URL:"' + osauthurl + '"',
+ 'OSUSERNAME:"' + kwargs['osusername'] + '"',
+ ('OSUSERDOMAINNAME:"' +
+ kwargs['osuserdomainname'] + '"'),
+ 'OSTENANTNAME:"' + kwargs['osprojectname'] + '"',
+ ('OSPROJECTDOMAINNAME:"' +
+ kwargs['osprojectdomainname'] + '"'),
+ 'OSPASSWORD:"' + kwargs['ospassword'] + '"',
+ 'ODL_SYSTEM_IP:' + kwargs['odlip'],
+ 'PORT:' + kwargs['odlwebport'],
+ 'RESTCONFPORT:' + kwargs['odlrestconfport']]
+ except KeyError:
+ self.__logger.exception("Cannot run ODL testcases. Please check")
return self.EX_RUN_ERROR
- if self.set_robotframework_vars(odlusername, odlpassword):
- try:
- os.makedirs(self.res_dir)
- except OSError as e:
- if e.errno != errno.EEXIST:
- self.logger.exception(
- "Cannot create {}".format(self.res_dir))
- return self.EX_RUN_ERROR
- stdout_file = os.path.join(self.res_dir, 'stdout.txt')
- output_dir = os.path.join(self.res_dir, 'output.xml')
- with open(stdout_file, 'w+') as stdout:
- robot.run(*suites, variable=variables,
- output=output_dir,
- log='NONE',
- report='NONE',
- stdout=stdout)
- stdout.seek(0, 0)
- self.logger.info("\n" + stdout.read())
- self.logger.info("ODL results were successfully generated")
- try:
- self.parse_results()
- self.logger.info("ODL results were successfully parsed")
- except RobotError as e:
- self.logger.error("Run tests before publishing: %s" %
- e.message)
- return self.EX_RUN_ERROR
- try:
- os.remove(stdout_file)
- except OSError:
- self.logger.warning("Cannot remove {}".format(stdout_file))
- return self.EX_OK
+ if not os.path.isfile(self.odl_variables_file):
+ self.__logger.info("Skip writting ODL creds")
else:
- return self.EX_RUN_ERROR
+ if not self.set_robotframework_vars(odlusername, odlpassword):
+ return self.EX_RUN_ERROR
+ return super(ODLTests, self).run(variable=variable, suites=suites)
def run(self, **kwargs):
+ """Run suites in OPNFV environment
+
+ It basically checks env vars to call main() with the keywords
+ required.
+
+ Args:
+ kwargs: Arbitrary keyword arguments.
+
+ Returns:
+ EX_OK if all suites ran well.
+ EX_RUN_ERROR otherwise.
+ """
try:
suites = self.default_suites
try:
suites = kwargs["suites"]
except KeyError:
pass
- neutron_url = op_utils.get_endpoint(service_type='network')
- kwargs = {'neutronip': urlparse.urlparse(neutron_url).hostname}
- kwargs['odlip'] = kwargs['neutronip']
+ cloud = os_client_config.make_shade()
+ neutron_id = cloud.search_services('neutron')[0].id
+ endpoint = cloud.search_endpoints(
+ filters={
+ 'interface': os.environ.get(
+ 'OS_INTERFACE', 'public').replace('URL', ''),
+ 'service_id': neutron_id})[0].url
+ kwargs = {'neutronurl': endpoint}
+ kwargs['odlip'] = env.get('SDN_CONTROLLER_IP')
kwargs['odlwebport'] = '8080'
kwargs['odlrestconfport'] = '8181'
kwargs['odlusername'] = 'admin'
kwargs['odlpassword'] = 'admin'
- installer_type = None
- if 'INSTALLER_TYPE' in os.environ:
- installer_type = os.environ['INSTALLER_TYPE']
+ installer_type = env.get('INSTALLER_TYPE')
kwargs['osusername'] = os.environ['OS_USERNAME']
- kwargs['ostenantname'] = os.environ['OS_TENANT_NAME']
+ kwargs['osuserdomainname'] = os.environ.get(
+ 'OS_USER_DOMAIN_NAME', 'Default')
+ kwargs['osprojectname'] = os.environ['OS_PROJECT_NAME']
+ kwargs['osprojectdomainname'] = os.environ.get(
+ 'OS_PROJECT_DOMAIN_NAME', 'Default')
kwargs['osauthurl'] = os.environ['OS_AUTH_URL']
kwargs['ospassword'] = os.environ['OS_PASSWORD']
if installer_type == 'fuel':
kwargs['odlwebport'] = '8282'
+ kwargs['odlrestconfport'] = '8282'
elif installer_type == 'apex' or installer_type == 'netvirt':
- kwargs['odlip'] = os.environ['SDN_CONTROLLER_IP']
kwargs['odlwebport'] = '8081'
kwargs['odlrestconfport'] = '8081'
- elif installer_type == 'joid':
- kwargs['odlip'] = os.environ['SDN_CONTROLLER']
elif installer_type == 'compass':
+ kwargs['odlrestconfport'] = '8080'
+ elif installer_type == 'daisy':
kwargs['odlwebport'] = '8181'
- else:
- kwargs['odlip'] = os.environ['SDN_CONTROLLER_IP']
- except KeyError as e:
- self.logger.error("Cannot run ODL testcases. "
- "Please check env var: "
- "%s" % str(e))
+ kwargs['odlrestconfport'] = '8087'
+ assert kwargs['odlip']
+ except KeyError as ex:
+ self.__logger.error(
+ "Cannot run ODL testcases. Please check env var: %s", str(ex))
return self.EX_RUN_ERROR
- except Exception:
- self.logger.exception("Cannot run ODL testcases.")
+ except Exception: # pylint: disable=broad-except
+ self.__logger.exception("Cannot run ODL testcases.")
return self.EX_RUN_ERROR
- return self.main(suites, **kwargs)
+ return self.run_suites(suites, **kwargs)
-class ODLParser(object):
+class ODLParser(object): # pylint: disable=too-few-public-methods
+ """Parser to run ODL test suites."""
def __init__(self):
self.parser = argparse.ArgumentParser()
self.parser.add_argument(
- '-n', '--neutronip', help='Neutron IP',
- default='127.0.0.1')
+ '-n', '--neutronurl', help='Neutron Endpoint',
+ default='http://127.0.0.1:9696')
self.parser.add_argument(
'-k', '--osauthurl', help='OS_AUTH_URL as defined by OpenStack',
- default='http://127.0.0.1:5000/v2.0')
+ default='http://127.0.0.1:5000/v3')
self.parser.add_argument(
'-a', '--osusername', help='Username for OpenStack',
default='admin')
self.parser.add_argument(
- '-b', '--ostenantname', help='Tenantname for OpenStack',
+ '-f', '--osuserdomainname', help='User domain name for OpenStack',
+ default='Default')
+ self.parser.add_argument(
+ '-b', '--osprojectname', help='Projet name for OpenStack',
default='admin')
+ self.parser.add_argument(
+ '-g', '--osprojectdomainname',
+ help='Project domain name for OpenStack',
+ default='Default')
self.parser.add_argument(
'-c', '--ospassword', help='Password for OpenStack',
default='admin')
'-p', '--pushtodb', help='Push results to DB',
action='store_true')
- def parse_args(self, argv=[]):
+ def parse_args(self, argv=None):
+ """Parse arguments.
+
+ It can call sys.exit if arguments are incorrect.
+
+ Returns:
+ the arguments from cmdline
+ """
+ if not argv:
+ argv = []
return vars(self.parser.parse_args(argv))
-if __name__ == '__main__':
+def main():
+ """Entry point"""
+ logging.basicConfig()
odl = ODLTests()
parser = ODLParser()
args = parser.parse_args(sys.argv[1:])
try:
- result = odl.main(ODLTests.default_suites, **args)
- if result != testcase_base.TestCase.EX_OK:
- sys.exit(result)
+ result = odl.run_suites(ODLTests.default_suites, **args)
+ if result != robotframework.RobotFramework.EX_OK:
+ return result
if args['pushtodb']:
- sys.exit(odl.push_to_db())
- except Exception:
- sys.exit(testcase_base.TestCase.EX_RUN_ERROR)
+ return odl.push_to_db()
+ return result
+ except Exception: # pylint: disable=broad-except
+ return robotframework.RobotFramework.EX_RUN_ERROR
Code Review / functest.git / blobdiff