Merge "Fix stack_get_by_name does not list nested stack issue"

heat/db/api.py

@@ -111,8 +111,13 @@ def stack_get(context, stack_id, show_deleted=False, tenant_safe=True):
                           tenant_safe=tenant_safe)
 
 
-def stack_get_by_name(context, stack_name, owner_id=None):
-    return IMPL.stack_get_by_name(context, stack_name, owner_id=owner_id)
+def stack_get_by_name_and_owner_id(context, stack_name, owner_id):
+    return IMPL.stack_get_by_name_and_owner_id(context, stack_name,
+                                               owner_id=owner_id)
+
+
+def stack_get_by_name(context, stack_name):
+    return IMPL.stack_get_by_name(context, stack_name)
 
 
 def stack_get_all(context):

heat/db/sqlalchemy/api.py

@@ -237,7 +237,7 @@ def resource_get_all_by_stack(context, stack_id):
     return results
 
 
-def stack_get_by_name(context, stack_name, owner_id=None):
+def stack_get_by_name_and_owner_id(context, stack_name, owner_id):
     query = soft_delete_aware_query(context, models.Stack).\
         filter_by(tenant=context.tenant_id).\
         filter_by(name=stack_name).\
@@ -246,6 +246,14 @@ def stack_get_by_name(context, stack_name, owner_id=None):
     return query.first()
 
 
+def stack_get_by_name(context, stack_name):
+    query = soft_delete_aware_query(context, models.Stack).\
+        filter_by(tenant=context.tenant_id).\
+        filter_by(name=stack_name)
+
+    return query.first()
+
+
 def stack_get(context, stack_id, show_deleted=False, tenant_safe=True):
     result = model_query(context, models.Stack).get(stack_id)
 

heat/engine/parser.py

@@ -427,8 +427,9 @@ class Stack(collections.Mapping):
         Get a Stack containing any in-progress resources from the previous
         stack state prior to an update.
         '''
-        s = db_api.stack_get_by_name(self.context, self._backup_name(),
-                                     owner_id=self.id)
+        s = db_api.stack_get_by_name_and_owner_id(self.context,
+                                                  self._backup_name(),
+                                                  owner_id=self.id)
         if s is not None:
             logger.debug(_('Loaded existing backup stack'))
             return self.load(self.context, stack=s)

heat/tests/test_sqlalchemy_api.py

@@ -95,12 +95,13 @@ class SqlAlchemyTest(HeatTestCase):
     def tearDown(self):
         super(SqlAlchemyTest, self).tearDown()
 
-    def _setup_test_stack(self, stack_name, stack_id=None):
+    def _setup_test_stack(self, stack_name, stack_id=None, owner_id=None):
         t = template_format.parse(wp_template)
         template = parser.Template(t)
         stack_id = stack_id or str(uuid.uuid4())
         stack = parser.Stack(self.ctx, stack_name, template,
-                             environment.Environment({'KeyName': 'test'}))
+                             environment.Environment({'KeyName': 'test'}),
+                             owner_id=owner_id)
         with utils.UUIDStub(stack_id):
             stack.store()
         return (t, stack)
@@ -176,6 +177,39 @@ class SqlAlchemyTest(HeatTestCase):
         st = db_api.stack_get_by_name(self.ctx, 'stack')
         self.assertIsNone(st)
 
+    def test_nested_stack_get_by_name(self):
+        stack1 = self._setup_test_stack('stack1', UUID1)[1]
+        stack2 = self._setup_test_stack('stack2', UUID2,
+                                        owner_id=stack1.id)[1]
+
+        result = db_api.stack_get_by_name(self.ctx, 'stack2')
+        self.assertEqual(UUID2, result.id)
+
+        stack2.delete()
+
+        result = db_api.stack_get_by_name(self.ctx, 'stack2')
+        self.assertIsNone(result)
+
+    def test_stack_get_by_name_and_owner_id(self):
+        stack1 = self._setup_test_stack('stack1', UUID1)[1]
+        stack2 = self._setup_test_stack('stack2', UUID2,
+                                        owner_id=stack1.id)[1]
+
+        result = db_api.stack_get_by_name_and_owner_id(self.ctx, 'stack2',
+                                                       None)
+        self.assertIsNone(result)
+
+        result = db_api.stack_get_by_name_and_owner_id(self.ctx, 'stack2',
+                                                       stack1.id)
+
+        self.assertEqual(UUID2, result.id)
+
+        stack2.delete()
+
+        result = db_api.stack_get_by_name_and_owner_id(self.ctx, 'stack2',
+                                                       stack1.id)
+        self.assertIsNone(result)
+
     def test_stack_get(self):
         stack = self._setup_test_stack('stack', UUID1)[1]