Isolate utility inside BookstoreCloneAPIHandler, add tests for utility

bookstore/clone.py

@@ -181,6 +181,23 @@ def initialize(self):
 
         self.session = aiobotocore.get_session()
 
+    def _build_s3_request_object(self, s3_bucket, s3_object_key, s3_version_id=None):
+        """Helper to build object request with the appropriate keys for S3 APIs.
+
+        Parameters
+        ----------
+        s3_bucket: str
+            Log (usually from the NotebookApp) for logging endpoint changes.
+        s3_object_key: str
+            The the path we wish to clone to.
+        s3_version_id: str, optional
+            The version id provided. Default is None, which gets the latest version
+        """
+        s3_kwargs = {"Bucket": s3_bucket, "Key": s3_object_key}
+        if s3_version_id is not None:
+            s3_kwargs['VersionId'] = s3_version_id
+        return s3_kwargs
+
     async def _clone(self, s3_bucket, s3_object_key, s3_version_id=None):
         """Main function that handles communicating with S3 to initiate the clone.
 
@@ -206,9 +223,7 @@ async def _clone(self, s3_bucket, s3_object_key, s3_version_id=None):
         ) as client:
             self.log.info(f"Processing clone of {s3_object_key}")
             try:
-                s3_kwargs = {"Bucket": s3_bucket, "Key": s3_object_key}
-                if s3_version_id is not None:
-                    s3_kwargs['VersionId'] = s3_version_id
+                s3_kwargs = self._build_s3_request_object(s3_bucket, s3_object_key, s3_version_id)
                 obj = await client.get_object(**s3_kwargs)
                 content = (await obj['Body'].read()).decode('utf-8')
             except ClientError as e:

bookstore/tests/test_clone.py

@@ -128,7 +128,9 @@ def test_gen_template_params_s3_version_id(self):
             'redirect_contents_url': '/',
             'source_description': "'my_key' version: my_version from the s3 bucket 'hello'",
         }
-        success_handler = self.get_handler('/bookstore/clone?s3_bucket=hello&s3_key=my_key&s3_version_id=my_version')
+        success_handler = self.get_handler(
+            '/bookstore/clone?s3_bucket=hello&s3_key=my_key&s3_version_id=my_version'
+        )
         output = success_handler.construct_template_params(
             s3_bucket="hello", s3_object_key="my_key", s3_version_id="my_version"
         )
@@ -225,6 +227,33 @@ async def test_private_clone_nonsense_params(self):
         with pytest.raises(HTTPError):
             await success_handler._clone(s3_bucket, s3_object_key)
 
+    def test_build_s3_request_object(self):
+        expected = {"Bucket": "my_bucket", "Key": "my_key"}
+        s3_bucket = "my_bucket"
+        s3_object_key = "my_key"
+        post_body_dict = {"s3_key": "my_key", "s3_bucket": "my_bucket"}
+        success_handler = self.post_handler(post_body_dict)
+        actual = success_handler._build_s3_request_object(s3_bucket, s3_object_key)
+        assert actual == expected
+
+    def test_build_s3_request_object_version_id(self):
+        expected = {"Bucket": "my_bucket", "Key": "my_key", "VersionId": "my_version"}
+
+        s3_bucket = "my_bucket"
+        s3_object_key = "my_key"
+        s3_version_id = "my_version"
+
+        post_body_dict = {
+            "s3_key": s3_object_key,
+            "s3_bucket": s3_bucket,
+            "s3_version_id": s3_version_id,
+        }
+        success_handler = self.post_handler(post_body_dict)
+        actual = success_handler._build_s3_request_object(
+            s3_bucket, s3_object_key, s3_version_id=s3_version_id
+        )
+        assert actual == expected
+
     def test_build_post_response_model(self):
         content = "some arbitrary content"
         expected = {