Get neutron endpoint via snaps in odl

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

diff --git a/functest/opnfv_tests/sdn/odl/odl.py b/functest/opnfv_tests/sdn/odl/odl.py
index f25dfe5..25cc758 100644 (file)
--- a/functest/opnfv_tests/sdn/odl/odl.py
+++ b/functest/opnfv_tests/sdn/odl/odl.py
@@ -26,10 +26,11 @@ import re
 import sys
 
 from six.moves import urllib
+from snaps.openstack.utils import keystone_utils
 
 from functest.core import robotframework
+from functest.opnfv_tests.openstack.snaps import snaps_utils
 from functest.utils import constants
-import functest.utils.openstack_utils as op_utils
 
 __author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
 
@@ -155,8 +156,9 @@ class ODLTests(robotframework.RobotFramework):
                 suites = kwargs["suites"]
             except KeyError:
                 pass
-            kwargs = {'neutronurl': op_utils.get_endpoint(
-                service_type='network')}
+            snaps_creds = snaps_utils.get_credentials()
+            kwargs = {'neutronurl': keystone_utils.get_endpoint(
+                snaps_creds, 'network')}
             kwargs['odlip'] = urllib.parse.urlparse(
                 kwargs['neutronurl']).hostname
             kwargs['odlwebport'] = '8080'
@@ -274,7 +276,6 @@ def main():
             return result
         if args['pushtodb']:
             return odl.push_to_db()
-        else:
-            return result
+        return result
     except Exception:  # pylint: disable=broad-except
         return robotframework.RobotFramework.EX_RUN_ERROR

diff --git a/functest/tests/unit/odl/test_odl.py b/functest/tests/unit/odl/test_odl.py
index b93ad31..d803d41 100644 (file)
--- a/functest/tests/unit/odl/test_odl.py
+++ b/functest/tests/unit/odl/test_odl.py
@@ -33,7 +33,7 @@ class ODLTesting(unittest.TestCase):
     logging.disable(logging.CRITICAL)
 
     _keystone_ip = "127.0.0.1"
-    _neutron_url = "http://127.0.0.2:9696"
+    _neutron_url = u"https://127.0.0.1:9696"
     _sdn_controller_ip = "127.0.0.3"
     _os_auth_url = "http://{}:5000/v3".format(_keystone_ip)
     _os_projectname = "admin"
@@ -269,65 +269,68 @@ class ODLRunTesting(ODLTesting):
     """The class testing ODLTests.run()."""
     # pylint: disable=missing-docstring
 
-    def _test_no_env_var(self, var):
-        with mock.patch('functest.utils.openstack_utils.get_endpoint',
-                        return_value=ODLTesting._neutron_url):
-            del os.environ[var]
-            self.assertEqual(self.test.run(),
-                             testcase.TestCase.EX_RUN_ERROR)
-
+    @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')
+    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')
     def _test_run(self, status=testcase.TestCase.EX_OK,
-                  exception=None, **kwargs):
+                  exception=None, *args, **kwargs):
         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')
-
-        with mock.patch('functest.utils.openstack_utils.get_endpoint',
-                        return_value=ODLTesting._neutron_url):
-            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)
-
+        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[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')
     def _test_multiple_suites(self, suites,
-                              status=testcase.TestCase.EX_OK, **kwargs):
+                              status=testcase.TestCase.EX_OK, *args, **kwargs):
         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')
-        with mock.patch('functest.utils.openstack_utils.get_endpoint',
-                        return_value=ODLTesting._neutron_url):
-            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)
+        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[1].assert_called_once_with(mock.ANY, 'network')
 
     def test_exc(self):
