$ 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
+from six import StringIO
+from six.moves import urllib
from functest.core import testcase
-import functest.utils.functest_logger as ft_logger
+from functest.utils import constants
import functest.utils.openstack_utils as op_utils
__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
class ODLTests(testcase.TestCase):
"""ODL test runner."""
- repos = "/home/opnfv/repos/"
- odl_test_repo = os.path.join(repos, "odl_test")
+ 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 = '/home/opnfv/functest/results/odl/'
- logger = ft_logger.Logger("opendaylight").getLogger()
+ 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"):
try:
for line in fileinput.input(odl_variables_files,
inplace=True):
- print re.sub("AUTH = .*",
+ print(re.sub("AUTH = .*",
("AUTH = [u'" + odlusername + "', u'" +
odlpassword + "']"),
- line.rstrip())
+ line.rstrip()))
return True
except Exception as ex: # pylint: disable=broad-except
- cls.logger.error("Cannot set ODL creds: %s", str(ex))
+ cls.__logger.error("Cannot set ODL creds: %s", str(ex))
return False
def parse_results(self):
result = robot.api.ExecutionResult(xml_file)
visitor = ODLResultVisitor()
result.visit(visitor)
- self.criteria = result.suite.status
+ 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 main(self, suites=None, **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,
* delete temporary files.
Args:
- **kwargs: Arbitrary keyword arguments.
+ kwargs: Arbitrary keyword arguments.
Returns:
EX_OK if all suites ran well.
odlusername = kwargs['odlusername']
odlpassword = kwargs['odlpassword']
osauthurl = kwargs['osauthurl']
- keystoneip = urlparse.urlparse(osauthurl).hostname
+ keystoneip = urllib.parse.urlparse(osauthurl).hostname
variables = ['KEYSTONE:' + keystoneip,
'NEUTRON:' + kwargs['neutronip'],
'OS_AUTH_URL:"' + osauthurl + '"',
'PORT:' + kwargs['odlwebport'],
'RESTCONFPORT:' + kwargs['odlrestconfport']]
except KeyError as ex:
- self.logger.error("Cannot run ODL testcases. Please check "
- "%s", str(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(
+ self.__logger.exception(
"Cannot create %s", 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")
+ 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")
+ self.__logger.info("ODL results were successfully parsed")
except RobotError as ex:
- self.logger.error("Run tests before publishing: %s",
- ex.message)
+ self.__logger.error("Run tests before publishing: %s",
+ ex.message)
return self.EX_RUN_ERROR
- try:
- os.remove(stdout_file)
- except OSError:
- self.logger.warning("Cannot remove %s", stdout_file)
return self.EX_OK
else:
return self.EX_RUN_ERROR
required.
Args:
- **kwargs: Arbitrary keyword arguments.
+ kwargs: Arbitrary keyword arguments.
Returns:
EX_OK if all suites ran well.
except KeyError:
pass
neutron_url = op_utils.get_endpoint(service_type='network')
- kwargs = {'neutronip': urlparse.urlparse(neutron_url).hostname}
+ kwargs = {'neutronip': urllib.parse.urlparse(neutron_url).hostname}
kwargs['odlip'] = kwargs['neutronip']
kwargs['odlwebport'] = '8080'
kwargs['odlrestconfport'] = '8181'
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))
+ 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.")
+ 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): # pylint: disable=too-few-public-methods
return vars(self.parser.parse_args(argv))
-if __name__ == '__main__':
- ODL = ODLTests()
- PARSER = ODLParser()
- ARGS = PARSER.parse_args(sys.argv[1:])
+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.TestCase.EX_OK:
- sys.exit(RESULT)
- if ARGS['pushtodb']:
- sys.exit(ODL.push_to_db())
+ 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
- sys.exit(testcase.TestCase.EX_RUN_ERROR)
+ return testcase.TestCase.EX_RUN_ERROR