Load existing resources using correct environment

In convergence we were loading resources from the database using the
current environment. This is incorrect when a previous update has
failed, meaning the resources in the database were created with a
non-current template and environment. If an attempt was made to change
the type of a resource but that resource was never updated, this will
result in us loading a resource with the wrong type. If the type has
been removed then it can result in errors just trying to show the stack.

Note that the Resource.load() method used during a convergence traversal
already does the Right Thing - it only uses the new type if it is a
valid substitution for the old type, and UpdateReplace is later raised
in Resource.update_convergence() if the type does not match in that
specified in the new environment. So we don't see any problems with
stack updates, just with API calls.

Since we cannot change the signature of Resource.__new__() without also
modifying the signature of __init__() in every resource plugin that has
implemented it (many of which are out of tree), instead substitute the
stack definition for the duration of creating the Resource object. This
will result in stack.env returning the environment the resource was last
updated with.

Change-Id: I3fbd14324fc4681b26747ee7505000b8fc9439f1
Story: #2005090
Task: 29688
(cherry picked from commit aa58fbcacfdc14bb2e8d5a4ff1ef7a47fe9268d0)

2 files changed, 47 insertions, 2 deletions

diff --git a/heat/engine/stack.py b/heat/engine/stack.py
index 2bb7cb238..72b2df7aa 100644
--- a/heat/engine/stack.py
+++ b/heat/engine/stack.py
@@ -12,6 +12,7 @@
 #    under the License.
 
 import collections
+import contextlib
 import copy
 import eventlet
 import functools
@@ -378,6 +379,15 @@ class Stack(collections.Mapping):
             self._db_resources = _db_resources
         return self._db_resources
 
+    @contextlib.contextmanager
+    def _previous_definition(self, stk_defn):
+        cur_defn = self.defn
+        try:
+            self.defn = stk_defn
+            yield
+        finally:
+            self.defn = cur_defn
+
     def _resource_from_db_resource(self, db_res, stk_def_cache=None):
         tid = db_res.current_template_id
         if tid is None:
@@ -406,8 +416,9 @@ class Stack(collections.Mapping):
         except KeyError:
             return None
 
-        res = resource.Resource(db_res.name, defn, self)
-        res._load_data(db_res)
+        with self._previous_definition(stk_def):
+            res = resource.Resource(db_res.name, defn, self)
+            res._load_data(db_res)
         return res
 
     def resource_get(self, name):
diff --git a/heat_integrationtests/functional/test_create_update.py b/heat_integrationtests/functional/test_create_update.py
index b5793685f..3ad403221 100644
--- a/heat_integrationtests/functional/test_create_update.py
+++ b/heat_integrationtests/functional/test_create_update.py
@@ -766,3 +766,37 @@ resources:
                           expected_status='UPDATE_FAILED')
         self.update_stack(stack_identifier, template=template,
                           expected_status='UPDATE_COMPLETE')
+
+    @test.requires_convergence
+    def test_update_failed_changed_env_list_resources(self):
+        template = {
+            'heat_template_version': 'rocky',
+            'resources': {
+                'test1': {
+                    'type': 'OS::Heat::TestResource',
+                    'properties': {
+                        'value': 'foo'
+                    }
+                },
+                'my_res': {
+                    'type': 'My::TestResource',
+                    'depends_on': 'test1'
+                },
+                'test2': {
+                    'depends_on': 'my_res',
+                    'type': 'OS::Heat::TestResource'
+                }
+            }
+        }
+        env = {'resource_registry':
+               {'My::TestResource': 'OS::Heat::TestResource'}}
+        stack_identifier = self.stack_create(
+            template=template, environment=env)
+        update_template = copy.deepcopy(template)
+        update_template['resources']['test1']['properties']['fail'] = 'true'
+        update_template['resources']['test2']['depends_on'] = 'test1'
+        del update_template['resources']['my_res']
+        self.update_stack(stack_identifier,
+                          template=update_template,
+                          expected_status='UPDATE_FAILED')
+        self.assertEqual(3, len(self.list_resources(stack_identifier)))