Consolidate bucket_exists logic into utils file

Added a helper function that can be used to determine if a s3 bucket
exists or not.

awscli/customizations/cloudtrail.py

@@ -15,6 +15,7 @@
 import sys
 
 from awscli.customizations.commands import BasicCommand
+from awscli.customizations.utils import s3_bucket_exists
 from botocore.exceptions import ClientError
 
 
@@ -232,13 +233,8 @@ def setup_new_bucket(self, bucket, prefix, custom_policy=None):
             policy = policy.replace('<Prefix>', prefix or '')
 
         LOG.debug('Bucket policy:\n{0}'.format(policy))
-
-        try:
-            self.s3.head_bucket(Bucket=bucket)
-        except ClientError:
-            # The bucket does not exists.  This is what we want.
-            pass
-        else:
+        bucket_exists = s3_bucket_exists(self.s3, bucket)
+        if bucket_exists:
             raise Exception('Bucket {bucket} already exists.'.format(
                 bucket=bucket))
 

awscli/customizations/configservice/subscribe.py

@@ -13,9 +13,8 @@
 import json
 import sys
 
-from botocore.exceptions import ClientError
-
 from awscli.customizations.commands import BasicCommand
+from awscli.customizations.utils import s3_bucket_exists
 from awscli.customizations.s3.utils import find_bucket_key
 
 
@@ -146,20 +145,10 @@ def prepare_bucket(self, s3_path):
         return bucket, key
 
     def _check_bucket_exists(self, bucket):
-        bucket_exists = True
         self._s3_client.meta.events.unregister(
             'after-call',
             unique_id='awscli-error-handler')
-        try:
-            # See if the bucket exists by running a head bucket
-            self._s3_client.head_bucket(Bucket=bucket)
-        except ClientError as e:
-            # If a client error is thrown. Check that it was a 404 error.
-            # If it was a 404 error, than the bucket does not exist.
-            error_code = int(e.response['Error']['Code'])
-            if error_code == 404:
-                bucket_exists = False
-        return bucket_exists
+        return s3_bucket_exists(self._s3_client, bucket)
 
     def _create_bucket(self, bucket):
         region_name = self._s3_client._endpoint.region_name

awscli/customizations/utils.py

@@ -14,6 +14,10 @@
 Utility functions to make it easier to work with customizations.
 
 """
+
+from botocore.exceptions import ClientError
+
+
 def rename_argument(argument_table, existing_name, new_name):
     current = argument_table[existing_name]
     argument_table[new_name] = current
@@ -60,3 +64,17 @@ def _get_group_for_key(key, groups):
     for group in groups:
         if key in group:
             return group
+
+
+def s3_bucket_exists(s3_client, bucket_name):
+    bucket_exists = True
+    try:
+        # See if the bucket exists by running a head bucket
+        s3_client.head_bucket(Bucket=bucket_name)
+    except ClientError as e:
+        # If a client error is thrown. Check that it was a 404 error.
+        # If it was a 404 error, than the bucket does not exist.
+        error_code = int(e.response['Error']['Code'])
+        if error_code == 404:
+            bucket_exists = False
+    return bucket_exists

tests/unit/customizations/test_utils.py

@@ -14,6 +14,7 @@
 from awscli.testutils import BaseAWSHelpOutputTest
 
 import mock
+from botocore.exceptions import ClientError
 
 from awscli.customizations import utils
 
@@ -63,3 +64,36 @@ def test_multiple_groups(self):
         parsed = FakeParsedArgs(foo='one', bar=None, qux='three')
         with self.assertRaises(ValueError):
             utils.validate_mutually_exclusive(parsed, *groups)
+
+
+class TestS3BucketExists(unittest.TestCase):
+    def setUp(self):
+        self.s3_client = mock.Mock()
+        self.bucket_name = 'mybucket'
+        self.error_response = {
+            'Error': {
+                'Code': '404',
+                'Message': 'Not Found'
+            }
+        }
+        self.bucket_no_exists_error = ClientError(
+            self.error_response,
+            'HeadBucket'
+        )
+
+    def test_bucket_exists(self):
+        self.assertTrue(
+            utils.s3_bucket_exists(self.s3_client, self.bucket_name))
+
+    def test_bucket_not_exists(self):
+        self.s3_client.head_bucket.side_effect = self.bucket_no_exists_error
+        self.assertFalse(
+            utils.s3_bucket_exists(self.s3_client, self.bucket_name))
+
+    def test_bucket_exists_with_non_404(self):
+        self.error_response['Error']['Code'] = '403'
+        self.error_response['Error']['Message'] = 'Forbidden'
+        forbidden_error = ClientError(self.error_response, 'HeadBucket')
+        self.s3_client.head_bucket.side_effect = forbidden_error
+        self.assertTrue(
+            utils.s3_bucket_exists(self.s3_client, self.bucket_name))