# 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
+"""
+
import argparse
import errno
import fileinput
import sys
import urlparse
-from robot.api import ExecutionResult, ResultVisitor
+import robot.api
from robot.errors import RobotError
import robot.run
from robot.utils.robottime import timestamp_to_secs
-import functest.core.testcase_base as testcase_base
+from functest.core import testcase
import functest.utils.functest_logger as ft_logger
import functest.utils.openstack_utils as op_utils
-import functest.utils.functest_constants as ft_constants
+
+__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
-class ODLResultVisitor(ResultVisitor):
+class ODLResultVisitor(robot.api.ResultVisitor):
+ """Visitor to get result details."""
def __init__(self):
self._data = []
output['name'] = test.name
output['parent'] = test.parent.name
output['status'] = test.status
- output['startime'] = test.starttime
+ output['starttime'] = test.starttime
output['endtime'] = test.endtime
output['critical'] = test.critical
output['text'] = test.message
self._data.append(output)
def get_data(self):
+ """Get the details of the result."""
return self._data
-class ODLTests(testcase_base.TestcaseBase):
+class ODLTests(testcase.TestCase):
+ """ODL test runner."""
- repos = ft_constants.REPOS_DIR
+ 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")
- res_dir = os.path.join(ft_constants.FUNCTEST_RESULTS_DIR, "odl")
-
+ default_suites = [basic_suite_dir, neutron_suite_dir]
+ res_dir = '/home/opnfv/functest/results/odl/'
logger = ft_logger.Logger("opendaylight").getLogger()
- def __init__(self):
- self.case_name = "odl"
-
@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:
odlpassword + "']"),
line.rstrip())
return True
- except Exception as e:
- cls.logger.error("Cannot set ODL creds: %s" % str(e))
+ except Exception as ex: # pylint: disable=broad-except
+ cls.logger.error("Cannot set ODL creds: %s", str(ex))
return False
def parse_results(self):
- output_dir = os.path.join(self.res_dir, 'output.xml')
- result = ExecutionResult(output_dir)
+ """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)
self.criteria = result.suite.status
self.details = {}
self.details['description'] = result.suite.name
self.details['tests'] = visitor.get_data()
- return self.criteria
- def main(self, **kwargs):
- dirs = [self.basic_suite_dir, self.neutron_suite_dir]
+ def main(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']
- variables = ['KEYSTONE:' + kwargs['keystoneip'],
+ 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:
+ except KeyError as ex:
self.logger.error("Cannot run ODL testcases. Please check "
- "%s" % str(e))
+ "%s", str(ex))
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:
+ except OSError as ex:
+ if ex.errno != errno.EEXIST:
self.logger.exception(
- "Cannot create {}".format(self.res_dir))
+ "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(*dirs, variable=variables,
+ robot.run(*suites, variable=variables,
output=output_dir,
log='NONE',
report='NONE',
self.logger.info("\n" + stdout.read())
self.logger.info("ODL results were successfully generated")
try:
- test_res = self.parse_results()
+ self.parse_results()
self.logger.info("ODL results were successfully parsed")
- if test_res is not "PASS":
- return self.EX_RUN_ERROR
- except RobotError as e:
- self.logger.error("Run tests before publishing: %s" %
- e.message)
+ except RobotError as ex:
+ 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 {}".format(stdout_file))
+ self.logger.warning("Cannot remove %s", stdout_file)
return self.EX_OK
else:
return self.EX_RUN_ERROR
- def run(self):
+ 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:
- kclient = op_utils.get_keystone_client()
- keystone_url = kclient.service_catalog.url_for(
- service_type='identity', endpoint_type='publicURL')
- neutron_url = kclient.service_catalog.url_for(
- service_type='network', endpoint_type='publicURL')
- kwargs = {'keystoneip': urlparse.urlparse(keystone_url).hostname}
- kwargs['neutronip'] = urlparse.urlparse(neutron_url).hostname
+ 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']
kwargs['odlwebport'] = '8080'
kwargs['odlrestconfport'] = '8181'
kwargs['odlusername'] = 'admin'
kwargs['odlpassword'] = 'admin'
-
- installer_type = ft_constants.CI_INSTALLER_TYPE
- kwargs['osusername'] = ft_constants.OS_USERNAME
- kwargs['ostenantname'] = ft_constants.OS_TENANT_NAME
- kwargs['ospassword'] = ft_constants.OS_PASSWORD
-
+ 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'] = '8282'
- elif installer_type == 'apex':
- if ft_constants.SDN_CONTROLLER_IP is None:
- return self.EX_RUN_ERROR
- kwargs['odlip'] = ft_constants.SDN_CONTROLLER_IP
- kwargs['odlwebport'] = '8181'
+ 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':
- if ft_constants.SDN_CONTROLLER is None:
- return self.EX_RUN_ERROR
- kwargs['odlip'] = ft_constants.SDN_CONTROLLER
+ kwargs['odlip'] = os.environ['SDN_CONTROLLER']
elif installer_type == 'compass':
kwargs['odlwebport'] = '8181'
else:
- if ft_constants.SDN_CONTROLLER_IP is None:
- return self.EX_RUN_ERROR
- kwargs['odlip'] = ft_constants.SDN_CONTROLLER_IP
- except KeyError as e:
+ kwargs['odlip'] = os.environ['SDN_CONTROLLER_IP']
+ except KeyError as ex:
self.logger.error("Cannot run ODL testcases. "
"Please check env var: "
- "%s" % str(e))
+ "%s", str(ex))
return self.EX_RUN_ERROR
- except Exception:
+ except Exception: # pylint: disable=broad-except
self.logger.exception("Cannot run ODL testcases.")
return self.EX_RUN_ERROR
- return self.main(**kwargs)
+ return self.main(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))
if __name__ == '__main__':
- parser = argparse.ArgumentParser()
- parser.add_argument('-k', '--keystoneip',
- help='Keystone IP',
- default='127.0.0.1')
- parser.add_argument('-n', '--neutronip',
- help='Neutron IP',
- default='127.0.0.1')
- parser.add_argument('-a', '--osusername',
- help='Username for OpenStack',
- default='admin')
- parser.add_argument('-b', '--ostenantname',
- help='Tenantname for OpenStack',
- default='admin')
- parser.add_argument('-c', '--ospassword',
- help='Password for OpenStack',
- default='admin')
- parser.add_argument('-o', '--odlip',
- help='OpenDaylight IP',
- default='127.0.0.1')
- parser.add_argument('-w', '--odlwebport',
- help='OpenDaylight Web Portal Port',
- default='8080')
- parser.add_argument('-r', '--odlrestconfport',
- help='OpenDaylight RESTConf Port',
- default='8181')
- parser.add_argument('-d', '--odlusername',
- help='Username for ODL',
- default='admin')
- parser.add_argument('-e', '--odlpassword',
- help='Password for ODL',
- default='admin')
- parser.add_argument('-p', '--pushtodb',
- help='Push results to DB',
- action='store_true')
-
- args = vars(parser.parse_args())
- odl = ODLTests()
+ ODL = ODLTests()
+ PARSER = ODLParser()
+ ARGS = PARSER.parse_args(sys.argv[1:])
try:
- result = odl.main(**args)
- if result != testcase_base.TestcaseBase.EX_OK:
- sys.exit(result)
- if args['pushtodb']:
- sys.exit(odl.push_to_db())
- except Exception:
- sys.exit(testcase_base.TestcaseBase.EX_RUN_ERROR)
+ 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())
+ except Exception: # pylint: disable=broad-except
+ sys.exit(testcase.TestCase.EX_RUN_ERROR)
Code Review / functest.git / blobdiff