Convergence resource operations

Updates enging resource module with required
wrapper functions for supporting convergence.

Implements: blueprint convergence-resource-operations

Change-Id: I4c69d7a8e52655f354d9a1feea81e6744a626f8a

4 files changed, 151 insertions, 23 deletions

diff --git a/heat/engine/resource.py b/heat/engine/resource.py
index 8bb9f327c..721149462 100644
--- a/heat/engine/resource.py
+++ b/heat/engine/resource.py
@@ -294,13 +294,6 @@ class Resource(object):
         rs.update_and_save({'rsrc_metadata': metadata})
         self._rsrc_metadata = metadata
 
-    def clear_requirers(self, gone_requires):
-        self.requires = set(self.requires) - set(gone_requires)
-        self.requires = list(self.requires)
-        self._store_or_update(self.action,
-                              self.status,
-                              self.status_reason)
-
     @classmethod
     def set_needed_by(cls, db_rsrc, needed_by):
         if db_rsrc:
@@ -619,6 +612,25 @@ class Resource(object):
         '''
         return self
 
+    def create_convergence(self, template_id, resource_data):
+        '''
+        Creates the resource by invoking the scheduler TaskRunner
+        and it persists the resource's current_template_id to template_id and
+        resource's requires to list of the required resource id from the
+        given resource_data.
+        '''
+
+        runner = scheduler.TaskRunner(self.create)
+        runner()
+
+        # update the resource db record
+        self.current_template_id = template_id
+        self.requires = (list({graph_key[0]
+                         for graph_key, data in resource_data.items()}))
+        self._store_or_update(self.action,
+                              self.status,
+                              self.status_reason)
+
     @scheduler.wrappertask
     def create(self):
         '''
@@ -759,6 +771,32 @@ class Resource(object):
         except ValueError:
             return True
 
+    def update_convergence(self, template_id, resource_data):
+        '''
+        Updates the resource by invoking the scheduler TaskRunner
+        and it persists the resource's current_template_id to template_id and
+        resource's requires to list of the required resource id from the
+        given resource_data and existing resource's requires.
+        '''
+
+        if self.status == self.IN_PROGRESS:
+            ex = UpdateInProgress(self.name)
+            LOG.exception(ex)
+            raise ex
+
+        # update the resource
+        runner = scheduler.TaskRunner(self.update, self.t)
+        runner()
+
+        # update the resource db record
+        self.current_template_id = template_id
+        current_requires = {graph_key[0]
+                            for graph_key, data in resource_data.items()}
+        self.requires = (list(set(self.requires) | current_requires))
+        self._store_or_update(self.action,
+                              self.status,
+                              self.status_reason)
+
     @scheduler.wrappertask
     def update(self, after, before=None, prev_resource=None):
         '''
@@ -949,6 +987,31 @@ class Resource(object):
                 msg = _('"%s" deletion policy not supported') % policy
                 raise exception.StackValidationFailed(message=msg)
 
+    def delete_convergence(self, template_id, resource_data):
+        '''
+        Deletes the resource by invoking the scheduler TaskRunner
+        and it persists the resource's current_template_id to template_id and
+        resource's requires to list of the required resource id from the
+        given resource_data and existing resource's requires.
+        '''
+        if self.status == self.IN_PROGRESS:
+            ex = UpdateInProgress(self.name)
+            LOG.exception(ex)
+            raise ex
+
+        # delete the resource
+        runner = scheduler.TaskRunner(self.delete)
+        runner()
+
+        # update the resource db record
+        self.current_template_id = template_id
+        current_requires = {graph_key[0]
+                            for graph_key, data in resource_data.items()}
+        self.requires = (list(set(self.requires) - current_requires))
+        self._store_or_update(self.action,
+                              self.status,
+                              self.status_reason)
+
     @scheduler.wrappertask
     def delete(self):
         '''
diff --git a/heat/engine/worker.py b/heat/engine/worker.py
index 3fcad9334..73209d7f7 100644
--- a/heat/engine/worker.py
+++ b/heat/engine/worker.py
@@ -210,18 +210,15 @@ def check_resource_update(rsrc, template_id, data):
     input_data = {in_data.name: in_data for in_data in data.values()}
 
     if rsrc.resource_id is None:
-        rsrc.create(template_id, input_data)
+        rsrc.create_convergence(template_id, input_data)
     else:
-        rsrc.update(template_id, input_data)
+        rsrc.update_convergence(template_id, input_data)
 
 
 def check_resource_cleanup(rsrc, template_id, data):
     '''
     Delete the Resource if appropriate.
     '''
-    # Clear out deleted resources from the requirers list
-    rsrc.clear_requirers(rsrc_id for rsrc_id, id in data.items()
-                         if id is None)
 
     if rsrc.current_template_id != template_id:
-        rsrc.delete(template_id, data)
+        rsrc.delete_convergence(template_id, data)
diff --git a/heat/tests/test_engine_worker.py b/heat/tests/test_engine_worker.py
index 1c061feab..511e65f85 100644
--- a/heat/tests/test_engine_worker.py
+++ b/heat/tests/test_engine_worker.py
@@ -277,32 +277,28 @@ class MiscMethodsTest(common.HeatTestCase):
             mock.ANY, {}, True)
         self.assertTrue(mock_sync.called)
 
