chore: Wrap botocore ClientError into ServerlessRepoClientError (#25)

Author: paoptu023

serverlessrepo/__version__.py

@@ -1,7 +1,7 @@
 """Serverlessrepo version and package meta-data."""
 
 __title__ = 'serverlessrepo'
-__version__ = '0.1.7'
+__version__ = '0.1.8'
 __license__ = 'Apache 2.0'
 __description__ = (
     'A Python library with convenience helpers for working '

serverlessrepo/exceptions.py

@@ -37,3 +37,15 @@ class S3PermissionsRequired(ServerlessRepoError):
               "permissions to the application artifacts you have uploaded to your S3 bucket. See " \
               "https://docs.aws.amazon.com/serverlessrepo/latest/devguide/serverless-app-publishing-applications.html" \
               " for more details."
+
+
+class InvalidS3UriError(ServerlessRepoError):
+    """Raised when the template contains invalid S3 URIs."""
+
+    MESSAGE = "{message}"
+
+
+class ServerlessRepoClientError(ServerlessRepoError):
+    """Wrapper for botocore ClientError."""
+
+    MESSAGE = "{message}"

serverlessrepo/publish.py

@@ -11,7 +11,7 @@
     yaml_dump, parse_template, get_app_metadata,
     parse_application_id, strip_app_metadata
 )
-from .exceptions import S3PermissionsRequired
+from .exceptions import ServerlessRepoClientError, S3PermissionsRequired, InvalidS3UriError
 
 CREATE_APPLICATION = 'CREATE_APPLICATION'
 UPDATE_APPLICATION = 'UPDATE_APPLICATION'
@@ -47,7 +47,7 @@ def publish_application(template, sar_client=None):
         actions = [CREATE_APPLICATION]
     except ClientError as e:
         if not _is_conflict_exception(e):
-            raise _wrap_s3_exception(e)
+            raise _wrap_client_error(e)
 
         # Update the application if it already exists
         error_message = e.response['Error']['Message']
@@ -57,7 +57,7 @@ def publish_application(template, sar_client=None):
             sar_client.update_application(**request)
             actions = [UPDATE_APPLICATION]
         except ClientError as e:
-            raise _wrap_s3_exception(e)
+            raise _wrap_client_error(e)
 
         # Create application version if semantic version is specified
         if app_metadata.semantic_version:
@@ -67,7 +67,7 @@ def publish_application(template, sar_client=None):
                 actions.append(CREATE_APPLICATION_VERSION)
             except ClientError as e:
                 if not _is_conflict_exception(e):
-                    raise _wrap_s3_exception(e)
+                    raise _wrap_client_error(e)
 
     return {
         'application_id': application_id,
@@ -195,33 +195,36 @@ def _create_application_version_request(app_metadata, application_id, template):
 
 def _is_conflict_exception(e):
     """
-    Check whether the boto3 ClientError is ConflictException.
+    Check whether the botocore ClientError is ConflictException.
 
-    :param e: boto3 exception
+    :param e: botocore exception
     :type e: ClientError
     :return: True if e is ConflictException
     """
     error_code = e.response['Error']['Code']
     return error_code == 'ConflictException'
 
 
-def _wrap_s3_exception(e):
+def _wrap_client_error(e):
     """
-    Wrap S3 access denied exception with a better error message.
+    Wrap botocore ClientError exception into ServerlessRepoClientError.
 
-    :param e: boto3 exception
+    :param e: botocore exception
     :type e: ClientError
-    :return: S3PermissionsRequired if S3 access denied or the original exception
+    :return: S3PermissionsRequired or InvalidS3UriError or general ServerlessRepoClientError
     """
     error_code = e.response['Error']['Code']
     message = e.response['Error']['Message']
 
-    if error_code == 'BadRequestException' and "Failed to copy S3 object. Access denied:" in message:
-        match = re.search('bucket=(.+?), key=(.+?)$', message)
-        if match:
-            return S3PermissionsRequired(bucket=match.group(1), key=match.group(2))
+    if error_code == 'BadRequestException':
+        if "Failed to copy S3 object. Access denied:" in message:
+            match = re.search('bucket=(.+?), key=(.+?)$', message)
+            if match:
+                return S3PermissionsRequired(bucket=match.group(1), key=match.group(2))
+        if "Invalid S3 URI" in message:
+            return InvalidS3UriError(message=message)
 
-    return e
+    return ServerlessRepoClientError(message=message)
 
 
 def _get_publish_details(actions, app_metadata_template):

tests/unit/test_publish.py

@@ -5,7 +5,12 @@
 from botocore.exceptions import ClientError
 
 from serverlessrepo import publish_application, update_application_metadata
-from serverlessrepo.exceptions import InvalidApplicationMetadataError, S3PermissionsRequired
+from serverlessrepo.exceptions import (
+    InvalidApplicationMetadataError,
+    S3PermissionsRequired,
+    InvalidS3UriError,
+    ServerlessRepoClientError
+)
 from serverlessrepo.parser import get_app_metadata, strip_app_metadata, yaml_dump
 from serverlessrepo.publish import (
     CREATE_APPLICATION,
@@ -70,6 +75,15 @@ def setUp(self):
             },
             'create_application'
         )
+        self.invalid_s3_uri_exception = ClientError(
+            {
+                'Error': {
+                    'Code': 'BadRequestException',
+                    'Message': 'Invalid S3 URI'
+                }
+            },
+            'create_application'
+        )
 
     def test_publish_raise_value_error_for_empty_template(self):
         with self.assertRaises(ValueError) as context:
@@ -137,18 +151,18 @@ def test_publish_raise_metadata_error_for_invalid_create_application_request(sel
         # create_application shouldn't be called if application metadata is invalid
         self.serverlessrepo_mock.create_application.assert_not_called()
 
-    def test_publish_raise_client_error_when_create_application(self):
+    def test_publish_raise_serverlessrepo_client_error_when_create_application(self):
         self.serverlessrepo_mock.create_application.side_effect = self.not_conflict_exception
 
         # should raise exception if it's not ConflictException
-        with self.assertRaises(ClientError):
+        with self.assertRaises(ServerlessRepoClientError):
             publish_application(self.template)
 
         # shouldn't call the following APIs if the exception isn't application already exists
         self.serverlessrepo_mock.update_application.assert_not_called()
         self.serverlessrepo_mock.create_application_version.assert_not_called()
 
-    def test_publish_raise_s3_error_when_create_application(self):
+    def test_publish_raise_s3_permission_error_when_create_application(self):
         self.serverlessrepo_mock.create_application.side_effect = self.s3_denied_exception
         with self.assertRaises(S3PermissionsRequired) as context:
             publish_application(self.template)
@@ -157,9 +171,13 @@ def test_publish_raise_s3_error_when_create_application(self):
         self.assertIn("The AWS Serverless Application Repository does not have read access to bucket "
                       "'test-bucket', key 'test-file'.", message)
 
-        # shouldn't call the following APIs if the exception isn't application already exists
-        self.serverlessrepo_mock.update_application.assert_not_called()
-        self.serverlessrepo_mock.create_application_version.assert_not_called()
+    def test_publish_raise_invalid_s3_uri_when_create_application(self):
+        self.serverlessrepo_mock.create_application.side_effect = self.invalid_s3_uri_exception
+        with self.assertRaises(InvalidS3UriError) as context:
+            publish_application(self.template)
+
+        message = str(context.exception)
+        self.assertIn("Invalid S3 URI", message)
 
     def test_publish_existing_application_should_update_application_if_version_not_specified(self):
         self.serverlessrepo_mock.create_application.side_effect = self.application_exists_error
@@ -187,23 +205,21 @@ def test_publish_existing_application_should_update_application_if_version_not_s
         # create_application_version shouldn't be called if version is not provided
         self.serverlessrepo_mock.create_application_version.assert_not_called()
 
-    def test_publish_raise_s3_error_when_update_application(self):
+    @patch('serverlessrepo.publish._wrap_client_error')
+    def test_publish_wrap_client_error_when_update_application(self, wrap_client_error_mock):
         self.serverlessrepo_mock.create_application.side_effect = self.application_exists_error
-        self.serverlessrepo_mock.update_application.side_effect = self.s3_denied_exception
-        with self.assertRaises(S3PermissionsRequired) as context:
+        self.serverlessrepo_mock.update_application.side_effect = self.not_conflict_exception
+        wrap_client_error_mock.return_value = ServerlessRepoClientError(message="client error")
+        with self.assertRaises(ServerlessRepoClientError):
             publish_application(self.template)
 
-        message = str(context.exception)
-        self.assertIn("The AWS Serverless Application Repository does not have read access to bucket "
-                      "'test-bucket', key 'test-file'.", message)
-
         # create_application_version shouldn't be called if update_application fails
         self.serverlessrepo_mock.create_application_version.assert_not_called()
 
     def test_publish_existing_application_should_update_application_if_version_exists(self):
         self.serverlessrepo_mock.create_application.side_effect = self.application_exists_error
         self.serverlessrepo_mock.create_application_version.side_effect = ClientError(
-            {'Error': {'Code': 'ConflictException'}},
+            {'Error': {'Code': 'ConflictException', 'Message': 'Random'}},
             'create_application_version'
         )
 
@@ -257,24 +273,14 @@ def test_publish_new_version_should_create_application_version(self):
         }
         self.serverlessrepo_mock.create_application_version.assert_called_once_with(**expected_request)
 
-    def test_publish_raise_client_error_when_create_application_version(self):
+    @patch('serverlessrepo.publish._wrap_client_error')
+    def test_publish_wrap_client_error_when_create_application_version(self, wrap_client_error_mock):
         self.serverlessrepo_mock.create_application.side_effect = self.application_exists_error
         self.serverlessrepo_mock.create_application_version.side_effect = self.not_conflict_exception
-
-        # should raise exception if it's not ConflictException
-        with self.assertRaises(ClientError):
-            publish_application(self.template)
-
-    def test_publish_raise_s3_error_when_create_application_version(self):
-        self.serverlessrepo_mock.create_application.side_effect = self.application_exists_error
-        self.serverlessrepo_mock.create_application_version.side_effect = self.s3_denied_exception
-        with self.assertRaises(S3PermissionsRequired) as context:
+        wrap_client_error_mock.return_value = ServerlessRepoClientError(message="client error")
+        with self.assertRaises(ServerlessRepoClientError):
             publish_application(self.template)
 
-        message = str(context.exception)
-        self.assertIn("The AWS Serverless Application Repository does not have read access to bucket "
-                      "'test-bucket', key 'test-file'.", message)
-
     def test_create_application_with_passed_in_sar_client(self):
         sar_client = Mock()
         sar_client.create_application.return_value = {