Add support to get glance endpoint

This patch will help to get glance endpoint form service
catalog based on input task.

Partially implements: blueprint glance-download-import
Co-Authored-By: Victor Coutellier <victor.coutellier@gmail.com>
Co-Authored-By: Pierre-Samuel Le Stang <pierre-samuel.le-stang@corp.ovh.com>

Change-Id: Ib4f19351ca7985474d838998b402d3c5f9fb195f

glance/async_/utils.py

@@ -19,6 +19,7 @@ from oslo_utils import encodeutils
 from oslo_utils import units
 from taskflow import task
 
+from glance.common import exception as glance_exception
 from glance.i18n import _LW
 
 
@@ -77,3 +78,20 @@ class OptionalTask(task.Task):
                        encodeutils.exception_to_unicode(exc))
                 LOG.warning(msg)
         return wrapper
+
+
+def get_glance_endpoint(context, region, interface):
+    """Return glance endpoint depending the input tasks
+
+    """
+    # We use the current context to retrieve the image
+    catalog = context.service_catalog
+
+    for service in catalog:
+        if service['type'] == 'image':
+            for endpoint in service['endpoints']:
+                if endpoint['region'].lower() == region.lower():
+                    return endpoint.get('%sURL' % interface)
+
+    raise glance_exception.GlanceEndpointNotFound(region=region,
+                                                  interface=interface)

glance/common/exception.py

@@ -453,3 +453,8 @@ class InvalidDataMigrationScript(GlanceException):
     message = _("Invalid data migration script '%(script)s'. A valid data "
                 "migration script must implement functions 'has_migrations' "
                 "and 'migrate'.")
+
+
+class GlanceEndpointNotFound(NotFound):
+    message = _("%(interface)s glance endpoint not "
+                "found for region %(region)s")

glance/tests/unit/async_/test_utils.py

@@ -0,0 +1,80 @@
+# Copyright 2022 OVHcloud
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+
+from unittest import mock
+
+from glance.async_ import utils
+import glance.common.exception
+from glance.tests.unit import base
+
+
+class TestGetGlanceEndpoint(base.IsolatedUnitTest):
+
+    def setUp(self):
+        super(TestGetGlanceEndpoint, self).setUp()
+
+        self.service_catalog = [
+            {
+                'endpoints': [
+                    {
+                        'adminURL': 'http://localhost:8080/',
+                        'region': 'RegionOne',
+                        'internalURL': 'http://internalURL/',
+                        'publicURL': 'http://publicURL/',
+                    },
+                ],
+                'type': 'object-store',
+            },
+            {
+                'endpoints': [
+                    {
+                        'adminURL': 'http://localhost:8080/',
+                        'region': 'RegionOne',
+                        'internalURL': 'http://RegionOneInternal/',
+                        'publicURL': 'http://RegionOnePublic/',
+                    },
+                ],
+                'type': 'image',
+            },
+            {
+                'endpoints': [
+                    {
+                        'adminURL': 'http://localhost:8080/',
+                        'region': 'RegionTwo',
+                        'internalURL': 'http://RegionTwoInternal/',
+                        'publicURL': 'http://RegionTwoPublic/',
+                    },
+                ],
+                'type': 'image',
+            }
+        ]
+
+        self.context = mock.MagicMock(service_catalog=self.service_catalog)
+
+    def test_return_matching_glance_endpoint(self):
+        self.assertEqual(utils.get_glance_endpoint(self.context,
+                                                   'RegionOne',
+                                                   'public'),
+                         'http://RegionOnePublic/')
+        self.assertEqual(utils.get_glance_endpoint(self.context,
+                                                   'RegionTwo',
+                                                   'internal'),
+                         'http://RegionTwoInternal/')
+
+    def test_glance_endpoint_not_found(self):
+        self.assertRaises(glance.common.exception.GlanceEndpointNotFound,
+                          utils.get_glance_endpoint, self.context,
+                          'RegionThree', 'public')