-        with mock.patch('functest.utils.openstack_utils.get_endpoint',
+        with mock.patch('snaps.openstack.utils.keystone_utils.get_endpoint',
                         side_effect=auth_plugins.MissingAuthPlugin()):
             self.assertEqual(self.test.run(),
                              testcase.TestCase.EX_RUN_ERROR)
@@ -346,27 +349,24 @@ class ODLRunTesting(ODLTesting):
 
     def test_run_suites_false(self):
         os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
-        self._test_run(testcase.TestCase.EX_RUN_ERROR,
+        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(status=testcase.TestCase.EX_RUN_ERROR,
-                           exception=Exception(),
+            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):
-        with mock.patch('functest.utils.openstack_utils.get_endpoint',
-                        return_value=ODLTesting._neutron_url):
-            self.assertEqual(self.test.run(),
-                             testcase.TestCase.EX_RUN_ERROR)
+        self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
 
     def test_without_installer_type(self):
         os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
-        self._test_run(testcase.TestCase.EX_OK,
+        self._test_run(testcase.TestCase.EX_OK, None,
                        odlip=self._sdn_controller_ip,
                        odlwebport=self._odl_webport)
 
@@ -380,69 +380,57 @@ class ODLRunTesting(ODLTesting):
 
     def test_fuel(self):
         os.environ["INSTALLER_TYPE"] = "fuel"
-        self._test_run(testcase.TestCase.EX_OK,
+        self._test_run(testcase.TestCase.EX_OK, None,
                        odlip=urllib.parse.urlparse(self._neutron_url).hostname,
                        odlwebport='8181',
                        odlrestconfport='8282')
 
     def test_apex_no_controller_ip(self):
-        with mock.patch('functest.utils.openstack_utils.get_endpoint',
-                        return_value=ODLTesting._neutron_url):
-            os.environ["INSTALLER_TYPE"] = "apex"
-            self.assertEqual(self.test.run(),
-                             testcase.TestCase.EX_RUN_ERROR)
+        os.environ["INSTALLER_TYPE"] = "apex"
+        self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
 
     def test_apex(self):
         os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
         os.environ["INSTALLER_TYPE"] = "apex"
-        self._test_run(testcase.TestCase.EX_OK,
+        self._test_run(testcase.TestCase.EX_OK, None,
                        odlip=self._sdn_controller_ip, odlwebport='8081',
                        odlrestconfport='8081')
 
     def test_netvirt_no_controller_ip(self):
-        with mock.patch('functest.utils.openstack_utils.get_endpoint',
-                        return_value=ODLTesting._neutron_url):
-            os.environ["INSTALLER_TYPE"] = "netvirt"
-            self.assertEqual(self.test.run(),
-                             testcase.TestCase.EX_RUN_ERROR)
+        os.environ["INSTALLER_TYPE"] = "netvirt"
+        self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
 
     def test_netvirt(self):
         os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
         os.environ["INSTALLER_TYPE"] = "netvirt"
-        self._test_run(testcase.TestCase.EX_OK,
+        self._test_run(testcase.TestCase.EX_OK, None,
                        odlip=self._sdn_controller_ip, odlwebport='8081',
                        odlrestconfport='8081')
 
     def test_joid_no_controller_ip(self):
-        with mock.patch('functest.utils.openstack_utils.get_endpoint',
-                        return_value=ODLTesting._neutron_url):
-            os.environ["INSTALLER_TYPE"] = "joid"
-            self.assertEqual(self.test.run(),
-                             testcase.TestCase.EX_RUN_ERROR)
+        os.environ["INSTALLER_TYPE"] = "joid"
+        self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
 
     def test_joid(self):
         os.environ["SDN_CONTROLLER"] = self._sdn_controller_ip
         os.environ["INSTALLER_TYPE"] = "joid"
-        self._test_run(testcase.TestCase.EX_OK,
+        self._test_run(testcase.TestCase.EX_OK, None,
                        odlip=self._sdn_controller_ip, odlwebport='8080')
 
     def test_compass(self):
         os.environ["INSTALLER_TYPE"] = "compass"
-        self._test_run(testcase.TestCase.EX_OK,
+        self._test_run(testcase.TestCase.EX_OK, None,
                        odlip=urllib.parse.urlparse(self._neutron_url).hostname,
                        odlrestconfport='8080')
 
     def test_daisy_no_controller_ip(self):
-        with mock.patch('functest.utils.openstack_utils.get_endpoint',
-                        return_value=ODLTesting._neutron_url):
-            os.environ["INSTALLER_TYPE"] = "daisy"
-            self.assertEqual(self.test.run(),
-                             testcase.TestCase.EX_RUN_ERROR)
+        os.environ["INSTALLER_TYPE"] = "daisy"
+        self.assertEqual(self.test.run(), testcase.TestCase.EX_RUN_ERROR)
 
     def test_daisy(self):
         os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
         os.environ["INSTALLER_TYPE"] = "daisy"
-        self._test_run(testcase.TestCase.EX_OK,
+        self._test_run(testcase.TestCase.EX_OK, None,
                        odlip=self._sdn_controller_ip, odlwebport='8181',
                        odlrestconfport='8087')