+++ /dev/null
-#!/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 fileinput
-import logging
-import os
-import re
-import sys
-
-import os_client_config
-from six.moves import urllib
-from xtesting.core import robotframework
-
-from functest.utils import config
-from functest.utils import env
-
-__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
-
-
-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]
- odl_variables_file = os.path.join(
- odl_test_repo, 'csit/variables/Variables.robot')
- __logger = logging.getLogger(__name__)
-
- def __init__(self, **kwargs):
- super().__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"):
- """Set credentials in csit/variables/Variables.robot.
-
- Returns:
- True if credentials are set.
- False otherwise.
- """
-
- try:
- for line in fileinput.input(cls.odl_variables_file,
- inplace=True):
- print(re.sub(
- "@{AUTH}.*",
- f"@{{AUTH}} {odlusername} {odlpassword}",
- line.rstrip()))
- return True
- except Exception: # pylint: disable=broad-except
- cls.__logger.exception("Cannot set ODL creds:")
- return False
-
- 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']
- keystoneurl = (f"{urllib.parse.urlparse(osauthurl).scheme}://"
- f"{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 not os.path.isfile(self.odl_variables_file):
- self.__logger.info("Skip writting ODL creds")
- else:
- if not self.set_robotframework_vars(odlusername, odlpassword):
- return self.EX_RUN_ERROR
- return super().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
- 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'] = env.get('SDN_CONTROLLER_WEBPORT')
- kwargs['odlrestconfport'] = env.get('SDN_CONTROLLER_RESTCONFPORT')
- kwargs['odlusername'] = env.get('SDN_CONTROLLER_USER')
- kwargs['odlpassword'] = env.get('SDN_CONTROLLER_PASSWORD')
- kwargs['osusername'] = os.environ['OS_USERNAME']
- 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']
- 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: # pylint: disable=broad-except
- self.__logger.exception("Cannot run ODL testcases.")
- return self.EX_RUN_ERROR
-
- return self.run_suites(suites, **kwargs)
-
-
-class ODLParser(): # pylint: disable=too-few-public-methods
- """Parser to run ODL test suites."""
-
- def __init__(self):
- self.parser = argparse.ArgumentParser()
- self.parser.add_argument(
- '-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/v3')
- self.parser.add_argument(
- '-a', '--osusername', help='Username for OpenStack',
- default='admin')
- self.parser.add_argument(
- '-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')
- 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 != robotframework.RobotFramework.EX_OK:
- return result
- if args['pushtodb']:
- return odl.push_to_db()
- return result
- except Exception: # pylint: disable=broad-except
- return robotframework.RobotFramework.EX_RUN_ERROR
+++ /dev/null
-#!/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 classes required to fully cover odl."""
-
-import logging
-import os
-import unittest
-
-import mock
-import munch
-from robot.errors import RobotError
-import six
-from six.moves import urllib
-from xtesting.core import testcase
-
-from functest.opnfv_tests.sdn.odl import odl
-
-__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
-
-
-class ODLTesting(unittest.TestCase):
-
- """The super class which testing classes could inherit."""
- # pylint: disable=missing-docstring
-
- logging.disable(logging.CRITICAL)
-
- _keystone_ip = "127.0.0.1"
- _neutron_url = "https://127.0.0.1:9696"
- _neutron_id = "dummy"
- _sdn_controller_ip = "127.0.0.3"
- _os_auth_url = f"http://{_keystone_ip}:5000/v3"
- _os_projectname = "admin"
- _os_username = "admin"
- _os_password = "admin"
- _odl_webport = "8080"
- _odl_restconfport = "8181"
- _odl_username = "admin"
- _odl_password = "admin"
- _os_userdomainname = 'Default'
- _os_projectdomainname = 'Default'
- _os_interface = "public"
-
- def setUp(self):
- for var in ("SDN_CONTROLLER", "SDN_CONTROLLER_IP"):
- if var in os.environ:
- del os.environ[var]
- os.environ["OS_AUTH_URL"] = self._os_auth_url
- os.environ["OS_USERNAME"] = self._os_username
- os.environ["OS_USER_DOMAIN_NAME"] = self._os_userdomainname
- os.environ["OS_PASSWORD"] = self._os_password
- os.environ["OS_PROJECT_NAME"] = self._os_projectname
- os.environ["OS_PROJECT_DOMAIN_NAME"] = self._os_projectdomainname
- os.environ["OS_PASSWORD"] = self._os_password
- os.environ["OS_INTERFACE"] = self._os_interface
- self.test = odl.ODLTests(case_name='odl', project_name='functest')
- self.defaultargs = {'odlusername': self._odl_username,
- 'odlpassword': self._odl_password,
- 'neutronurl': f"http://{self._keystone_ip}:9696",
- 'osauthurl': self._os_auth_url,
- 'osusername': self._os_username,
- 'osuserdomainname': self._os_userdomainname,
- 'osprojectname': self._os_projectname,
- 'osprojectdomainname': self._os_projectdomainname,
- 'ospassword': self._os_password,
- 'odlip': self._keystone_ip,
- 'odlwebport': self._odl_webport,
- 'odlrestconfport': self._odl_restconfport,
- 'pushtodb': False}
-
-
-class ODLRobotTesting(ODLTesting):
-
- """The class testing ODLTests.set_robotframework_vars()."""
- # pylint: disable=missing-docstring
-
- @mock.patch('fileinput.input', side_effect=Exception())
- def test_set_vars_ko(self, mock_method):
- self.assertFalse(self.test.set_robotframework_vars())
- mock_method.assert_called_once_with(
- os.path.join(odl.ODLTests.odl_test_repo,
- 'csit/variables/Variables.robot'), inplace=True)
-
- @mock.patch('fileinput.input', return_value=[])
- def test_set_vars_empty(self, mock_method):
- self.assertTrue(self.test.set_robotframework_vars())
- mock_method.assert_called_once_with(
- os.path.join(odl.ODLTests.odl_test_repo,
- 'csit/variables/Variables.robot'), inplace=True)
-
- @mock.patch('sys.stdout', new_callable=six.StringIO)
- def _test_set_vars(self, msg1, msg2, *args):
- line = mock.MagicMock()
- line.__iter__.return_value = [msg1]
- with mock.patch('fileinput.input', return_value=line) as mock_method:
- self.assertTrue(self.test.set_robotframework_vars())
- mock_method.assert_called_once_with(
- os.path.join(odl.ODLTests.odl_test_repo,
- 'csit/variables/Variables.robot'), inplace=True)
- self.assertEqual(args[0].getvalue(), f"{msg2}\n")
-
- def test_set_vars_auth_default(self):
- self._test_set_vars(
- "@{AUTH} ",
- "@{AUTH} admin admin")
-
- def test_set_vars_auth1(self):
- self._test_set_vars(
- "@{AUTH1} foo bar",
- "@{AUTH1} foo bar")
-
- @mock.patch('sys.stdout', new_callable=six.StringIO)
- def test_set_vars_auth_foo(self, *args):
- line = mock.MagicMock()
- line.__iter__.return_value = ["@{AUTH} "]
- with mock.patch('fileinput.input', return_value=line) as mock_method:
- self.assertTrue(self.test.set_robotframework_vars('foo', 'bar'))
- mock_method.assert_called_once_with(
- os.path.join(odl.ODLTests.odl_test_repo,
- 'csit/variables/Variables.robot'), inplace=True)
- self.assertEqual(
- args[0].getvalue(),
- "@{AUTH} foo bar\n")
-
-
-class ODLMainTesting(ODLTesting):
-
- """The class testing ODLTests.run_suites()."""
- # pylint: disable=missing-docstring
-
- def _get_run_suites_kwargs(self, key=None):
- kwargs = {'odlusername': self._odl_username,
- 'odlpassword': self._odl_password,
- 'neutronurl': self._neutron_url,
- 'osauthurl': self._os_auth_url,
- 'osusername': self._os_username,
- 'osuserdomainname': self._os_userdomainname,
- 'osprojectname': self._os_projectname,
- 'osprojectdomainname': self._os_projectdomainname,
- 'ospassword': self._os_password,
- 'odlip': self._sdn_controller_ip,
- 'odlwebport': self._odl_webport,
- 'odlrestconfport': self._odl_restconfport}
- if key:
- del kwargs[key]
- return kwargs
-
- def _test_run_suites(self, status, *args):
- kwargs = self._get_run_suites_kwargs()
- self.assertEqual(self.test.run_suites(**kwargs), status)
- if args:
- args[0].assert_called_once_with(self.test.odl_variables_file)
- if len(args) > 1:
- variable = [
- ('KEYSTONEURL:'
- f'{urllib.parse.urlparse(self._os_auth_url).scheme}://'
- f'{urllib.parse.urlparse(self._os_auth_url).netloc}'),
- f'NEUTRONURL:{self._neutron_url}',
- f'OS_AUTH_URL:"{self._os_auth_url}"',
- f'OSUSERNAME:"{self._os_username}"',
- f'OSUSERDOMAINNAME:"{self._os_userdomainname}"',
- f'OSTENANTNAME:"{self._os_projectname}"',
- f'OSPROJECTDOMAINNAME:"{self._os_projectdomainname}"',
- f'OSPASSWORD:"{self._os_password}"',
- f'ODL_SYSTEM_IP:{self._sdn_controller_ip}',
- f'PORT:{self._odl_webport}',
- f'RESTCONFPORT:{self._odl_restconfport}']
- args[1].assert_called_once_with(
- odl.ODLTests.basic_suite_dir, odl.ODLTests.neutron_suite_dir,
- include=[],
- log='NONE',
- output=os.path.join(self.test.res_dir, 'output.xml'),
- report='NONE', stdout=mock.ANY, variable=variable,
- variablefile=[])
-
- def _test_no_keyword(self, key):
- kwargs = self._get_run_suites_kwargs(key)
- self.assertEqual(self.test.run_suites(**kwargs),
- testcase.TestCase.EX_RUN_ERROR)
-
- def test_no_odlusername(self):
- self._test_no_keyword('odlusername')
-
- def test_no_odlpassword(self):
- self._test_no_keyword('odlpassword')
-
- def test_no_neutronurl(self):
- self._test_no_keyword('neutronurl')
-
- def test_no_osauthurl(self):
- self._test_no_keyword('osauthurl')
-
- def test_no_osusername(self):
- self._test_no_keyword('osusername')
-
- def test_no_osprojectname(self):
- self._test_no_keyword('osprojectname')
-
- def test_no_ospassword(self):
- self._test_no_keyword('ospassword')
-
- def test_no_odlip(self):
- self._test_no_keyword('odlip')
-
- def test_no_odlwebport(self):
- self._test_no_keyword('odlwebport')
-
- def test_no_odlrestconfport(self):
- self._test_no_keyword('odlrestconfport')
-
- @mock.patch('os.path.isfile', return_value=True)
- def test_set_vars_ko(self, *args):
- with mock.patch.object(self.test, 'set_robotframework_vars',
- return_value=False) as mock_object:
- self._test_run_suites(testcase.TestCase.EX_RUN_ERROR)
- mock_object.assert_called_once_with(
- self._odl_username, self._odl_password)
- args[0].assert_called_once_with(self.test.odl_variables_file)
-
- @mock.patch('os.makedirs')
- @mock.patch('robot.run', side_effect=RobotError)
- @mock.patch('os.path.isfile', return_value=True)
- def test_run_ko(self, *args):
- with mock.patch.object(self.test, 'set_robotframework_vars',
- return_value=True), \
- self.assertRaises(RobotError):
- self._test_run_suites(testcase.TestCase.EX_RUN_ERROR, *args)
-
- @mock.patch('os.makedirs')
- @mock.patch('robot.run')
- @mock.patch('os.path.isfile', return_value=True)
- def test_parse_results_ko(self, *args):
- with mock.patch.object(self.test, 'set_robotframework_vars',
- return_value=True), \
- mock.patch.object(self.test, 'parse_results',
- side_effect=RobotError):
- self._test_run_suites(testcase.TestCase.EX_RUN_ERROR, *args)
-
- @mock.patch('os.makedirs')
- @mock.patch('robot.run')
- @mock.patch('os.path.isfile', return_value=True)
- def test_generate_report_ko(self, *args):
- with mock.patch.object(self.test, 'set_robotframework_vars',
- return_value=True), \
- mock.patch.object(self.test, 'parse_results'), \
- mock.patch.object(self.test, 'generate_report',
- return_value=1):
- self._test_run_suites(testcase.TestCase.EX_OK, *args)
-
- @mock.patch('os.makedirs')
- @mock.patch('robot.run')
- @mock.patch('os.path.isfile', return_value=True)
- def test_generate_report_exc(self, *args):
- with mock.patch.object(self.test, 'set_robotframework_vars',
- return_value=True), \
- mock.patch.object(self.test, 'parse_results'), \
- mock.patch.object(self.test, 'generate_report',
- side_effect=Exception):
- self._test_run_suites(testcase.TestCase.EX_RUN_ERROR, *args)
-
- @mock.patch('os.makedirs')
- @mock.patch('robot.run')
- @mock.patch('os.path.isfile', return_value=True)
- def test_ok(self, *args):
- with mock.patch.object(self.test, 'set_robotframework_vars',
- return_value=True), \
- mock.patch.object(self.test, 'parse_results'), \
- mock.patch.object(self.test, 'generate_report',
- return_value=0):
- self._test_run_suites(testcase.TestCase.EX_OK, *args)
-
- @mock.patch('os.makedirs')
- @mock.patch('robot.run')
- @mock.patch('os.path.isfile', return_value=False)
- def test_ok_no_creds(self, *args):
- with mock.patch.object(self.test, 'set_robotframework_vars',
- return_value=True) as mock_method, \
- mock.patch.object(self.test, 'parse_results'), \
- mock.patch.object(self.test, 'generate_report',
- return_value=0):
- self._test_run_suites(testcase.TestCase.EX_OK, *args)
- mock_method.assert_not_called()
-
- @mock.patch('os.makedirs')
- @mock.patch('robot.run', return_value=1)
- @mock.patch('os.path.isfile', return_value=True)
- def test_testcases_in_failure(self, *args):
- with mock.patch.object(self.test, 'set_robotframework_vars',
- return_value=True), \
- mock.patch.object(self.test, 'parse_results'), \
- mock.patch.object(self.test, 'generate_report',
- return_value=0):
- self._test_run_suites(testcase.TestCase.EX_OK, *args)
-
-
-class ODLRunTesting(ODLTesting):
- """The class testing ODLTests.run()."""
- # pylint: disable=too-many-public-methods,missing-docstring
-
- @mock.patch('os_client_config.make_shade', side_effect=Exception)
- def test_no_cloud(self, *args):
- self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
- args[0].assert_called_once_with()
-
- @mock.patch('os_client_config.make_shade')
- def test_no_service1(self, *args):
- args[0].return_value.search_services.return_value = None
- self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
- args[0].return_value.search_services.assert_called_once_with('neutron')
- args[0].return_value.search_endpoints.assert_not_called()
-
- @mock.patch('os_client_config.make_shade')
- def test_no_service2(self, *args):
- args[0].return_value.search_services.return_value = []
- self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
- args[0].return_value.search_services.assert_called_once_with('neutron')
- args[0].return_value.search_endpoints.assert_not_called()
-
- @mock.patch('os_client_config.make_shade')
- def test_no_service3(self, *args):
- args[0].return_value.search_services.return_value = [
- munch.Munch()]
- self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
- args[0].return_value.search_services.assert_called_once_with('neutron')
- args[0].return_value.search_endpoints.assert_not_called()
-
- @mock.patch('os_client_config.make_shade')
- def test_no_endpoint1(self, *args):
- args[0].return_value.search_services.return_value = [
- munch.Munch(id=self._neutron_id)]
- args[0].return_value.search_endpoints.return_value = None
- self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
- args[0].return_value.search_services.assert_called_once_with('neutron')
- args[0].return_value.search_endpoints.assert_called_once_with(
- filters={'interface': self._os_interface,
- 'service_id': self._neutron_id})
-
- @mock.patch('os_client_config.make_shade')
- def test_no_endpoint2(self, *args):
- args[0].return_value.search_services.return_value = [
- munch.Munch(id=self._neutron_id)]
- args[0].return_value.search_endpoints.return_value = []
- self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
- args[0].return_value.search_services.assert_called_once_with('neutron')
- args[0].return_value.search_endpoints.assert_called_once_with(
- filters={'interface': self._os_interface,
- 'service_id': self._neutron_id})
-
- @mock.patch('os_client_config.make_shade')
- def test_no_endpoint3(self, *args):
- args[0].return_value.search_services.return_value = [
- munch.Munch(id=self._neutron_id)]
- args[0].return_value.search_endpoints.return_value = [munch.Munch()]
- self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
- args[0].return_value.search_services.assert_called_once_with('neutron')
- args[0].return_value.search_endpoints.assert_called_once_with(
- filters={'interface': self._os_interface,
- 'service_id': self._neutron_id})
-
- @mock.patch('os_client_config.make_shade')
- def test_endpoint_interface(self, *args):
- args[0].return_value.search_services.return_value = [
- munch.Munch(id=self._neutron_id)]
- args[0].return_value.search_endpoints.return_value = [munch.Munch()]
- self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
- args[0].return_value.search_services.assert_called_once_with('neutron')
- args[0].return_value.search_endpoints.assert_called_once_with(
- filters={'interface': self._os_interface,
- 'service_id': self._neutron_id})
-
- @mock.patch('os_client_config.make_shade')
- def _test_no_env_var(self, var, *args):
- del os.environ[var]
- self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
- args[0].assert_called_once_with()
-
- @mock.patch('os_client_config.make_shade')
- def _test_missing_value(self, *args):
- self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
- args[0].assert_called_once_with()
-
- @mock.patch('os_client_config.make_shade')
- def _test_run(self, status=testcase.TestCase.EX_OK,
- exception=None, *args, **kwargs):
- # pylint: disable=keyword-arg-before-vararg
- args[0].return_value.search_services.return_value = [
- munch.Munch(id=self._neutron_id)]
- args[0].return_value.search_endpoints.return_value = [
- munch.Munch(url=self._neutron_url)]
- odlip = kwargs['odlip'] if 'odlip' in kwargs else '127.0.0.3'
- odlwebport = kwargs['odlwebport'] if 'odlwebport' in kwargs else '8080'
- odlrestconfport = (kwargs['odlrestconfport']
- if 'odlrestconfport' in kwargs else '8181')
- if exception:
- self.test.run_suites = mock.Mock(side_effect=exception)
- else:
- self.test.run_suites = mock.Mock(return_value=status)
- self.assertEqual(self.test.run(), status)
- self.test.run_suites.assert_called_once_with(
- odl.ODLTests.default_suites, neutronurl=self._neutron_url,
- odlip=odlip, odlpassword=self._odl_password,
- odlrestconfport=odlrestconfport, odlusername=self._odl_username,
- odlwebport=odlwebport, osauthurl=self._os_auth_url,
- ospassword=self._os_password, osprojectname=self._os_projectname,
- osusername=self._os_username,
- osprojectdomainname=self._os_projectdomainname,
- osuserdomainname=self._os_userdomainname)
- args[0].assert_called_once_with()
- args[0].return_value.search_services.assert_called_once_with('neutron')
- args[0].return_value.search_endpoints.assert_called_once_with(
- filters={
- 'interface': os.environ.get(
- "OS_INTERFACE", "public").replace('URL', ''),
- 'service_id': self._neutron_id})
-
- @mock.patch('os_client_config.make_shade')
- def _test_multiple_suites(self, suites,
- status=testcase.TestCase.EX_OK, *args, **kwargs):
- # pylint: disable=keyword-arg-before-vararg
- args[0].return_value.search_endpoints.return_value = [
- munch.Munch(url=self._neutron_url)]
- args[0].return_value.search_services.return_value = [
- munch.Munch(id=self._neutron_id)]
- odlip = kwargs['odlip'] if 'odlip' in kwargs else '127.0.0.3'
- odlwebport = kwargs['odlwebport'] if 'odlwebport' in kwargs else '8080'
- odlrestconfport = (kwargs['odlrestconfport']
- if 'odlrestconfport' in kwargs else '8181')
- self.test.run_suites = mock.Mock(return_value=status)
- self.assertEqual(self.test.run(suites=suites), status)
- self.test.run_suites.assert_called_once_with(
- suites, neutronurl=self._neutron_url, odlip=odlip,
- odlpassword=self._odl_password, odlrestconfport=odlrestconfport,
- odlusername=self._odl_username, odlwebport=odlwebport,
- osauthurl=self._os_auth_url, ospassword=self._os_password,
- osprojectname=self._os_projectname, osusername=self._os_username,
- osprojectdomainname=self._os_projectdomainname,
- osuserdomainname=self._os_userdomainname)
- args[0].assert_called_once_with()
- args[0].return_value.search_services.assert_called_once_with('neutron')
- args[0].return_value.search_endpoints.assert_called_once_with(
- filters={'interface': os.environ.get("OS_INTERFACE", "public"),
- 'service_id': self._neutron_id})
-
- def test_exc(self):
- with mock.patch('os_client_config.make_shade',
- side_effect=Exception()):
- self.assertEqual(self.test.run(),
- testcase.TestCase.EX_RUN_ERROR)
-
- def test_no_os_auth_url(self):
- self._test_no_env_var("OS_AUTH_URL")
-
- def test_no_os_username(self):
- self._test_no_env_var("OS_USERNAME")
-
- def test_no_os_password(self):
- self._test_no_env_var("OS_PASSWORD")
-
- def test_no_os__name(self):
- self._test_no_env_var("OS_PROJECT_NAME")
-
- def test_run_suites_false(self):
- os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
- self._test_run(testcase.TestCase.EX_RUN_ERROR, None,
- odlip=self._sdn_controller_ip,
- odlwebport=self._odl_webport)
-
- def test_run_suites_exc(self):
- with self.assertRaises(Exception):
- os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
- self._test_run(testcase.TestCase.EX_RUN_ERROR,
- Exception(),
- odlip=self._sdn_controller_ip,
- odlwebport=self._odl_webport)
-
- def test_no_sdn_controller_ip(self):
- self._test_missing_value()
-
- def test_without_installer_type(self):
- os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
- self._test_run(testcase.TestCase.EX_OK, None,
- odlip=self._sdn_controller_ip,
- odlwebport=self._odl_webport)
-
- def test_without_os_interface(self):
- del os.environ["OS_INTERFACE"]
- os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
- self._test_run(testcase.TestCase.EX_OK, None,
- odlip=self._sdn_controller_ip,
- odlwebport=self._odl_webport)
-
- def test_os_interface_public(self):
- os.environ["OS_INTERFACE"] = "public"
- os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
- self._test_run(testcase.TestCase.EX_OK, None,
- odlip=self._sdn_controller_ip,
- odlwebport=self._odl_webport)
-
- def test_os_interface_publicurl(self):
- os.environ["OS_INTERFACE"] = "publicURL"
- os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
- self._test_run(testcase.TestCase.EX_OK, None,
- odlip=self._sdn_controller_ip,
- odlwebport=self._odl_webport)
-
- def test_os_interface_internal(self):
- os.environ["OS_INTERFACE"] = "internal"
- os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
- self._test_run(testcase.TestCase.EX_OK, None,
- odlip=self._sdn_controller_ip,
- odlwebport=self._odl_webport)
-
- def test_os_interface_admin(self):
- os.environ["OS_INTERFACE"] = "admin"
- os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
- self._test_run(testcase.TestCase.EX_OK, None,
- odlip=self._sdn_controller_ip,
- odlwebport=self._odl_webport)
-
- def test_suites(self):
- os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
- self._test_multiple_suites(
- [odl.ODLTests.basic_suite_dir],
- testcase.TestCase.EX_OK,
- odlip=self._sdn_controller_ip,
- odlwebport=self._odl_webport)
-
-
-class ODLArgParserTesting(ODLTesting):
-
- """The class testing ODLParser."""
- # pylint: disable=missing-docstring
-
- def setUp(self):
- self.parser = odl.ODLParser()
- super().setUp()
-
- def test_default(self):
- self.assertEqual(self.parser.parse_args(), self.defaultargs)
-
- def test_basic(self):
- self.defaultargs['neutronurl'] = self._neutron_url
- self.defaultargs['odlip'] = self._sdn_controller_ip
- self.assertEqual(
- self.parser.parse_args(
- [f"--neutronurl={self._neutron_url}",
- f"--odlip={self._sdn_controller_ip}"]),
- self.defaultargs)
-
- @mock.patch('sys.stderr', new_callable=six.StringIO)
- def test_fail(self, mock_method):
- self.defaultargs['foo'] = 'bar'
- with self.assertRaises(SystemExit):
- self.parser.parse_args(["--foo=bar"])
- self.assertTrue(mock_method.getvalue().startswith("usage:"))
-
- def _test_arg(self, arg, value):
- self.defaultargs[arg] = value
- self.assertEqual(
- self.parser.parse_args([f"--{arg}={value}"]),
- self.defaultargs)
-
- def test_odlusername(self):
- self._test_arg('odlusername', 'foo')
-
- def test_odlpassword(self):
- self._test_arg('odlpassword', 'foo')
-
- def test_osauthurl(self):
- self._test_arg('osauthurl', 'http://127.0.0.4:5000/v2')
-
- def test_neutronurl(self):
- self._test_arg('neutronurl', 'http://127.0.0.4:9696')
-
- def test_osusername(self):
- self._test_arg('osusername', 'foo')
-
- def test_osuserdomainname(self):
- self._test_arg('osuserdomainname', 'domain')
-
- def test_osprojectname(self):
- self._test_arg('osprojectname', 'foo')
-
- def test_osprojectdomainname(self):
- self._test_arg('osprojectdomainname', 'domain')
-
- def test_ospassword(self):
- self._test_arg('ospassword', 'foo')
-
- def test_odlip(self):
- self._test_arg('odlip', '127.0.0.4')
-
- def test_odlwebport(self):
- self._test_arg('odlwebport', '80')
-
- def test_odlrestconfport(self):
- self._test_arg('odlrestconfport', '80')
-
- def test_pushtodb(self):
- self.defaultargs['pushtodb'] = True
- self.assertEqual(self.parser.parse_args(["--pushtodb"]),
- self.defaultargs)
-
- def test_multiple_args(self):
- self.defaultargs['neutronurl'] = self._neutron_url
- self.defaultargs['odlip'] = self._sdn_controller_ip
- self.assertEqual(
- self.parser.parse_args(
- [f"--neutronurl={self._neutron_url}",
- f"--odlip={self._sdn_controller_ip}"]),
- self.defaultargs)
-
-
-if __name__ == "__main__":
- logging.disable(logging.CRITICAL)
- unittest.main(verbosity=2)
Code Review / functest.git / commitdiff