Update ODL testcase to OpenStack Shade

Change-Id: I7832b5ed274e181449f9db9fb03a2d27038520ae
Signed-off-by: Cédric Ollivier <cedric.ollivier@orange.com>

2 files changed, 132 insertions, 28 deletions

diff --git a/functest/opnfv_tests/sdn/odl/odl.py b/functest/opnfv_tests/sdn/odl/odl.py
index 7c61e88df..cc56c6202 100644
--- a/functest/opnfv_tests/sdn/odl/odl.py
+++ b/functest/opnfv_tests/sdn/odl/odl.py
@@ -25,10 +25,9 @@ import os
 import re
 import sys
 
+import os_client_config
 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 +156,12 @@ 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 = 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'),
+                         '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..d1dfea8a7 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,98 @@ 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.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('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.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()
-        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')
+    @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()
-        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')
+    @mock.patch('os_client_config.make_shade')
     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']
@@ -313,14 +380,18 @@ class ODLRunTesting(ODLTesting):
             osprojectdomainname=self._os_projectdomainname,
             osuserdomainname=self._os_userdomainname)
         args[0].assert_called_once_with()
-        args[1].assert_called_once_with(mock.ANY, 'network')
+        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('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.make_shade')
     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']
@@ -336,10 +407,13 @@ class ODLRunTesting(ODLTesting):
             osprojectdomainname=self._os_projectdomainname,
             osuserdomainname=self._os_userdomainname)
         args[0].assert_called_once_with()
-        args[1].assert_called_once_with(mock.ANY, 'network')
+        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('snaps.openstack.utils.keystone_utils.get_endpoint',
+        with mock.patch('os_client_config.make_shade',
                         side_effect=Exception()):
             self.assertEqual(self.test.run(),
                              testcase.TestCase.EX_RUN_ERROR)
@@ -379,6 +453,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(