Implement batch support for copy

src/gcp_storage_emulator/handlers/objects.py

@@ -366,36 +366,49 @@ def ls(request, response, storage, *args, **kwargs):
 
 def copy(request, response, storage, *args, **kwargs):
     try:
-        obj = storage.get_file_obj(
-            request.params["bucket_name"], request.params["object_id"]
+        dest_obj = _copy(
+            request.base_url,
+            storage,
+            request.params["bucket_name"],
+            request.params["object_id"],
+            request.params["dest_bucket_name"],
+            request.params["dest_object_id"],
         )
+        if dest_obj is None:
+            response.status = HTTPStatus.NOT_FOUND
+        else:
+            response.json(dest_obj)
+    except Conflict as err:
+        _handle_conflict(response, err)
+
+
+def _copy(base_url, storage, bucket_name, object_id, dest_bucket_name, dest_object_id):
+    try:
+        obj = storage.get_file_obj(bucket_name, object_id)
     except NotFound:
-        response.status = HTTPStatus.NOT_FOUND
-        return
+        return None
 
     dest_obj = _make_object_resource(
-        request.base_url,
-        request.params["dest_bucket_name"],
-        request.params["dest_object_id"],
+        base_url,
+        dest_bucket_name,
+        dest_object_id,
         obj["contentType"],
         obj["size"],
         obj,
     )
 
-    file = storage.get_file(request.params["bucket_name"], request.params["object_id"])
+    file = storage.get_file(bucket_name, object_id)
     try:
         dest_obj = _checksums(file, dest_obj)
         storage.create_file(
-            request.params["dest_bucket_name"],
-            request.params["dest_object_id"],
+            dest_bucket_name,
+            dest_object_id,
             file,
             dest_obj,
         )
-        response.json(dest_obj)
+        return dest_obj
     except NotFound:
-        response.status = HTTPStatus.NOT_FOUND
-    except Conflict as err:
-        _handle_conflict(response, err)
+        return None
 
 
 def rewrite(request, response, storage, *args, **kwargs):
@@ -571,6 +584,22 @@ def batch(request, response, storage, *args, **kwargs):
             if resp_data:
                 response.write("HTTP/1.1 204 No Content\r\n")
                 response.write("Content-Type: application/json; charset=UTF-8\r\n")
+        if method == "POST":  # kludgy heuristics, currently only supports COPY
+            if object_id:
+                resp_data = _copy(
+                    request.base_url,
+                    storage,
+                    bucket_name,
+                    object_id,
+                    item["dest_bucket_name"],
+                    item["dest_object_id"],
+                )
+            if resp_data:
+                response.write("HTTP/1.1 200 OK\r\n")
+                response.write("Content-Type: application/json; charset=UTF-8\r\n")
+                response.write(json.dumps(resp_data))
+                response.write("\r\n\r\n")
+
         if not resp_data:
             msg = "No such object: {}/{}".format(bucket_name, object_id)
             resp_data = deepcopy(NOT_FOUND)

src/gcp_storage_emulator/server.py

@@ -106,6 +106,10 @@ def _health_check(req, res, storage):
     r"^(?P<method>[\w]+).*{}/b/(?P<bucket_name>[-.\w]+)([\?].*)?$".format(
         settings.API_ENDPOINT
     ),
+    r"^(?P<method>[\w]+).*{}/b/(?P<bucket_name>[-.\w]+)/o/(?P<object_id>[^\?]+[^/])/copyTo/b/(?P<dest_bucket_name>[-.\w]+)/o/"
+    r"(?P<dest_object_id>[^\?]+[^/])([\?].*)?$".format(
+        settings.API_ENDPOINT
+    ),
     r"^Content-Type:\s*(?P<content_type>[-.\w/]+)$",
 )
 

tests/test_server.py

@@ -758,6 +758,23 @@ def test_batch_delete_buckets(self):
         with self.assertRaises(NotFound):
             self._client.get_bucket("batchbucket2")
 
+    def test_batch_copy_existing(self):
+        bucket = self._client.create_bucket("bucket_name")
+
+        source = []
+        target = []
+        for i in range(2):
+            source.append(bucket.blob("a/{}.txt".format(i)))
+            source[-1].upload_from_string("text {}".format(i))
+            target.append(bucket.blob("b/{}.txt".format(i)))
+
+        with self._client.batch():
+            for i in range(2):
+                bucket.copy_blob(source[i], bucket, target[i].name)
+
+        blobs = self._client.list_blobs(bucket)
+        self._assert_blob_list(blobs, source + target)
+
     def test_resumable_upload_small_chunk_size(self):
         content = b"a" * 10000000
         bucket = self._client.create_bucket("testbucket")