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):
@@ -552,25 +565,17 @@ def batch(request, response, storage, *args, **kwargs):
         bucket_name = item.get("bucket_name")
         object_id = item.get("object_id")
         meta = item.get("meta")
-        if method == "PATCH":
-            resp_data = _patch(storage, bucket_name, object_id, meta)
-            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 method == "DELETE":
-            if object_id:
-                resp_data = _delete(storage, bucket_name, object_id)
-            else:
-                try:
-                    storage.delete_bucket(bucket_name)
-                    resp_data = True
-                except (Conflict, NotFound):
-                    pass
-            if resp_data:
-                response.write("HTTP/1.1 204 No Content\r\n")
-                response.write("Content-Type: application/json; charset=UTF-8\r\n")
+
+        handler = f"_batch_{method.lower()}"
+        try:
+            handler = globals()[handler]
+        except KeyError:
+            pass
+        else:
+            resp_data = handler(
+                request, item, storage, bucket_name, object_id, meta, response
+            )
+
         if not resp_data:
             msg = "No such object: {}/{}".format(bucket_name, object_id)
             resp_data = deepcopy(NOT_FOUND)
@@ -582,3 +587,52 @@ def batch(request, response, storage, *args, **kwargs):
             response.write("\r\n\r\n")
 
     response.write("--{}--".format(boundary))
+
+
+def _batch_patch(request, item, storage, bucket_name, object_id, meta, response):
+    resp_data = _patch(storage, bucket_name, object_id, meta)
+    if not resp_data:
+        return None
+    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")
+    return resp_data
+
+
+def _batch_delete(request, item, storage, bucket_name, object_id, meta, response):
+    resp_data = None
+    if object_id:
+        resp_data = _delete(storage, bucket_name, object_id)
+    else:
+        try:
+            storage.delete_bucket(bucket_name)
+            resp_data = True
+        except (Conflict, NotFound):
+            pass
+    if not resp_data:
+        return None
+    response.write("HTTP/1.1 204 No Content\r\n")
+    response.write("Content-Type: application/json; charset=UTF-8\r\n")
+    return resp_data
+
+
+# kludgy heuristics, currently only supports COPY
+def _batch_post(request, item, storage, bucket_name, object_id, meta, response):
+    resp_data = None
+    if object_id:
+        resp_data = _copy(
+            request.base_url,
+            storage,
+            bucket_name,
+            object_id,
+            item["dest_bucket_name"],
+            item["dest_object_id"],
+        )
+    if not resp_data:
+        return None
+    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")
+    return resp_data

src/gcp_storage_emulator/server.py

@@ -106,6 +106,8 @@ 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")