Filter List Regions by 'parent_region_id'

List Regions (GET /regions) has an optional query parameter
'parent_region_id', but the API still returns all the regions.

This fix adds support of filtering regions by an optional query
parameter 'parent_region_id'.

Closes-Bug: #1350192

Change-Id: I59f8d852870bd3ad26622bfcd00dfadfdf9d5b42

keystone/catalog/backends/kvs.py

@@ -14,6 +14,7 @@
 
 
 from keystone import catalog
+from keystone.common import driver_hints
 from keystone.common import kvs
 
 
@@ -30,7 +31,7 @@ def _delete_child_regions(self, region_id):
         Recursively delete any region that has the supplied region
         as its parent.
         """
-        children = [r for r in self.list_regions()
+        children = [r for r in self.list_regions(driver_hints.Hints())
                     if r['parent_region_id'] == region_id]
         for child in children:
             self._delete_child_regions(child['id'])
@@ -58,7 +59,7 @@ def create_region(self, region):
         self.db.set('region_list', list(region_list))
         return region
 
-    def list_regions(self):
+    def list_regions(self, hints):
         return [self.get_region(x) for x in self.db.get('region_list', [])]
 
     def get_region(self, region_id):

keystone/catalog/backends/sql.py

@@ -82,9 +82,10 @@ class Endpoint(sql.ModelBase, sql.DictBase):
 class Catalog(catalog.Driver):
 
     # Regions
-    def list_regions(self):
+    def list_regions(self, hints):
         session = sql.get_session()
-        regions = session.query(Region).all()
+        regions = session.query(Region)
+        regions = sql.filter_limit_query(Region, regions, hints)
         return [s.to_dict() for s in list(regions)]
 
     def _get_region(self, session, region_id):

keystone/catalog/controllers.py

@@ -201,10 +201,11 @@ def create_region(self, context, region):
             RegionV3.wrap_member(context, ref),
             status=(201, 'Created'))
 
-    @controller.protected()
-    def list_regions(self, context):
-        refs = self.catalog_api.list_regions()
-        return RegionV3.wrap_collection(context, refs)
+    @controller.filterprotected('parent_region_id')
+    def list_regions(self, context, filters):
+        hints = RegionV3.build_driver_hints(context, filters)
+        refs = self.catalog_api.list_regions(hints)
+        return RegionV3.wrap_collection(context, refs, hints=hints)
 
     @controller.protected()
     def get_region(self, context, region_id):

keystone/catalog/core.py

@@ -112,6 +112,10 @@ def delete_region(self, region_id):
         except exception.NotFound:
             raise exception.RegionNotFound(region_id=region_id)
 
+    @manager.response_truncated
+    def list_regions(self, hints=None):
+        return self.driver.list_regions(hints or driver_hints.Hints())
+
     def create_service(self, service_id, service_ref):
         service_ref.setdefault('enabled', True)
         return self.driver.create_service(service_id, service_ref)
@@ -193,9 +197,13 @@ def create_region(self, region_ref):
         raise exception.NotImplemented()  # pragma: no cover
 
     @abc.abstractmethod
-    def list_regions(self):
+    def list_regions(self, hints):
         """List all regions.
 
+        :param hints: contains the list of filters yet to be satisfied.
+                      Any filters satisfied here will be removed so that
+                      the caller will know if any filters remain.
+
         :returns: list of region_refs or an empty list.
 
         """

keystone/tests/test_backend.py

@@ -3583,6 +3583,29 @@ def test_region_crud(self):
                           self.catalog_api.get_region,
                           region_id)
 
+    def _create_region_with_parent_id(self, parent_id=None):
+        new_region = {
+            'id': uuid.uuid4().hex,
+            'description': uuid.uuid4().hex,
+            'parent_region_id': parent_id
+        }
+        self.catalog_api.create_region(
+            new_region)
+        return new_region
+
+    def test_list_regions_filtered_by_parent_region_id(self):
+        new_region = self._create_region_with_parent_id()
+        parent_id = new_region['id']
+        new_region = self._create_region_with_parent_id(parent_id)
+        new_region = self._create_region_with_parent_id(parent_id)
+
+        # filter by parent_region_id
+        hints = driver_hints.Hints()
+        hints.add_filter('parent_region_id', parent_id)
+        regions = self.catalog_api.list_regions(hints)
+        for region in regions:
+            self.assertEqual(parent_id, region['parent_region_id'])
+
     @tests.skip_if_cache_disabled('catalog')
     def test_cache_layer_region_crud(self):
         region_id = uuid.uuid4().hex

keystone/tests/test_backend_kvs.py

@@ -229,6 +229,9 @@ def test_get_v3_catalog_endpoint_disabled(self):
         f = super(KvsCatalog, self).test_get_v3_catalog_endpoint_disabled
         self.assertRaises(exception.NotFound, f)
 
+    def test_list_regions_filtered_by_parent_region_id(self):
+        self.skipTest('KVS backend does not support hints')
+
 
 class KvsTokenCacheInvalidation(tests.TestCase,
                                 test_backend.TokenCacheInvalidation):

keystone/tests/test_backend_templated.py

@@ -111,3 +111,6 @@ def test_get_v3_catalog(self):
              'name': "'Identity Service'",
              'id': '1'}]
         self.assertEqual(exp_catalog, catalog_ref)
+
+    def test_list_regions_filtered_by_parent_region_id(self):
+        self.skipTest('Templated backend does not support hints')

keystone/tests/test_v3_catalog.py

@@ -215,6 +215,26 @@ def test_list_regions(self):
         r = self.get('/regions')
         self.assertValidRegionListResponse(r, ref=self.region)
 
+    def _create_region_with_parent_id(self, parent_id=None):
+        ref = self.new_region_ref()
+        ref['parent_region_id'] = parent_id
+        return self.post(
+            '/regions',
+            body={'region': ref})
+
+    def test_list_regions_filtered_by_parent_region_id(self):
+        """Call ``GET /regions?parent_region_id={parent_region_id}``."""
+        new_region = self._create_region_with_parent_id()
+        parent_id = new_region.result['region']['id']
+
+        new_region = self._create_region_with_parent_id(parent_id)
+        new_region = self._create_region_with_parent_id(parent_id)
+
+        r = self.get('/regions?parent_region_id=%s' % parent_id)
+
+        for region in r.result['regions']:
+            self.assertEqual(parent_id, region['parent_region_id'])
+
     def test_list_regions_xml(self):
         """Call ``GET /regions (xml data)``."""
         r = self.get('/regions', content_type='xml')