-    @mock.patch.object(resource.Resource, 'create')
+    @mock.patch.object(resource.Resource, 'create_convergence')
     def test_check_resource_update_create(self, mock_create):
         worker.check_resource_update(self.resource, self.resource.stack.t.id,
                                      {})
         self.assertTrue(mock_create.called)
 
-    @mock.patch.object(resource.Resource, 'update')
+    @mock.patch.object(resource.Resource, 'update_convergence')
     def test_check_resource_update_update(self, mock_update):
         self.resource.resource_id = 'physical-res-id'
         worker.check_resource_update(self.resource, self.resource.stack.t.id,
                                      {})
         self.assertTrue(mock_update.called)
 
-    @mock.patch.object(resource.Resource, 'delete')
-    @mock.patch.object(resource.Resource, 'clear_requirers')
-    def test_check_resource_cleanup_delete(self, mock_cr, mock_delete):
+    @mock.patch.object(resource.Resource, 'delete_convergence')
+    def test_check_resource_cleanup_delete(self, mock_delete):
         self.resource.current_template_id = 'new-template-id'
         worker.check_resource_cleanup(self.resource, self.resource.stack.t.id,
                                       {})
-        self.assertTrue(mock_cr.called)
         self.assertTrue(mock_delete.called)
 
-    @mock.patch.object(resource.Resource, 'delete')
-    @mock.patch.object(resource.Resource, 'clear_requirers')
-    def test_check_resource_cleanup_nodelete(self, mock_cr, mock_delete):
+    @mock.patch.object(resource.Resource, 'delete_convergence')
+    def test_check_resource_cleanup_nodelete(self, mock_delete):
         worker.check_resource_cleanup(self.resource, self.resource.stack.t.id,
                                       {})
-        self.assertTrue(mock_cr.called)
         self.assertFalse(mock_delete.called)
diff --git a/heat/tests/test_resource.py b/heat/tests/test_resource.py
index 71b784a7b..7719279d1 100644
--- a/heat/tests/test_resource.py
+++ b/heat/tests/test_resource.py
@@ -1393,6 +1393,78 @@ class ResourceTest(common.HeatTestCase):
         res_obj = res_objs['test_res_enc']
         self.assertEqual('string', res_obj.properties_data['prop1'])
 
+    @mock.patch.object(resource.Resource, '_store_or_update')
+    @mock.patch.object(resource.Resource, 'create')
+    def test_create_convergence(self,
+                                mock_create,
+                                mock_store_update_method):
+        tmpl = rsrc_defn.ResourceDefinition('test_res', 'Foo')
+        res = generic_rsrc.GenericResource('test_res', tmpl, self.stack)
+        res.create_convergence('template_key', {(1, True): {},
+                                                (1, True): {}})
+
+        mock_create.assert_called_once_with()
+        self.assertEqual('template_key', res.current_template_id)
+        self.assertEqual([1], res.requires)
+        self.assertTrue(mock_store_update_method.called)
+
+    @mock.patch.object(resource.Resource, '_store_or_update')
+    @mock.patch.object(resource.Resource, 'update')
+    def test_update_convergence(self,
+                                mock_update,
+                                mock_store_update_method
+                                ):
+        tmpl = rsrc_defn.ResourceDefinition('test_res', 'Foo')
+        res = generic_rsrc.GenericResource('test_res', tmpl, self.stack)
+        res.requires = [2]
+        res.update_convergence('template_key', {(1, True): {},
+                                                (1, True): {}})
+
+        mock_update.assert_called_once_with(res.t)
+        self.assertEqual('template_key', res.current_template_id)
+        self.assertEqual([1, 2], res.requires)
+        self.assertTrue(mock_store_update_method.called)
+
+    def test_update_in_progress_convergence(self):
+        tmpl = rsrc_defn.ResourceDefinition('test_res', 'Foo')
+        res = generic_rsrc.GenericResource('test_res', tmpl, self.stack)
+        res.status = resource.Resource.IN_PROGRESS
+        ex = self.assertRaises(resource.UpdateInProgress,
+                               res.update_convergence,
+                               'template_key',
+                               {})
+        msg = ("The resource %s is already being updated." %
+               res.name)
+        self.assertEqual(msg, six.text_type(ex))
+
+    @mock.patch.object(resource.Resource, '_store_or_update')
+    @mock.patch.object(resource.Resource, 'delete')
+    def test_delete_convergence(self,
+                                mock_delete,
+                                mock_store_update_method):
+        tmpl = rsrc_defn.ResourceDefinition('test_res', 'Foo')
+        res = generic_rsrc.GenericResource('test_res', tmpl, self.stack)
+        res.requires = [1, 2]
+        res.delete_convergence('template_key', {(1, True): {},
+                                                (1, True): {}})
+
+        mock_delete.assert_called_once_with()
+        self.assertEqual('template_key', res.current_template_id)
+        self.assertEqual([2], res.requires)
+        self.assertTrue(mock_store_update_method.called)
+
+    def test_delete_in_progress_convergence(self):
+        tmpl = rsrc_defn.ResourceDefinition('test_res', 'Foo')
+        res = generic_rsrc.GenericResource('test_res', tmpl, self.stack)
+        res.status = resource.Resource.IN_PROGRESS
+        ex = self.assertRaises(resource.UpdateInProgress,
+                               res.delete_convergence,
+                               'template_key',
+                               {})
+        msg = ("The resource %s is already being updated." %
+               res.name)
+        self.assertEqual(msg, six.text_type(ex))
+
 
 class ResourceAdoptTest(common.HeatTestCase):
     def setUp(self):