Updating member flow

Adding ability to plug networks when creating member.

Change-Id: I48fcb64839f46e4ce0754cd9b25b47cd13b7fd75

octavia/common/constants.py

@@ -79,6 +79,7 @@ SUPPORTED_AMPHORA_TYPES = (AMPHORA_VM,)
 
 # Task/Flow constants
 AMPHORA = 'amphora'
+AMPHORAE = 'amphorae'
 AMPHORA_ID = 'amphora_id'
 DELTA = 'delta'
 DELTAS = 'deltas'

octavia/controller/worker/flows/member_flows.py

@@ -19,6 +19,7 @@ from octavia.common import constants
 from octavia.controller.worker.tasks import amphora_driver_tasks
 from octavia.controller.worker.tasks import database_tasks
 from octavia.controller.worker.tasks import model_tasks
+from octavia.controller.worker.tasks import network_tasks
 
 
 class MemberFlows(object):
@@ -29,11 +30,20 @@ class MemberFlows(object):
         :returns: The flow for creating a member
         """
         create_member_flow = linear_flow.Flow(constants.CREATE_MEMBER_FLOW)
+        create_member_flow.add(network_tasks.CalculateDelta(
+            requires=constants.LOADBALANCER,
+            provides=constants.DELTAS))
+        create_member_flow.add(network_tasks.HandleNetworkDeltas(
+            requires=constants.DELTAS))
+        create_member_flow.add(amphora_driver_tasks.AmphoraePostNetworkPlug(
+            requires=constants.LOADBALANCER
+        ))
         create_member_flow.add(amphora_driver_tasks.ListenerUpdate(
-            requires=['listener', 'vip']))
+            requires=(constants.LISTENER, constants.VIP)))
         create_member_flow.add(database_tasks.
                                MarkLBAndListenerActiveInDB(
-                                   requires=['loadbalancer', 'listener']))
+                                   requires=(constants.LOADBALANCER,
+                                             constants.LISTENER)))
 
         return create_member_flow
 

octavia/controller/worker/tasks/amphora_driver_tasks.py

@@ -163,6 +163,25 @@ class AmphoraPostNetworkPlug(BaseAmphoraTask):
                                  status=constants.ERROR)
 
 
+class AmphoraePostNetworkPlug(BaseAmphoraTask):
+    """Task to notify the amphorae post network plug."""
+
+    def execute(self, loadbalancer):
+        """Execute post_network_plug routine."""
+        for amphora in loadbalancer.amphorae:
+            self.amphora_driver.post_network_plug(amphora)
+            LOG.debug("Posted network plug for the compute instance")
+
+    def revert(self, result, loadbalancer, *args, **kwargs):
+        """Handle a failed post network plug."""
+        if isinstance(result, failure.Failure):
+            return
+        LOG.warn(_LW("Reverting post network plug."))
+        for amphora in loadbalancer.amphorae:
+            self.amphora_repo.update(db_apis.get_session(), id=amphora.id,
+                                     status=constants.ERROR)
+
+
 class AmphoraPostVIPPlug(BaseAmphoraTask):
     """Task to notify the amphora post VIP plug."""
 

octavia/tests/unit/controller/worker/flows/test_member_flows.py

@@ -37,7 +37,7 @@ class TestMemberFlows(base.TestCase):
         self.assertIn('vip', member_flow.requires)
 
         self.assertEqual(len(member_flow.requires), 3)
-        self.assertEqual(len(member_flow.provides), 0)
+        self.assertEqual(len(member_flow.provides), 1)
 
     def test_get_delete_member_flow(self):
 

octavia/tests/unit/controller/worker/tasks/test_amphora_driver_tasks.py

@@ -31,6 +31,7 @@ _listener_mock = mock.MagicMock()
 _listener_mock.id = LISTENER_ID
 _vip_mock = mock.MagicMock()
 _LB_mock = mock.MagicMock()
+_amphorae_mock = [_amphora_mock]
 
 
 @mock.patch('octavia.db.repositories.AmphoraRepository.update')
@@ -210,6 +211,30 @@ class TestDatabaseTasks(base.TestCase):
 
         self.assertIsNone(amp)
 
+    def test_amphorae_post_network_plug(self, mock_driver,
+                                        mock_generate_uuid,
+                                        mock_log,
+                                        mock_get_session,
+                                        mock_listener_repo_update,
+                                        mock_amphora_repo_update):
+
+        _LB_mock.amphorae = [_amphora_mock]
+        amphora_post_network_plug_obj = (amphora_driver_tasks.
+                                         AmphoraePostNetworkPlug())
+        amphora_post_network_plug_obj.execute(_LB_mock)
+
+        (mock_driver.post_network_plug.
+            assert_called_once_with)(_amphora_mock)
+
+        # Test revert
+        amp = amphora_post_network_plug_obj.revert(None, _LB_mock)
+        repo.AmphoraRepository.update.assert_called_once_with(
+            'TEST',
+            id=AMP_ID,
+            status=constants.ERROR)
+
+        self.assertIsNone(amp)
+
     @mock.patch('octavia.db.repositories.LoadBalancerRepository.update')
     def test_amphora_post_vip_plug(self,
                                    mock_loadbalancer_repo_update,