#!/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 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 robot.api
from robot.errors import RobotError
import robot.run
from robot.utils.robottime import timestamp_to_secs
from six import StringIO
from six.moves import urllib
from functest.core import testcase
from functest.utils import constants
import functest.utils.openstack_utils as op_utils
__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
class ODLResultVisitor(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 ODLTests(testcase.TestCase):
"""ODL test runner."""
odl_test_repo = constants.CONST.__getattribute__('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 = os.path.join(
constants.CONST.__getattribute__('dir_results'), 'odl')
__logger = logging.getLogger(__name__)
@classmethod
def set_robotframework_vars(cls, odlusername="admin", odlpassword="admin"):
"""Set credentials in csit/variables/Variables.py.
Returns:
True if credentials are set.
False otherwise.
"""
odl_variables_files = os.path.join(cls.odl_test_repo,
'csit/variables/Variables.py')
try:
for line in fileinput.input(odl_variables_files,
inplace=True):
print(re.sub("AUTH = .*",
("AUTH = [u'" + odlusername + "', u'" +
odlpassword + "']"),
line.rstrip()))
return True
except Exception as ex: # pylint: disable=broad-except
cls.__logger.error("Cannot set ODL creds: %s", str(ex))
return False
def parse_results(self):
"""Parse output.xml and get the details in it."""
xml_file = os.path.join(self.res_dir, 'output.xml')
result = robot.api.ExecutionResult(xml_file)
visitor = ODLResultVisitor()
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_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,
* neutronip,
* osusername,
* ostenantname,
* 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 = urllib.parse.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 ex:
self.__logger.error("Cannot run ODL testcases. Please check "
"%s", str(ex))
return self.EX_RUN_ERROR
if self.set_robotframework_vars(odlusername, odlpassword):
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
output_dir = os.path.join(self.res_dir, 'output.xml')
stream = StringIO()
robot.run(*suites, variable=variables, output=output_dir,
log='NONE', report='NONE', stdout=stream)
self.__logger.info("\n" + stream.getvalue())
self.__logger.info("ODL results were successfully generated")
try:
self.parse_results()
self.__logger.info("ODL results were successfully parsed")
except RobotError as ex:
self.__logger.error("Run tests before publishing: %s",
ex.message)
return self.EX_RUN_ERROR
return self.EX_OK
else:
return self.EX_RUN_ERROR
def run(self, **kwargs):
"""Run suites in OPNFV environment
It basically check 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': urllib.parse.urlparse(neutron_url).hostname}
kwargs['odlip'] = kwargs['neutronip']
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']
kwargs['osusername'] = os.environ['OS_USERNAME']
kwargs['ostenantname'] = os.environ['OS_TENANT_NAME']
kwargs['osauthurl'] = os.environ['OS_AUTH_URL']
kwargs['ospassword'] = os.environ['OS_PASSWORD']
if installer_type == 'fuel':
kwargs['odlwebport'] = '8181'
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['odlip'] = os.environ['SDN_CONTROLLER_IP']
kwargs['odlwebport'] = '8181'
kwargs['odlrestconfport'] = '8087'
else:
kwargs['odlip'] = os.environ['SDN_CONTROLLER_IP']
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: # pylint: disable=broad-except
self.__logger.exception("Cannot run ODL testcases.")
return self.EX_RUN_ERROR
return self.run_suites(suites, **kwargs)
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')
self.parser.add_argument(
'-k', '--osauthurl', help='OS_AUTH_URL as defined by OpenStack',
default='http://127.0.0.1:5000/v2.0')
self.parser.add_argument(
'-a', '--osusername', help='Username for OpenStack',
default='admin')
self.parser.add_argument(
'-b', '--ostenantname', help='Tenantname for OpenStack',
default='admin')
self.parser.add_argument(
'-c', '--ospassword', help='Password for OpenStack',
default='admin')
self.parser.add_argument(
'-o', '--odlip', help='OpenDaylight IP',
default='127.0.0.1')
self.parser.add_argument(
'-w', '--odlwebport', help='OpenDaylight Web Portal Port',
default='8080')
self.parser.add_argument(
'-r', '--odlrestconfport', help='OpenDaylight RESTConf Port',
default='8181')
self.parser.add_argument(
'-d', '--odlusername', help='Username for ODL',
default='admin')
self.parser.add_argument(
'-e', '--odlpassword', help='Password for ODL',
default='admin')
self.parser.add_argument(
'-p', '--pushtodb', help='Push results to DB',
action='store_true')
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))
def main():
"""Entry point"""
logging.basicConfig()
odl = ODLTests()
parser = ODLParser()
args = parser.parse_args(sys.argv[1:])
try:
result = odl.run_suites(ODLTests.default_suites, **args)
if result != testcase.TestCase.EX_OK:
return result
if args['pushtodb']:
return odl.push_to_db()
else:
return result
except Exception: # pylint: disable=broad-except
return testcase.TestCase.EX_RUN_ERROR