Add DB API to get the count of stacks per tenant

This method will be used to limit the number of stacks any one tenant
can create.

Change-Id: Ic990bf84fc1aa5592504f7c423fbce953615a4a1

diff --git a/heat/db/api.py b/heat/db/api.py
index aea21db85159a4e45176b491ca80c0e0e0367cbe..b932063bd4513ab013d8e144b908248b8894b1e1 100644 (file)
--- a/heat/db/api.py
+++ b/heat/db/api.py
@@ -129,6 +129,10 @@ def stack_get_all_by_tenant(context):
     return IMPL.stack_get_all_by_tenant(context)
 
 
+def stack_count_all_by_tenant(context):
+    return IMPL.stack_count_all_by_tenant(context)
+
+
 def stack_create(context, values):
     return IMPL.stack_create(context, values)
 

diff --git a/heat/db/sqlalchemy/api.py b/heat/db/sqlalchemy/api.py
index b6844f1138d2fd0790923d87c0006340bba5ac73..2fb86b9c3d8a9f57d9f5c711eced7c1d5c86b902 100644 (file)
--- a/heat/db/sqlalchemy/api.py
+++ b/heat/db/sqlalchemy/api.py
@@ -213,11 +213,19 @@ def stack_get_all(context):
     return results
 
 
-def stack_get_all_by_tenant(context):
-    results = soft_delete_aware_query(context, models.Stack).\
+def _query_stack_get_all_by_tenant(context):
+    query = soft_delete_aware_query(context, models.Stack).\
         filter_by(owner_id=None).\
-        filter_by(tenant=context.tenant_id).all()
-    return results
+        filter_by(tenant=context.tenant_id)
+    return query
+
+
+def stack_get_all_by_tenant(context):
+    return _query_stack_get_all_by_tenant(context).all()
+
+
+def stack_count_all_by_tenant(context):
+    return _query_stack_get_all_by_tenant(context).count()
 
 
 def stack_create(context, values):

diff --git a/heat/tests/test_sqlalchemy_api.py b/heat/tests/test_sqlalchemy_api.py
index 6a0845f641db3781089cc66633b7a53acf4851cb..2bc2cd4bf70024d2ddda360bab2c86fbde5f85e6 100644 (file)
--- a/heat/tests/test_sqlalchemy_api.py
+++ b/heat/tests/test_sqlalchemy_api.py
@@ -203,6 +203,20 @@ class SqlAlchemyTest(HeatTestCase):
         st_db = db_api.stack_get_all_by_tenant(self.ctx)
         self.assertEqual(0, len(st_db))
 
+    def test_stack_count_all_by_tenant(self):
+        stacks = [self._setup_test_stack('stack', x)[1] for x in UUIDs]
+
+        st_db = db_api.stack_count_all_by_tenant(self.ctx)
+        self.assertEqual(2, st_db)
+
+        stacks[0].delete()
+        st_db = db_api.stack_count_all_by_tenant(self.ctx)
+        self.assertEqual(1, st_db)
+
+        stacks[1].delete()
+        st_db = db_api.stack_count_all_by_tenant(self.ctx)
+        self.assertEqual(0, st_db)
+
     def test_event_get_all_by_stack(self):
         stack = self._setup_test_stack('stack', UUID1)[1]