Update kinesis subscription if exists

Author: Jude

lambda_uploader/subscribers.py

@@ -41,23 +41,34 @@ def subscribe(self):
             if self.starting_position_ts:
                 self._lambda_client \
                     .create_event_source_mapping(
-                            EventSourceArn=self.stream_name,
-                            FunctionName=self.function_name,
-                            BatchSize=self.batch_size,
-                            StartingPosition=self.starting_position,
-                            StartingPositionTimestamp=self.starting_position_ts)
+                        EventSourceArn=self.stream_name,
+                        FunctionName=self.function_name,
+                        BatchSize=self.batch_size,
+                        StartingPosition=self.starting_position,
+                        StartingPositionTimestamp=self.starting_position_ts)
             else:
                 self._lambda_client \
                     .create_event_source_mapping(
-                            EventSourceArn=self.stream_name,
-                            FunctionName=self.function_name,
-                            BatchSize=self.batch_size,
-                            StartingPosition=self.starting_position)
+                        EventSourceArn=self.stream_name,
+                        FunctionName=self.function_name,
+                        BatchSize=self.batch_size,
+                        StartingPosition=self.starting_position)
             LOG.debug('Subscription created')
         except botocore.exceptions.ClientError as ex:
             response_code = ex.response['Error']['Code']
             if response_code == 'ResourceConflictException':
-                LOG.debug('Subscription exists')
+                LOG.debug('Subscription exists. Updating ...')
+                resp = self._lambda_client\
+                           .list_event_source_mappings(
+                                FunctionName=self.function_name,
+                                EventSourceArn=self.stream_name)
+                uuid = resp['EventSourceMappings'][0]['UUID']
+                self._lambda_client \
+                    .update_event_source_mapping(
+                        UUID=uuid,
+                        FunctionName=self.function_name,
+                        Enabled=True,
+                        BatchSize=self.batch_size)
             else:
                 LOG.error('Subscription failed, error=%s' % str(ex))
                 raise ex

tests/test_subscribers.py

@@ -1,6 +1,7 @@
 from os import path
 from mock import patch, Mock
 import nose.tools as nt
+import botocore
 
 from lambda_uploader import subscribers, config
 
@@ -21,3 +22,21 @@ def test_successfully_adds_kinesis_subscription(self, mocked_session):
                              config_file=path.join(EX_CONFIG, 'lambda-with-subscription.json'))
         subscribers.create_subscriptions(conf, None)
         nt.assert_equals(True, _mocked_lambda.create_event_source_mapping.called)
+
+    @patch('lambda_uploader.subscribers.boto3.session.Session')
+    def test_successfully_updates_kinesis_subscription(self, mocked_session):
+        resonse = {"Error": {"Code": "ResourceConflictException", "Message": ""}}
+        err = botocore.exceptions.ClientError(resonse, "create_event_source_mapping")
+        _mocked_lambda = Mock()
+        _mocked_lambda.create_event_source_mapping.side_effect = err
+        _mocked_lambda.list_event_source_mappings.return_value = {
+            'EventSourceMappings': [{'UUID': 'myuuid'}]
+        }
+        _mocked_session = Mock()
+        _mocked_session.client = Mock()
+        _mocked_session.client.return_value = _mocked_lambda
+        mocked_session.return_value = _mocked_session
+        conf = config.Config(path.dirname(__file__),
+                             config_file=path.join(EX_CONFIG, 'lambda-with-subscription.json'))
+        subscribers.create_subscriptions(conf, None)
+        nt.assert_equals(True, _mocked_lambda.update_event_source_mapping.called)