summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--functest/opnfv_tests/sdn/odl/odl.py14
-rw-r--r--functest/tests/unit/odl/test_odl.py176
2 files changed, 155 insertions, 35 deletions
diff --git a/functest/opnfv_tests/sdn/odl/odl.py b/functest/opnfv_tests/sdn/odl/odl.py
index 7c61e88df..a975a081f 100644
--- a/functest/opnfv_tests/sdn/odl/odl.py
+++ b/functest/opnfv_tests/sdn/odl/odl.py
@@ -25,10 +25,10 @@ import os
import re
import sys
+import os_client_config
+import shade
from six.moves import urllib
-from snaps.openstack.utils import keystone_utils
-from functest.opnfv_tests.openstack.snaps import snaps_utils
from functest.utils import config
from functest.utils import env
from xtesting.core import robotframework
@@ -157,9 +157,13 @@ class ODLTests(robotframework.RobotFramework):
suites = kwargs["suites"]
except KeyError:
pass
- snaps_creds = snaps_utils.get_credentials()
- kwargs = {'neutronurl': keystone_utils.get_endpoint(
- snaps_creds, 'network')}
+ cloud = shade.OperatorCloud(
+ cloud_config=os_client_config.get_config())
+ neutron_id = cloud.search_services('neutron')[0].id
+ endpoint = cloud.search_endpoints(
+ filters={'interface': os.environ.get('OS_INTERFACE', 'public'),
+ 'service_id': neutron_id})[0].url
+ kwargs = {'neutronurl': endpoint}
kwargs['odlip'] = env.get('SDN_CONTROLLER_IP')
kwargs['odlwebport'] = '8080'
kwargs['odlrestconfport'] = '8181'
diff --git a/functest/tests/unit/odl/test_odl.py b/functest/tests/unit/odl/test_odl.py
index 6304d37a3..62296c01c 100644
--- a/functest/tests/unit/odl/test_odl.py
+++ b/functest/tests/unit/odl/test_odl.py
@@ -14,6 +14,7 @@ import os
import unittest
import mock
+import munch
from robot.errors import RobotError
import six
from six.moves import urllib
@@ -33,6 +34,7 @@ class ODLTesting(unittest.TestCase):
_keystone_ip = "127.0.0.1"
_neutron_url = u"https://127.0.0.1:9696"
+ _neutron_id = u"dummy"
_sdn_controller_ip = "127.0.0.3"
_os_auth_url = "http://{}:5000/v3".format(_keystone_ip)
_os_projectname = "admin"
@@ -44,6 +46,7 @@ class ODLTesting(unittest.TestCase):
_odl_password = "admin"
_os_userdomainname = 'Default'
_os_projectdomainname = 'Default'
+ _os_interface = "public"
def setUp(self):
for var in ("INSTALLER_TYPE", "SDN_CONTROLLER", "SDN_CONTROLLER_IP"):
@@ -56,6 +59,7 @@ class ODLTesting(unittest.TestCase):
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,
@@ -265,35 +269,109 @@ class ODLMainTesting(ODLTesting):
class ODLRunTesting(ODLTesting):
-
"""The class testing ODLTests.run()."""
- # pylint: disable=missing-docstring
+ # pylint: disable=too-many-public-methods,missing-docstring
+
+ @mock.patch('os_client_config.get_config')
+ @mock.patch('shade.OperatorCloud', 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(cloud_config=mock.ANY)
+
+ @mock.patch('os_client_config.get_config')
+ @mock.patch('shade.OperatorCloud')
+ 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.get_config')
+ @mock.patch('shade.OperatorCloud')
+ 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.get_config')
+ @mock.patch('shade.OperatorCloud')
+ 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.get_config')
+ @mock.patch('shade.OperatorCloud')
+ 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.get_config')
+ @mock.patch('shade.OperatorCloud')
+ 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.get_config')
+ @mock.patch('shade.OperatorCloud')
+ 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.get_config')
+ @mock.patch('shade.OperatorCloud')
+ 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('snaps.openstack.utils.keystone_utils.get_endpoint',
- return_value=ODLTesting._neutron_url)
- @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_utils.'
- 'get_credentials')
+ @mock.patch('os_client_config.get_config')
+ @mock.patch('shade.OperatorCloud')
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()
- args[1].assert_called_once_with(mock.ANY, 'network')
+ args[0].assert_called_once_with(cloud_config=mock.ANY)
- @mock.patch('snaps.openstack.utils.keystone_utils.get_endpoint',
- return_value=ODLTesting._neutron_url)
- @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_utils.'
- 'get_credentials')
+ @mock.patch('os_client_config.get_config')
+ @mock.patch('shade.OperatorCloud')
def _test_missing_value(self, *args):
self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
- args[0].assert_called_once_with()
- args[1].assert_called_once_with(mock.ANY, 'network')
+ args[0].assert_called_once_with(cloud_config=mock.ANY)
- @mock.patch('snaps.openstack.utils.keystone_utils.get_endpoint',
- return_value=ODLTesting._neutron_url)
- @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_utils.'
- 'get_credentials')
+ @mock.patch('os_client_config.get_config')
+ @mock.patch('shade.OperatorCloud')
def _test_run(self, status=testcase.TestCase.EX_OK,
exception=None, *args, **kwargs):
+ 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']
@@ -312,15 +390,20 @@ class ODLRunTesting(ODLTesting):
osusername=self._os_username,
osprojectdomainname=self._os_projectdomainname,
osuserdomainname=self._os_userdomainname)
- args[0].assert_called_once_with()
- args[1].assert_called_once_with(mock.ANY, 'network')
-
- @mock.patch('snaps.openstack.utils.keystone_utils.get_endpoint',
- return_value=ODLTesting._neutron_url)
- @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_utils.'
- 'get_credentials')
+ args[0].assert_called_once_with(cloud_config=mock.ANY)
+ 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})
+
+ @mock.patch('os_client_config.get_config')
+ @mock.patch('shade.OperatorCloud')
def _test_multiple_suites(self, suites,
status=testcase.TestCase.EX_OK, *args, **kwargs):
+ 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']
@@ -335,14 +418,19 @@ class ODLRunTesting(ODLTesting):
osprojectname=self._os_projectname, osusername=self._os_username,
osprojectdomainname=self._os_projectdomainname,
osuserdomainname=self._os_userdomainname)
- args[0].assert_called_once_with()
- args[1].assert_called_once_with(mock.ANY, 'network')
-
- def test_exc(self):
- with mock.patch('snaps.openstack.utils.keystone_utils.get_endpoint',
+ args[0].assert_called_once_with(cloud_config=mock.ANY)
+ 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})
+
+ @mock.patch('os_client_config.get_config')
+ def test_exc(self, *args):
+ with mock.patch('shade.OperatorCloud',
side_effect=Exception()):
self.assertEqual(self.test.run(),
testcase.TestCase.EX_RUN_ERROR)
+ args[0].assert_called_once_with()
def test_no_os_auth_url(self):
self._test_no_env_var("OS_AUTH_URL")
@@ -379,6 +467,34 @@ class ODLRunTesting(ODLTesting):
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_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(