From 9d0a3abba27ce1bc5fe03d7bb332f206695e460c Mon Sep 17 00:00:00 2001
From: Cathy Ouyang <cathyo@google.com>
Date: Thu, 15 Jun 2023 20:46:44 -0700
Subject: [PATCH 01/14] fix: add support to resumable uploads

---
 testbench/common.py      |  5 +++++
 testbench/rest_server.py | 34 ++++++++++++++++++++++++++++++++++
 2 files changed, 39 insertions(+)

diff --git a/testbench/common.py b/testbench/common.py
index 7a7363a4..23e49551 100644
--- a/testbench/common.py
+++ b/testbench/common.py
@@ -611,6 +611,11 @@ def corrupt_media(media):
     return b"B" + media[1:] if media[0:1] == b"A" else b"A" + media[1:]
 
 
+def interrupt_media(media, num_bytes):
+    """ Returns partial media due to forced interruption. """
+    return media[:num_bytes]
+
+
 # === HEADERS === #
 
 
diff --git a/testbench/rest_server.py b/testbench/rest_server.py
index ae7315ba..877fd676 100644
--- a/testbench/rest_server.py
+++ b/testbench/rest_server.py
@@ -948,8 +948,14 @@ def resumable_upload_chunk(bucket_name):
     if upload_id is None:
         testbench.error.missing("upload_id in resumable_upload_chunk", None)
     upload = db.get_upload(upload_id, None)
+
     if upload.complete:
         return gcs_type.object.Object.rest(upload.metadata)
+    # Return status for incomplete resumable upload queries
+    if len(request.data) == 0:
+        override_308 = request.headers.get("X-Guploader-No-308") == "yes"
+        return upload.resumable_status_rest(override_308=override_308)
+
     upload.transfer.add(request.environ.get("HTTP_TRANSFER_ENCODING", ""))
     content_length = request.headers.get("content-length", None)
     data = testbench.common.extract_media(request)
@@ -959,6 +965,34 @@ def resumable_upload_chunk(bucket_name):
         testbench.error.invalid("content-length header", None)
     content_range = request.headers.get("content-range")
     custom_header_value = request.headers.get("x-goog-emulator-custom-header")
+
+    # Handle instructions here
+    instruction = testbench.common.extract_instruction(request, context=None)
+    if instruction == "return-503-after-256K":
+        data = testbench.common.interrupt_media(data, 262144)
+        upload.media += data
+        upload.complete = False
+        blob, _ = gcs_type.object.Object.init(
+            upload.request,
+            upload.metadata,
+            upload.media,
+            upload.bucket,
+            False,
+            None,
+        )
+        blob.metadata.metadata["x_emulator_transfer_encoding"] = ":".join(
+            upload.transfer
+        )
+        db.insert_object(
+            bucket_name,
+            blob,
+            context=None,
+            preconditions=testbench.common.make_json_preconditions(
+                upload.request
+            ),
+        )
+        return flask.Response("Service Unavailable", status=503)
+
     if content_range is not None:
         items = list(testbench.common.content_range_split.match(content_range).groups())
         # TODO(#27) - maybe this should be an assert()

From e66c785127757e90abd1fcda84506324bf4e60b7 Mon Sep 17 00:00:00 2001
From: Cathy Ouyang <cathyo@google.com>
Date: Thu, 15 Jun 2023 21:02:04 -0700
Subject: [PATCH 02/14] dedup

---
 testbench/rest_server.py | 6 ------
 1 file changed, 6 deletions(-)

diff --git a/testbench/rest_server.py b/testbench/rest_server.py
index 877fd676..989fc101 100644
--- a/testbench/rest_server.py
+++ b/testbench/rest_server.py
@@ -948,14 +948,8 @@ def resumable_upload_chunk(bucket_name):
     if upload_id is None:
         testbench.error.missing("upload_id in resumable_upload_chunk", None)
     upload = db.get_upload(upload_id, None)
-
     if upload.complete:
         return gcs_type.object.Object.rest(upload.metadata)
-    # Return status for incomplete resumable upload queries
-    if len(request.data) == 0:
-        override_308 = request.headers.get("X-Guploader-No-308") == "yes"
-        return upload.resumable_status_rest(override_308=override_308)
-
     upload.transfer.add(request.environ.get("HTTP_TRANSFER_ENCODING", ""))
     content_length = request.headers.get("content-length", None)
     data = testbench.common.extract_media(request)

From ccb66fdaabff0c98f71bc4090f0d5511ccfd9560 Mon Sep 17 00:00:00 2001
From: Cathy Ouyang <cathyo@google.com>
Date: Fri, 16 Jun 2023 16:21:09 -0700
Subject: [PATCH 03/14] update upload only; no need to init blob yet

---
 testbench/rest_server.py | 21 +--------------------
 1 file changed, 1 insertion(+), 20 deletions(-)

diff --git a/testbench/rest_server.py b/testbench/rest_server.py
index 989fc101..7f5bd169 100644
--- a/testbench/rest_server.py
+++ b/testbench/rest_server.py
@@ -960,31 +960,12 @@ def resumable_upload_chunk(bucket_name):
     content_range = request.headers.get("content-range")
     custom_header_value = request.headers.get("x-goog-emulator-custom-header")
 
-    # Handle instructions here
+    # Handle "return-503-after-256K" instructions for uploads
     instruction = testbench.common.extract_instruction(request, context=None)
     if instruction == "return-503-after-256K":
         data = testbench.common.interrupt_media(data, 262144)
         upload.media += data
         upload.complete = False
-        blob, _ = gcs_type.object.Object.init(
-            upload.request,
-            upload.metadata,
-            upload.media,
-            upload.bucket,
-            False,
-            None,
-        )
-        blob.metadata.metadata["x_emulator_transfer_encoding"] = ":".join(
-            upload.transfer
-        )
-        db.insert_object(
-            bucket_name,
-            blob,
-            context=None,
-            preconditions=testbench.common.make_json_preconditions(
-                upload.request
-            ),
-        )
         return flask.Response("Service Unavailable", status=503)
 
     if content_range is not None:

From 6eed30ced537594655a9a74b34e5a809abe593fd Mon Sep 17 00:00:00 2001
From: Cathy Ouyang <cathyo@google.com>
Date: Fri, 23 Jun 2023 10:18:31 -0700
Subject: [PATCH 04/14] uploads should ignore request bytes already persisted

---
 testbench/common.py      |  6 +++---
 testbench/rest_server.py | 25 ++++++++++++++++++++-----
 2 files changed, 23 insertions(+), 8 deletions(-)

diff --git a/testbench/common.py b/testbench/common.py
index 23e49551..c3862597 100644
--- a/testbench/common.py
+++ b/testbench/common.py
@@ -611,9 +611,9 @@ def corrupt_media(media):
     return b"B" + media[1:] if media[0:1] == b"A" else b"A" + media[1:]
 
 
-def interrupt_media(media, num_bytes):
-    """ Returns partial media due to forced interruption. """
-    return media[:num_bytes]
+def partial_media(media, range_end, range_start=0):
+    """ Returns partial media due to forced interruption or server validation. """
+    return media[range_start:range_end]
 
 
 # === HEADERS === #
diff --git a/testbench/rest_server.py b/testbench/rest_server.py
index 7f5bd169..7eb7d121 100644
--- a/testbench/rest_server.py
+++ b/testbench/rest_server.py
@@ -950,6 +950,9 @@ def resumable_upload_chunk(bucket_name):
     upload = db.get_upload(upload_id, None)
     if upload.complete:
         return gcs_type.object.Object.rest(upload.metadata)
+    # The server should ignore any request bytes that have already been persisted.
+    # Thus we check the last_byte_persisted in the upload.
+    last_byte_persisted = 0 if len(upload.media) == 0 else (len(upload.media) - 1)
     upload.transfer.add(request.environ.get("HTTP_TRANSFER_ENCODING", ""))
     content_length = request.headers.get("content-length", None)
     data = testbench.common.extract_media(request)
@@ -963,10 +966,12 @@ def resumable_upload_chunk(bucket_name):
     # Handle "return-503-after-256K" instructions for uploads
     instruction = testbench.common.extract_instruction(request, context=None)
     if instruction == "return-503-after-256K":
-        data = testbench.common.interrupt_media(data, 262144)
-        upload.media += data
-        upload.complete = False
-        return flask.Response("Service Unavailable", status=503)
+        if last_byte_persisted < 262144:
+            data = testbench.common.partial_media(data, range_end=262144, range_start=last_byte_persisted)
+            upload.media += data
+            upload.complete = False
+        if len(upload.media) >= 262144:
+            return flask.Response("Service Unavailable", status=503)
 
     if content_range is not None:
         items = list(testbench.common.content_range_split.match(content_range).groups())
@@ -1009,7 +1014,8 @@ def resumable_upload_chunk(bucket_name):
                     blob.rest_metadata(), projection, fields
                 )
             return upload.resumable_status_rest()
-        _, chunk_last_byte = [v for v in items[0].split("-")]
+        # In addition to chunk_last_byte, we also need to inspect chunk_first_byte.
+        chunk_first_byte, chunk_last_byte = [v for v in items[0].split("-")]
         x_upload_content_length = int(
             upload.request.headers.get("x-upload-content-length", 0)
         )
@@ -1032,6 +1038,15 @@ def resumable_upload_chunk(bucket_name):
                 None,
                 rest_code=400,
             )
+        # Validate chunk_first_byte to ignore any request bytes that have already been persisted.
+        if chunk_first_byte == "*":
+            # TODO what is the use case and how to handle
+            pass
+        elif int(chunk_first_byte) <= last_byte_persisted:
+            # Ignore request bytes that have already been persisted.
+            chunk_first_byte = last_byte_persisted + 1
+            data = testbench.common.partial_media(data, range_end=(chunk_last_byte + 1), range_start=chunk_first_byte)
+
         upload.media += data
         upload.complete = total_object_size == len(upload.media) or (
             chunk_last_byte + 1 == total_object_size

From b972177f13ce7ef84ebb812b7956add725e483e9 Mon Sep 17 00:00:00 2001
From: Cathy Ouyang <cathyo@google.com>
Date: Fri, 23 Jun 2023 11:59:08 -0700
Subject: [PATCH 05/14] wip update retry test after bytes in resum uploads

---
 testbench/common.py      | 17 +++++++++++++++++
 testbench/rest_server.py | 38 +++++++++++++++++++++++++++++++-------
 2 files changed, 48 insertions(+), 7 deletions(-)

diff --git a/testbench/common.py b/testbench/common.py
index c3862597..db2137d6 100644
--- a/testbench/common.py
+++ b/testbench/common.py
@@ -841,6 +841,23 @@ def wrapper(*args, **kwargs):
     return retry_test
 
 
+def handle_retry_resumable_uploads_error_after_bytes(upload, data, database, request):
+    test_id = request.headers.get("x-retry-test-id", None)
+    if not test_id:
+        return 0, 0, ""
+    next_instruction = database.peek_next_instruction(test_id, "storage.objects.insert")
+    if not next_instruction:
+        return 0, 0, ""
+    error_after_bytes_matches = testbench.common.retry_return_error_after_bytes.match(next_instruction)
+    if error_after_bytes_matches:
+        # database.dequeue_next_instruction(test_id, "storage.objects.insert")
+        items = list(error_after_bytes_matches.groups())
+        error_code = int(items[0])
+        after_bytes = int(items[1]) * 1024
+        return error_code, after_bytes, test_id
+    return 0, 0, ""
+
+
 def handle_gzip_request(request):
     """
     Handle gzip compressed JSON payloads when Content-Encoding: gzip is present on metadata requests.
diff --git a/testbench/rest_server.py b/testbench/rest_server.py
index 7eb7d121..2f81c954 100644
--- a/testbench/rest_server.py
+++ b/testbench/rest_server.py
@@ -963,15 +963,37 @@ def resumable_upload_chunk(bucket_name):
     content_range = request.headers.get("content-range")
     custom_header_value = request.headers.get("x-goog-emulator-custom-header")
 
+    ### TEMP ###
+    # In addition to chunk_last_byte, we also need to inspect chunk_first_byte.
+    items = list(testbench.common.content_range_split.match(content_range).groups())
+    chunk_first_byte, chunk_last_byte = [v for v in items[0].split("-")]
+    ### TEMP ###
+    ### Probably move this entire logic to a helper method?!
     # Handle "return-503-after-256K" instructions for uploads
     instruction = testbench.common.extract_instruction(request, context=None)
-    if instruction == "return-503-after-256K":
-        if last_byte_persisted < 262144:
-            data = testbench.common.partial_media(data, range_end=262144, range_start=last_byte_persisted)
+    error_code, after_bytes, test_id = testbench.common.handle_retry_resumable_uploads_error_after_bytes(upload, data, db, request)
+    if instruction or error_code:
+        if instruction == "return-503-after-256K":
+            error_code = 503
+            after_bytes = 262144
+        if last_byte_persisted < after_bytes:
+            range_start = 0
+            # Ignore request bytes that have already been persisted.
+            if int(chunk_first_byte) < last_byte_persisted:
+                range_start = last_byte_persisted - int(chunk_first_byte) + 1
+            elif int(chunk_first_byte) == last_byte_persisted and last_byte_persisted != 0:
+                range_start = int(chunk_first_byte) + 1
+            data = testbench.common.partial_media(data, range_end=after_bytes, range_start=range_start)
             upload.media += data
             upload.complete = False
-        if len(upload.media) >= 262144:
-            return flask.Response("Service Unavailable", status=503)
+        if len(upload.media) >= after_bytes:
+            db.dequeue_next_instruction(test_id, "storage.objects.insert")
+            testbench.error.generic(
+                "Fault injected during a resumable upload",
+                rest_code=error_code,
+                grpc_code=None,
+                context=None,
+            )
 
     if content_range is not None:
         items = list(testbench.common.content_range_split.match(content_range).groups())
@@ -981,6 +1003,8 @@ def resumable_upload_chunk(bucket_name):
         #   assert((items[0] != items[1]) or items[0] == '*')
         if len(items) != 2 or (items[0] == items[1] and items[0] != "*"):
             testbench.error.invalid("content-range header", None)
+        # In addition to chunk_last_byte, we also need to inspect chunk_first_byte.
+        chunk_first_byte, chunk_last_byte = [v for v in items[0].split("-")]
         # TODO(#27) - maybe this should be an assert()
         # We check if the upload is complete before we get here.
         #   assert(not upload.completed)
@@ -1014,8 +1038,8 @@ def resumable_upload_chunk(bucket_name):
                     blob.rest_metadata(), projection, fields
                 )
             return upload.resumable_status_rest()
-        # In addition to chunk_last_byte, we also need to inspect chunk_first_byte.
-        chunk_first_byte, chunk_last_byte = [v for v in items[0].split("-")]
+        # # In addition to chunk_last_byte, we also need to inspect chunk_first_byte.
+        # chunk_first_byte, chunk_last_byte = [v for v in items[0].split("-")]
         x_upload_content_length = int(
             upload.request.headers.get("x-upload-content-length", 0)
         )

From 6f652edd4a6b71111ebb5faef0a7cbb657eaf7b2 Mon Sep 17 00:00:00 2001
From: Cathy Ouyang <cathyo@google.com>
Date: Fri, 23 Jun 2023 14:34:42 -0700
Subject: [PATCH 06/14] cleanup logic around ignoring duplicate request bytes

---
 testbench/rest_server.py | 79 ++++++++++++++++++----------------------
 1 file changed, 36 insertions(+), 43 deletions(-)

diff --git a/testbench/rest_server.py b/testbench/rest_server.py
index 2f81c954..a84c278b 100644
--- a/testbench/rest_server.py
+++ b/testbench/rest_server.py
@@ -963,38 +963,6 @@ def resumable_upload_chunk(bucket_name):
     content_range = request.headers.get("content-range")
     custom_header_value = request.headers.get("x-goog-emulator-custom-header")
 
-    ### TEMP ###
-    # In addition to chunk_last_byte, we also need to inspect chunk_first_byte.
-    items = list(testbench.common.content_range_split.match(content_range).groups())
-    chunk_first_byte, chunk_last_byte = [v for v in items[0].split("-")]
-    ### TEMP ###
-    ### Probably move this entire logic to a helper method?!
-    # Handle "return-503-after-256K" instructions for uploads
-    instruction = testbench.common.extract_instruction(request, context=None)
-    error_code, after_bytes, test_id = testbench.common.handle_retry_resumable_uploads_error_after_bytes(upload, data, db, request)
-    if instruction or error_code:
-        if instruction == "return-503-after-256K":
-            error_code = 503
-            after_bytes = 262144
-        if last_byte_persisted < after_bytes:
-            range_start = 0
-            # Ignore request bytes that have already been persisted.
-            if int(chunk_first_byte) < last_byte_persisted:
-                range_start = last_byte_persisted - int(chunk_first_byte) + 1
-            elif int(chunk_first_byte) == last_byte_persisted and last_byte_persisted != 0:
-                range_start = int(chunk_first_byte) + 1
-            data = testbench.common.partial_media(data, range_end=after_bytes, range_start=range_start)
-            upload.media += data
-            upload.complete = False
-        if len(upload.media) >= after_bytes:
-            db.dequeue_next_instruction(test_id, "storage.objects.insert")
-            testbench.error.generic(
-                "Fault injected during a resumable upload",
-                rest_code=error_code,
-                grpc_code=None,
-                context=None,
-            )
-
     if content_range is not None:
         items = list(testbench.common.content_range_split.match(content_range).groups())
         # TODO(#27) - maybe this should be an assert()
@@ -1003,8 +971,6 @@ def resumable_upload_chunk(bucket_name):
         #   assert((items[0] != items[1]) or items[0] == '*')
         if len(items) != 2 or (items[0] == items[1] and items[0] != "*"):
             testbench.error.invalid("content-range header", None)
-        # In addition to chunk_last_byte, we also need to inspect chunk_first_byte.
-        chunk_first_byte, chunk_last_byte = [v for v in items[0].split("-")]
         # TODO(#27) - maybe this should be an assert()
         # We check if the upload is complete before we get here.
         #   assert(not upload.completed)
@@ -1038,8 +1004,8 @@ def resumable_upload_chunk(bucket_name):
                     blob.rest_metadata(), projection, fields
                 )
             return upload.resumable_status_rest()
-        # # In addition to chunk_last_byte, we also need to inspect chunk_first_byte.
-        # chunk_first_byte, chunk_last_byte = [v for v in items[0].split("-")]
+        # In addition to chunk_last_byte, we also need to inspect chunk_first_byte.
+        chunk_first_byte, chunk_last_byte = [v for v in items[0].split("-")]
         x_upload_content_length = int(
             upload.request.headers.get("x-upload-content-length", 0)
         )
@@ -1062,14 +1028,41 @@ def resumable_upload_chunk(bucket_name):
                 None,
                 rest_code=400,
             )
+        ### Handle "return-503-after-256K" instructions for uploads
+        ### Probably move this entire logic to a helper method?!
+        instruction = testbench.common.extract_instruction(request, context=None)
+        error_code, after_bytes, test_id = testbench.common.handle_retry_resumable_uploads_error_after_bytes(upload, data, db, request)
+        if instruction or error_code:
+            if instruction == "return-503-after-256K":
+                error_code = 503
+                after_bytes = 262144
+            if last_byte_persisted < after_bytes:
+                range_start = 0
+                # Ignore request bytes that have already been persisted.
+                if int(chunk_first_byte) < last_byte_persisted:
+                    range_start = last_byte_persisted - int(chunk_first_byte) + 1
+                elif int(chunk_first_byte) == last_byte_persisted and last_byte_persisted != 0:
+                    range_start = int(chunk_first_byte) + 1
+                data = testbench.common.partial_media(data, range_end=after_bytes, range_start=range_start)
+                upload.media += data
+                upload.complete = False
+            if len(upload.media) >= after_bytes:
+                if test_id:
+                    db.dequeue_next_instruction(test_id, "storage.objects.insert")
+                testbench.error.generic(
+                    "Fault injected during a resumable upload",
+                    rest_code=error_code,
+                    grpc_code=None,
+                    context=None,
+                )
         # Validate chunk_first_byte to ignore any request bytes that have already been persisted.
-        if chunk_first_byte == "*":
-            # TODO what is the use case and how to handle
-            pass
-        elif int(chunk_first_byte) <= last_byte_persisted:
-            # Ignore request bytes that have already been persisted.
-            chunk_first_byte = last_byte_persisted + 1
-            data = testbench.common.partial_media(data, range_end=(chunk_last_byte + 1), range_start=chunk_first_byte)
+        range_start = 0
+        if chunk_first_byte != "*":
+            if int(chunk_first_byte) < last_byte_persisted:
+                range_start = last_byte_persisted - int(chunk_first_byte) + 1
+            elif int(chunk_first_byte) == last_byte_persisted and last_byte_persisted != 0:
+                range_start = int(chunk_first_byte) + 1
+        data = testbench.common.partial_media(data, range_end=(chunk_last_byte + 1), range_start=range_start)
 
         upload.media += data
         upload.complete = total_object_size == len(upload.media) or (

From 69d536758c00857a12f5b4e0e0e2cfee1b0a50c4 Mon Sep 17 00:00:00 2001
From: Cathy Ouyang <cathyo@google.com>
Date: Fri, 23 Jun 2023 15:13:41 -0700
Subject: [PATCH 07/14] add helper method and update comments

---
 testbench/common.py      | 48 ++++++++++++++++++++++------------------
 testbench/rest_server.py | 31 +++++---------------------
 2 files changed, 33 insertions(+), 46 deletions(-)

diff --git a/testbench/common.py b/testbench/common.py
index db2137d6..c34a156a 100644
--- a/testbench/common.py
+++ b/testbench/common.py
@@ -789,25 +789,6 @@ def handle_retry_test_instruction(database, request, socket_closer, method):
         return __get_streamer_response_fn(
             database, method, socket_closer, test_id, limit=after_bytes
         )
-    error_after_bytes_matches = testbench.common.retry_return_error_after_bytes.match(
-        next_instruction
-    )
-    if error_after_bytes_matches and method == "storage.objects.insert":
-        items = list(error_after_bytes_matches.groups())
-        error_code = int(items[0])
-        after_bytes = int(items[1]) * 1024
-        # Upload failures should allow to not complete after certain bytes
-        upload_id = request.args.get("upload_id", None)
-        if upload_id is not None:
-            upload = database.get_upload(upload_id, None)
-            if upload is not None and len(upload.media) >= after_bytes:
-                database.dequeue_next_instruction(test_id, method)
-                testbench.error.generic(
-                    "Fault injected after uploading %d bytes" % len(upload.media),
-                    rest_code=error_code,
-                    grpc_code=StatusCode.INTERNAL,  # not really used
-                    context=None,
-                )
     retry_return_short_response = testbench.common.retry_return_short_response.match(
         next_instruction
     )
@@ -841,7 +822,8 @@ def wrapper(*args, **kwargs):
     return retry_test
 
 
-def handle_retry_resumable_uploads_error_after_bytes(upload, data, database, request):
+def get_retry_uploads_error_after_bytes(database, request):
+    """Retrieve error code and #bytes corresponding to uploads from retry test instructions."""
     test_id = request.headers.get("x-retry-test-id", None)
     if not test_id:
         return 0, 0, ""
@@ -850,13 +832,37 @@ def handle_retry_resumable_uploads_error_after_bytes(upload, data, database, req
         return 0, 0, ""
     error_after_bytes_matches = testbench.common.retry_return_error_after_bytes.match(next_instruction)
     if error_after_bytes_matches:
-        # database.dequeue_next_instruction(test_id, "storage.objects.insert")
         items = list(error_after_bytes_matches.groups())
         error_code = int(items[0])
         after_bytes = int(items[1]) * 1024
         return error_code, after_bytes, test_id
     return 0, 0, ""
 
+def handle_retry_uploads_error_after_bytes(upload, data, database, error_code, after_bytes, last_byte_persisted, chunk_first_byte, test_id=0):
+    """
+    Handle error-after-bytes instructions for resumable uploads and commit only partial data before forcing a testbench error.
+    This helper method also ignores request bytes that have already been persisted, which aligns with GCS behavior.
+    """
+    if last_byte_persisted < after_bytes:
+        range_start = 0
+        # Ignore request bytes that have already been persisted.
+        if int(chunk_first_byte) < last_byte_persisted:
+            range_start = last_byte_persisted - int(chunk_first_byte) + 1
+        elif int(chunk_first_byte) == last_byte_persisted and last_byte_persisted != 0:
+            range_start = int(chunk_first_byte) + 1
+        data = testbench.common.partial_media(data, range_end=after_bytes, range_start=range_start)
+        upload.media += data
+        upload.complete = False
+    if len(upload.media) >= after_bytes:
+        if test_id:
+            database.dequeue_next_instruction(test_id, "storage.objects.insert")
+        testbench.error.generic(
+            "Fault injected during a resumable upload",
+            rest_code=error_code,
+            grpc_code=None,
+            context=None,
+        )
+
 
 def handle_gzip_request(request):
     """
diff --git a/testbench/rest_server.py b/testbench/rest_server.py
index a84c278b..3e32b419 100644
--- a/testbench/rest_server.py
+++ b/testbench/rest_server.py
@@ -950,8 +950,6 @@ def resumable_upload_chunk(bucket_name):
     upload = db.get_upload(upload_id, None)
     if upload.complete:
         return gcs_type.object.Object.rest(upload.metadata)
-    # The server should ignore any request bytes that have already been persisted.
-    # Thus we check the last_byte_persisted in the upload.
     last_byte_persisted = 0 if len(upload.media) == 0 else (len(upload.media) - 1)
     upload.transfer.add(request.environ.get("HTTP_TRANSFER_ENCODING", ""))
     content_length = request.headers.get("content-length", None)
@@ -1028,34 +1026,17 @@ def resumable_upload_chunk(bucket_name):
                 None,
                 rest_code=400,
             )
-        ### Handle "return-503-after-256K" instructions for uploads
-        ### Probably move this entire logic to a helper method?!
+        ### Handle error-after-bytes instructions, either retry test or x-goog-emulator-instructions.
         instruction = testbench.common.extract_instruction(request, context=None)
-        error_code, after_bytes, test_id = testbench.common.handle_retry_resumable_uploads_error_after_bytes(upload, data, db, request)
+        error_code, after_bytes, test_id = testbench.common.get_retry_uploads_error_after_bytes(db, request)
         if instruction or error_code:
             if instruction == "return-503-after-256K":
                 error_code = 503
                 after_bytes = 262144
-            if last_byte_persisted < after_bytes:
-                range_start = 0
-                # Ignore request bytes that have already been persisted.
-                if int(chunk_first_byte) < last_byte_persisted:
-                    range_start = last_byte_persisted - int(chunk_first_byte) + 1
-                elif int(chunk_first_byte) == last_byte_persisted and last_byte_persisted != 0:
-                    range_start = int(chunk_first_byte) + 1
-                data = testbench.common.partial_media(data, range_end=after_bytes, range_start=range_start)
-                upload.media += data
-                upload.complete = False
-            if len(upload.media) >= after_bytes:
-                if test_id:
-                    db.dequeue_next_instruction(test_id, "storage.objects.insert")
-                testbench.error.generic(
-                    "Fault injected during a resumable upload",
-                    rest_code=error_code,
-                    grpc_code=None,
-                    context=None,
-                )
-        # Validate chunk_first_byte to ignore any request bytes that have already been persisted.
+            testbench.common.handle_retry_uploads_error_after_bytes(upload, data, db, error_code, after_bytes, last_byte_persisted, chunk_first_byte, test_id)
+        # The testbench should ignore any request bytes that have already been persisted,
+        # to be aligned with GCS behavior (https://cloud.google.com/storage/docs/resumable-uploads#resent-data).
+        # Thus we validate chunk_first_byte against last_byte_persisted.
         range_start = 0
         if chunk_first_byte != "*":
             if int(chunk_first_byte) < last_byte_persisted:

From 01517ab89023723ec00936458171f73f0f1ab8d5 Mon Sep 17 00:00:00 2001
From: Cathy Ouyang <cathyo@google.com>
Date: Fri, 23 Jun 2023 15:18:14 -0700
Subject: [PATCH 08/14] black lint

---
 testbench/common.py      | 22 ++++++++++++++++++----
 testbench/rest_server.py | 27 ++++++++++++++++++++++-----
 2 files changed, 40 insertions(+), 9 deletions(-)

diff --git a/testbench/common.py b/testbench/common.py
index c34a156a..9156b846 100644
--- a/testbench/common.py
+++ b/testbench/common.py
@@ -612,7 +612,7 @@ def corrupt_media(media):
 
 
 def partial_media(media, range_end, range_start=0):
-    """ Returns partial media due to forced interruption or server validation. """
+    """Returns partial media due to forced interruption or server validation."""
     return media[range_start:range_end]
 
 
@@ -830,7 +830,9 @@ def get_retry_uploads_error_after_bytes(database, request):
     next_instruction = database.peek_next_instruction(test_id, "storage.objects.insert")
     if not next_instruction:
         return 0, 0, ""
-    error_after_bytes_matches = testbench.common.retry_return_error_after_bytes.match(next_instruction)
+    error_after_bytes_matches = testbench.common.retry_return_error_after_bytes.match(
+        next_instruction
+    )
     if error_after_bytes_matches:
         items = list(error_after_bytes_matches.groups())
         error_code = int(items[0])
@@ -838,7 +840,17 @@ def get_retry_uploads_error_after_bytes(database, request):
         return error_code, after_bytes, test_id
     return 0, 0, ""
 
-def handle_retry_uploads_error_after_bytes(upload, data, database, error_code, after_bytes, last_byte_persisted, chunk_first_byte, test_id=0):
+
+def handle_retry_uploads_error_after_bytes(
+    upload,
+    data,
+    database,
+    error_code,
+    after_bytes,
+    last_byte_persisted,
+    chunk_first_byte,
+    test_id=0,
+):
     """
     Handle error-after-bytes instructions for resumable uploads and commit only partial data before forcing a testbench error.
     This helper method also ignores request bytes that have already been persisted, which aligns with GCS behavior.
@@ -850,7 +862,9 @@ def handle_retry_uploads_error_after_bytes(upload, data, database, error_code, a
             range_start = last_byte_persisted - int(chunk_first_byte) + 1
         elif int(chunk_first_byte) == last_byte_persisted and last_byte_persisted != 0:
             range_start = int(chunk_first_byte) + 1
-        data = testbench.common.partial_media(data, range_end=after_bytes, range_start=range_start)
+        data = testbench.common.partial_media(
+            data, range_end=after_bytes, range_start=range_start
+        )
         upload.media += data
         upload.complete = False
     if len(upload.media) >= after_bytes:
diff --git a/testbench/rest_server.py b/testbench/rest_server.py
index 3e32b419..2608fb33 100644
--- a/testbench/rest_server.py
+++ b/testbench/rest_server.py
@@ -960,7 +960,6 @@ def resumable_upload_chunk(bucket_name):
         testbench.error.invalid("content-length header", None)
     content_range = request.headers.get("content-range")
     custom_header_value = request.headers.get("x-goog-emulator-custom-header")
-
     if content_range is not None:
         items = list(testbench.common.content_range_split.match(content_range).groups())
         # TODO(#27) - maybe this should be an assert()
@@ -1028,12 +1027,25 @@ def resumable_upload_chunk(bucket_name):
             )
         ### Handle error-after-bytes instructions, either retry test or x-goog-emulator-instructions.
         instruction = testbench.common.extract_instruction(request, context=None)
-        error_code, after_bytes, test_id = testbench.common.get_retry_uploads_error_after_bytes(db, request)
+        (
+            error_code,
+            after_bytes,
+            test_id,
+        ) = testbench.common.get_retry_uploads_error_after_bytes(db, request)
         if instruction or error_code:
             if instruction == "return-503-after-256K":
                 error_code = 503
                 after_bytes = 262144
-            testbench.common.handle_retry_uploads_error_after_bytes(upload, data, db, error_code, after_bytes, last_byte_persisted, chunk_first_byte, test_id)
+            testbench.common.handle_retry_uploads_error_after_bytes(
+                upload,
+                data,
+                db,
+                error_code,
+                after_bytes,
+                last_byte_persisted,
+                chunk_first_byte,
+                test_id,
+            )
         # The testbench should ignore any request bytes that have already been persisted,
         # to be aligned with GCS behavior (https://cloud.google.com/storage/docs/resumable-uploads#resent-data).
         # Thus we validate chunk_first_byte against last_byte_persisted.
@@ -1041,9 +1053,14 @@ def resumable_upload_chunk(bucket_name):
         if chunk_first_byte != "*":
             if int(chunk_first_byte) < last_byte_persisted:
                 range_start = last_byte_persisted - int(chunk_first_byte) + 1
-            elif int(chunk_first_byte) == last_byte_persisted and last_byte_persisted != 0:
+            elif (
+                int(chunk_first_byte) == last_byte_persisted
+                and last_byte_persisted != 0
+            ):
                 range_start = int(chunk_first_byte) + 1
-        data = testbench.common.partial_media(data, range_end=(chunk_last_byte + 1), range_start=range_start)
+        data = testbench.common.partial_media(
+            data, range_end=(chunk_last_byte + 1), range_start=range_start
+        )
 
         upload.media += data
         upload.complete = total_object_size == len(upload.media) or (

From baf45cee807dd8c7be2fb71c54145af3640523b7 Mon Sep 17 00:00:00 2001
From: Cathy Ouyang <cathyo@google.com>
Date: Thu, 29 Jun 2023 11:32:02 -0700
Subject: [PATCH 09/14] update tests

---
 tests/test_testbench_object_upload.py |  2 +-
 tests/test_testbench_retry.py         | 33 ++++++++++++++++-----------
 2 files changed, 21 insertions(+), 14 deletions(-)

diff --git a/tests/test_testbench_object_upload.py b/tests/test_testbench_object_upload.py
index ba81bcfb..87ab6eaf 100644
--- a/tests/test_testbench_object_upload.py
+++ b/tests/test_testbench_object_upload.py
@@ -268,7 +268,7 @@ def test_upload_resumable_x_upload_content_length(self):
             "/upload/storage/v1/b/bucket-name/o",
             query_string={"upload_id": upload_id},
             headers={
-                "content-range": "bytes {last:d}-*/*".format(last=len(chunk) - 1),
+                "content-range": "bytes {last:d}-*/*".format(last=len(chunk)),
             },
             data=chunk,
         )
diff --git a/tests/test_testbench_retry.py b/tests/test_testbench_retry.py
index 9279b54e..d84a96f7 100644
--- a/tests/test_testbench_retry.py
+++ b/tests/test_testbench_retry.py
@@ -438,16 +438,8 @@ def test_retry_test_return_error_after_bytes(self):
             "/storage/v1/b", data=json.dumps({"name": "bucket-name"})
         )
         self.assertEqual(response.status_code, 200)
-        # Use the XML API to inject an object with some data.
-        media = self._create_block(256)
-        response = self.client.put(
-            "/bucket-name/256k.txt",
-            content_type="text/plain",
-            data=media,
-        )
-        self.assertEqual(response.status_code, 200)
 
-        # Setup a failure for reading back the object.
+        # Setup a failure for uploading an object.
         response = self.client.post(
             "/retry_test",
             data=json.dumps(
@@ -481,11 +473,24 @@ def test_retry_test_return_error_after_bytes(self):
             "/upload/storage/v1/b/bucket-name/o",
             query_string={"upload_id": upload_id},
             headers={
-                "content-range": "bytes 0-{len:d}/*".format(len=UPLOAD_QUANTUM - 1),
+                "content-range": "bytes 0-{len:d}/{obj_size:d}".format(
+                    len=UPLOAD_QUANTUM - 1, obj_size=2 * UPLOAD_QUANTUM
+                ),
                 "x-retry-test-id": id,
             },
             data=chunk,
         )
+        self.assertEqual(response.status_code, 504, msg=response.data)
+
+        # Check the status of a resumable upload.
+        response = self.client.put(
+            "/upload/storage/v1/b/bucket-name/o",
+            query_string={"upload_id": upload_id},
+            headers={
+                "content-range": "bytes */*",
+                "x-retry-test-id": id,
+            },
+        )
         self.assertEqual(response.status_code, 308, msg=response.data)
         self.assertIn("range", response.headers)
         self.assertEqual(
@@ -496,14 +501,16 @@ def test_retry_test_return_error_after_bytes(self):
             "/upload/storage/v1/b/bucket-name/o",
             query_string={"upload_id": upload_id},
             headers={
-                "content-range": "bytes {beg:d}-{end:d}/*".format(
-                    beg=UPLOAD_QUANTUM, end=2 * UPLOAD_QUANTUM - 1
+                "content-range": "bytes {beg:d}-{end:d}/{obj_size:d}".format(
+                    beg=UPLOAD_QUANTUM,
+                    end=2 * UPLOAD_QUANTUM - 1,
+                    obj_size=2 * UPLOAD_QUANTUM,
                 ),
                 "x-retry-test-id": id,
             },
             data=chunk,
         )
-        self.assertEqual(response.status_code, 504, msg=response.data)
+        self.assertEqual(response.status_code, 200, msg=response.data)
 
 
 if __name__ == "__main__":

From 2f940e9004832fa0862d7829d3104f15cde2163c Mon Sep 17 00:00:00 2001
From: Cathy Ouyang <cathyo@google.com>
Date: Thu, 29 Jun 2023 15:33:58 -0700
Subject: [PATCH 10/14] handle after_byte instruction only when neccessary

---
 .../LICENSE                                   | 202 +++++++++++++
 .../METADATA                                  | 279 ++++++++++++++++++
 .../top_level.txt                             |   5 +
 testbench/common.py                           |   9 +-
 testbench/rest_server.py                      |   3 +-
 5 files changed, 495 insertions(+), 3 deletions(-)
 create mode 100644 pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/LICENSE
 create mode 100644 pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/METADATA
 create mode 100644 pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/top_level.txt

diff --git a/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/LICENSE b/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/LICENSE
new file mode 100644
index 00000000..d6456956
--- /dev/null
+++ b/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/LICENSE
@@ -0,0 +1,202 @@
+
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright [yyyy] [name of copyright owner]
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
diff --git a/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/METADATA b/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/METADATA
new file mode 100644
index 00000000..1d7ca6ac
--- /dev/null
+++ b/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/METADATA
@@ -0,0 +1,279 @@
+Metadata-Version: 2.1
+Name: googleapis-storage-testbench
+Version: 0.33.0
+Summary: A testbench for Google Cloud Storage client libraries
+Home-page: https://github.com/googleapis/storage-testbench
+Author: Google LLC
+Author-email: googleapis-packages@google.com
+Project-URL: Bug Tracker, https://github.com/googleapis/storage-testbench/issues
+Classifier: Programming Language :: Python :: 3
+Classifier: License :: OSI Approved :: Apache-2.0
+Classifier: Operating System :: OS Independent
+Requires-Python: >=3.6
+Description-Content-Type: text/markdown
+License-File: LICENSE
+Requires-Dist: grpcio (==1.48.2)
+Requires-Dist: googleapis-common-protos (==1.59.0)
+Requires-Dist: protobuf (==4.23.2)
+Requires-Dist: flask (==2.3.2)
+Requires-Dist: requests-toolbelt (==1.0.0)
+Requires-Dist: scalpl (==0.4.2)
+Requires-Dist: crc32c (==2.3)
+Requires-Dist: gunicorn (==20.1.0)
+Requires-Dist: waitress (==2.1.2)
+
+# Storage Testbench
+
+**This is not an officially supported Google product**
+
+This repository is used by Storage Client libraries to test integration tests locally
+and reproduce Storage API transient errors. The testbench emulates the Storage API and
+is expected to be used by Storage library maintainers.
+
+
+## Table of Contents
+- [Storage Testbench](#storage-testbench)
+  - [Table of Contents](#table-of-contents)
+  - [Issue Policy](#issue-policy)
+  - [What is this testbench?](#what-is-this-testbench)
+  - [When to use this testbench](#when-to-use-this-testbench)
+  - [How to use this testbench](#how-to-use-this-testbench)
+    - [Initial set up](#initial-set-up)
+    - [Run the testbench](#run-the-testbench)
+    - [Start the gRPC server](#start-the-gRPC-server)
+    - [Check that the testbench is running](#check-that-the-testbench-is-running)
+  - [Updating Proto Files](#updating-proto-files)
+  - [Force Failures](#force-failures)
+    - [return-broken-stream](#return-broken-stream)
+    - [return-corrupted-data](#return-corrupted-data)
+    - [stall-always](#stall-always)
+    - [stall-at-256KiB](#stall-at-256kib)
+    - [return-503-after-256K](#return-503-after-256k)
+    - [return-503-after-256K/retry-N](#return-503-after-256kretry-n)
+  - [Retry Test API](#retry-test-api)
+    - [Creating a new Retry Test](#creating-a-new-retry-test)
+    - [Get a Retry Test resource](#get-a-retry-test-resource)
+    - [Delete a Retry Test resource](#delete-a-retry-test-resource)
+    - [Causing a failure using x-retry-test-id header](#causing-a-failure-using-x-retry-test-id-header)
+    - [Forced Failures Supported](#forced-failures-supported)
+
+## Issue Policy
+
+Repository provides no dedicated support for issues filed.
+Issues will be addressed when time permits.
+
+## What is this testbench?
+
+This testbench fakes the Google Cloud Storage (GCS) API. You can configure the GCS client libraries to make calls to this fake rather than to the actual API.
+* The testbench fakes the JSON API, both over REST and gRPC. It has limited support for the XML API.
+* Generally, the error codes are similar to the ones generated by GCS, but the error messages are not.
+* The testbench performs far fewer error checks, and no permission checks (ACL/IAM).
+
+## When to use this testbench
+
+In general, this testbench is best suited for integration tests that are hard (or just annoying) to reliably run against production. The primary example of this are errors that make the client library go through its retry path.
+
+This testbench can be useful to test HMAC keys, which are really hard to test against production due to quota restrictions.
+
+It is useful as well to test features that are not yet deployed to production: you can implement them in the testbench and then write the library code before production is "ready".
+
+## How to use this testbench
+
+### Initial set up
+
+1. [Set up python if you haven't already](https://cloud.google.com/python/docs/setup)
+2. [Clone this repository](https://docs.github.com/en/github/creating-cloning-and-archiving-repositories/cloning-a-repository-from-github/cloning-a-repository#cloning-a-repository)
+
+   From the terminal:
+   ```bash
+   git clone https://github.com/googleapis/storage-testbench.git
+   ```
+3. Switch to the cloned directory:
+   ```bash
+   cd storage-testbench
+    ```
+4. [Create a virtual environment](https://cloud.google.com/python/docs/setup#installing_and_using_virtualenv)
+    * keep this virtual environment active whenever you run the testbench
+5. Install dependencies:
+    ```bash
+    pip install -e .
+    ```
+
+### Run the testbench
+
+To start the testbench, run this command from a terminal:
+
+ On Non-Windows
+```bash
+python3 testbench_run.py localhost 9000 10
+```
+
+On Windows
+```bash
+py testbench_run.py localhost 9000 10
+```
+
+> ⚠️ Ensure that the virtual environment you created to install the dependencies is active.
+
+
+### Start the gRPC server
+
+If you want to test the gRPC API, you must make a request which will start
+the testbench's gRPC server. Also, the gRPC server must run on a different port
+than the HTTP server. To serve gRPC requests on port 8888, run:
+
+```bash
+curl -s --retry 5 --retry-max-time 40 "http://localhost:9000/start_grpc?port=8888"
+```
+
+### Check that the testbench is running
+
+Ensure the testbench is running by sending it a request from a different terminal, such as:
+
+```bash
+curl -X GET localhost:9000
+```
+
+The response you get should be: `OK`
+
+Now you can use the testbench (while it's running) with the client libraries.
+
+## Updating Proto Files
+
+From time to time you may need to update the files generated by protobuf and
+gRPC. To do so, clone the [protos](https://github.com/googleapis/googleapis) and
+run the grpc_tools generator:
+
+```shell
+cd $HOME/storage-testbench
+
+# This creates a new directory with the protos from `googleapis`.  If
+# the clone already exists use:
+#    git -C $HOME/googleapis pull
+git -C $HOME clone https://github.com/googleapis/googleapis
+
+pip install grpcio-tools
+python -m grpc_tools.protoc -I$HOME/googleapis \
+    --python_out=. --grpc_python_out=. \
+    $HOME/googleapis/google/iam/v1/iam_policy.proto
+python -m grpc_tools.protoc -I$HOME/googleapis \
+    --python_out=. --grpc_python_out=. \
+    $HOME/googleapis/google/iam/v1/options.proto
+python -m grpc_tools.protoc -I$HOME/googleapis \
+    --python_out=. --grpc_python_out=. \
+    $HOME/googleapis/google/iam/v1/policy.proto
+python -m grpc_tools.protoc -I$HOME/googleapis \
+    --python_out=. --grpc_python_out=. \
+    $HOME/googleapis/google/storage/v2/storage.proto
+```
+
+Then commit the files generated in `google/**`:
+
+```shell
+git commit -m"chore: update protos" google
+```
+
+## Force Failures
+
+You can force the following failures by using the `x-goog-emulator-instructions` header.
+The `x-goog-testbench-instructions` header is deprecated, but supported for
+backwards compatibility and provides the same functionality as
+`x-goog-emulator-instructions`, please change your code to use `x-goog-emulator-instructions` instead.
+
+### return-broken-stream
+
+Set request headers with `x-goog-emulator-instructions: return-broken-stream`.
+Testbench will fail after sending 1024*1024 bytes.
+
+### return-corrupted-data
+
+Set request headers with `x-goog-emulator-instructions: return-corrupted-data`.
+Testbench will return corrupted data.
+
+### stall-always
+
+Set request headers with `x-goog-emulator-instructions: stall-always`.
+Testbench will stall at the beginning.
+
+### stall-at-256KiB
+
+Set request headers with `x-goog-emulator-instructions: stall-at-256KiB`.
+Testbench will stall at 256KiB bytes.
+
+### return-503-after-256K
+
+Set request headers with `x-goog-emulator-instructions: return-503-after-256K`.
+Testbench will return a `HTTP 503` after sending 256KiB bytes.
+
+### return-503-after-256K/retry-N
+
+Set request headers with `x-goog-emulator-instructions: return-503-after-256K/retry-1` up to `x-goog-emulator-instructions: return-503-after-256K/retry-N`.
+
+For N==1 and N==2 behave like `return-305-after-256K`, for `N>=3` ignore the
+failure instruction and return successfully. This is used to test failures during
+retry, the client cooperates by sending the retry counter in the failure
+instructions.
+
+
+## Retry Test API
+
+The "Retry Test API" offers a mechanism to describe more complex retry scenarios
+while sending a single, constant header through all the HTTP requests from a
+test program. Retry Test provides accounting of failures used to validate
+the expected failures were experienced by the testbench and not accidentally missed.
+
+Previous versions of the GCS testbench used a custom header in the RPC to
+control the behavior of each RPC, for some test scenarios this required sending
+different header with the first retry attempt vs. subsequent attempts. Producing
+different headers in each attempt is not easy to implement with some client libraries.
+
+Sending a constant header with all RPCs can be implemented across all client libraries,
+and to some degree decouples the test setup from the test execution.
+
+### Creating a new Retry Test
+
+The following cURL request will create a Retry Test resource which emits a 503
+when a buckets list operation is received by the testbench with the returned
+retry test ID.
+
+```bash
+curl -X POST "http://localhost:9000/retry_test" -H 'Content-Type: application/json' \
+     -d '{"instructions":{"storage.buckets.list": ["return-503"]}}'
+```
+
+### Get a Retry Test resource
+
+Get Retry Test resource by id "1d05c20627844214a9ff7cbcf696317d".
+
+```bash
+curl -X GET "http://localhost:9000/retry_test/1d05c20627844214a9ff7cbcf696317d"
+```
+
+### Delete a Retry Test resource
+
+Delete Retry Test resource by id "1d05c20627844214a9ff7cbcf696317d".
+
+```bash
+curl -X DELETE "http://localhost:9000/retry_test/1d05c20627844214a9ff7cbcf696317d"
+```
+
+### Causing a failure using x-retry-test-id header
+
+The following cURL request will attempt to list buckets and the testbench will emit
+a `503` error once based on the Retry Test created above. Subsequent list buckets
+operations will succeed.
+
+```bash
+curl -H "x-retry-test-id: 1d05c20627844214a9ff7cbcf696317d" "http://localhost:9100/storage/v1/b?project=test"
+```
+
+### Forced Failures Supported
+
+| Failure Id              | Description
+| ----------------------- | ---
+| return-X                                  | Testbench will fail with HTTP code provided for `X`, e.g. return-503 returns a 503
+| return-X-after-YK                         | Testbench will return X after YKiB of uploaded data
+| return-broken-stream-final-chunk-after-YB | Testbench will break connection on final chunk of a resumable upload after Y bytes
+| return-broken-stream                      | Testbench will fail after a few bytes
+| return-broken-stream-after-YK             | Testbench will fail after YKiB of downloaded data
+| return-reset-connection                   | Testbench will fail with a reset connection
diff --git a/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/top_level.txt b/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/top_level.txt
new file mode 100644
index 00000000..01d339b0
--- /dev/null
+++ b/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/top_level.txt
@@ -0,0 +1,5 @@
+gcs
+google/iam/v1
+google/storage/v2
+testbench
+testbench/servers
diff --git a/testbench/common.py b/testbench/common.py
index 9156b846..a2a2b1a1 100644
--- a/testbench/common.py
+++ b/testbench/common.py
@@ -849,21 +849,26 @@ def handle_retry_uploads_error_after_bytes(
     after_bytes,
     last_byte_persisted,
     chunk_first_byte,
+    chunk_last_byte,
     test_id=0,
 ):
     """
     Handle error-after-bytes instructions for resumable uploads and commit only partial data before forcing a testbench error.
     This helper method also ignores request bytes that have already been persisted, which aligns with GCS behavior.
     """
-    if last_byte_persisted < after_bytes:
+    if after_bytes > last_byte_persisted and after_bytes <= (chunk_last_byte + 1):
         range_start = 0
         # Ignore request bytes that have already been persisted.
         if int(chunk_first_byte) < last_byte_persisted:
             range_start = last_byte_persisted - int(chunk_first_byte) + 1
         elif int(chunk_first_byte) == last_byte_persisted and last_byte_persisted != 0:
             range_start = int(chunk_first_byte) + 1
+        range_end = len(data)
+        # Only partial data will be commited due to the instructed interruption.
+        if after_bytes <= chunk_last_byte:
+            range_end = len(data) - (chunk_last_byte - after_bytes + 1)
         data = testbench.common.partial_media(
-            data, range_end=after_bytes, range_start=range_start
+            data, range_end=range_end, range_start=range_start
         )
         upload.media += data
         upload.complete = False
diff --git a/testbench/rest_server.py b/testbench/rest_server.py
index 2608fb33..16a496f1 100644
--- a/testbench/rest_server.py
+++ b/testbench/rest_server.py
@@ -1009,6 +1009,7 @@ def resumable_upload_chunk(bucket_name):
         if chunk_last_byte == "*":
             x_upload_content_length = len(upload.media)
             chunk_last_byte = len(upload.media) - 1
+            # chunk_last_byte = (int(chunk_first_byte) + len(data) - 1) if chunk_first_byte != "*" else last_byte_persisted + len(data)
         else:
             chunk_last_byte = int(chunk_last_byte)
         total_object_size = (
@@ -1044,6 +1045,7 @@ def resumable_upload_chunk(bucket_name):
                 after_bytes,
                 last_byte_persisted,
                 chunk_first_byte,
+                chunk_last_byte,
                 test_id,
             )
         # The testbench should ignore any request bytes that have already been persisted,
@@ -1061,7 +1063,6 @@ def resumable_upload_chunk(bucket_name):
         data = testbench.common.partial_media(
             data, range_end=(chunk_last_byte + 1), range_start=range_start
         )
-
         upload.media += data
         upload.complete = total_object_size == len(upload.media) or (
             chunk_last_byte + 1 == total_object_size

From adff1212e7517ae7f926d9145ff353c5ab748df4 Mon Sep 17 00:00:00 2001
From: Cathy Ouyang <cathyo@google.com>
Date: Thu, 29 Jun 2023 15:38:36 -0700
Subject: [PATCH 11/14] fix previous commit

---
 .../LICENSE                                   | 202 -------------
 .../METADATA                                  | 279 ------------------
 .../top_level.txt                             |   5 -
 testbench/rest_server.py                      |   1 -
 4 files changed, 487 deletions(-)
 delete mode 100644 pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/LICENSE
 delete mode 100644 pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/METADATA
 delete mode 100644 pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/top_level.txt

diff --git a/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/LICENSE b/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/LICENSE
deleted file mode 100644
index d6456956..00000000
--- a/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/LICENSE
+++ /dev/null
@@ -1,202 +0,0 @@
-
-                                 Apache License
-                           Version 2.0, January 2004
-                        http://www.apache.org/licenses/
-
-   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
-
-   1. Definitions.
-
-      "License" shall mean the terms and conditions for use, reproduction,
-      and distribution as defined by Sections 1 through 9 of this document.
-
-      "Licensor" shall mean the copyright owner or entity authorized by
-      the copyright owner that is granting the License.
-
-      "Legal Entity" shall mean the union of the acting entity and all
-      other entities that control, are controlled by, or are under common
-      control with that entity. For the purposes of this definition,
-      "control" means (i) the power, direct or indirect, to cause the
-      direction or management of such entity, whether by contract or
-      otherwise, or (ii) ownership of fifty percent (50%) or more of the
-      outstanding shares, or (iii) beneficial ownership of such entity.
-
-      "You" (or "Your") shall mean an individual or Legal Entity
-      exercising permissions granted by this License.
-
-      "Source" form shall mean the preferred form for making modifications,
-      including but not limited to software source code, documentation
-      source, and configuration files.
-
-      "Object" form shall mean any form resulting from mechanical
-      transformation or translation of a Source form, including but
-      not limited to compiled object code, generated documentation,
-      and conversions to other media types.
-
-      "Work" shall mean the work of authorship, whether in Source or
-      Object form, made available under the License, as indicated by a
-      copyright notice that is included in or attached to the work
-      (an example is provided in the Appendix below).
-
-      "Derivative Works" shall mean any work, whether in Source or Object
-      form, that is based on (or derived from) the Work and for which the
-      editorial revisions, annotations, elaborations, or other modifications
-      represent, as a whole, an original work of authorship. For the purposes
-      of this License, Derivative Works shall not include works that remain
-      separable from, or merely link (or bind by name) to the interfaces of,
-      the Work and Derivative Works thereof.
-
-      "Contribution" shall mean any work of authorship, including
-      the original version of the Work and any modifications or additions
-      to that Work or Derivative Works thereof, that is intentionally
-      submitted to Licensor for inclusion in the Work by the copyright owner
-      or by an individual or Legal Entity authorized to submit on behalf of
-      the copyright owner. For the purposes of this definition, "submitted"
-      means any form of electronic, verbal, or written communication sent
-      to the Licensor or its representatives, including but not limited to
-      communication on electronic mailing lists, source code control systems,
-      and issue tracking systems that are managed by, or on behalf of, the
-      Licensor for the purpose of discussing and improving the Work, but
-      excluding communication that is conspicuously marked or otherwise
-      designated in writing by the copyright owner as "Not a Contribution."
-
-      "Contributor" shall mean Licensor and any individual or Legal Entity
-      on behalf of whom a Contribution has been received by Licensor and
-      subsequently incorporated within the Work.
-
-   2. Grant of Copyright License. Subject to the terms and conditions of
-      this License, each Contributor hereby grants to You a perpetual,
-      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
-      copyright license to reproduce, prepare Derivative Works of,
-      publicly display, publicly perform, sublicense, and distribute the
-      Work and such Derivative Works in Source or Object form.
-
-   3. Grant of Patent License. Subject to the terms and conditions of
-      this License, each Contributor hereby grants to You a perpetual,
-      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
-      (except as stated in this section) patent license to make, have made,
-      use, offer to sell, sell, import, and otherwise transfer the Work,
-      where such license applies only to those patent claims licensable
-      by such Contributor that are necessarily infringed by their
-      Contribution(s) alone or by combination of their Contribution(s)
-      with the Work to which such Contribution(s) was submitted. If You
-      institute patent litigation against any entity (including a
-      cross-claim or counterclaim in a lawsuit) alleging that the Work
-      or a Contribution incorporated within the Work constitutes direct
-      or contributory patent infringement, then any patent licenses
-      granted to You under this License for that Work shall terminate
-      as of the date such litigation is filed.
-
-   4. Redistribution. You may reproduce and distribute copies of the
-      Work or Derivative Works thereof in any medium, with or without
-      modifications, and in Source or Object form, provided that You
-      meet the following conditions:
-
-      (a) You must give any other recipients of the Work or
-          Derivative Works a copy of this License; and
-
-      (b) You must cause any modified files to carry prominent notices
-          stating that You changed the files; and
-
-      (c) You must retain, in the Source form of any Derivative Works
-          that You distribute, all copyright, patent, trademark, and
-          attribution notices from the Source form of the Work,
-          excluding those notices that do not pertain to any part of
-          the Derivative Works; and
-
-      (d) If the Work includes a "NOTICE" text file as part of its
-          distribution, then any Derivative Works that You distribute must
-          include a readable copy of the attribution notices contained
-          within such NOTICE file, excluding those notices that do not
-          pertain to any part of the Derivative Works, in at least one
-          of the following places: within a NOTICE text file distributed
-          as part of the Derivative Works; within the Source form or
-          documentation, if provided along with the Derivative Works; or,
-          within a display generated by the Derivative Works, if and
-          wherever such third-party notices normally appear. The contents
-          of the NOTICE file are for informational purposes only and
-          do not modify the License. You may add Your own attribution
-          notices within Derivative Works that You distribute, alongside
-          or as an addendum to the NOTICE text from the Work, provided
-          that such additional attribution notices cannot be construed
-          as modifying the License.
-
-      You may add Your own copyright statement to Your modifications and
-      may provide additional or different license terms and conditions
-      for use, reproduction, or distribution of Your modifications, or
-      for any such Derivative Works as a whole, provided Your use,
-      reproduction, and distribution of the Work otherwise complies with
-      the conditions stated in this License.
-
-   5. Submission of Contributions. Unless You explicitly state otherwise,
-      any Contribution intentionally submitted for inclusion in the Work
-      by You to the Licensor shall be under the terms and conditions of
-      this License, without any additional terms or conditions.
-      Notwithstanding the above, nothing herein shall supersede or modify
-      the terms of any separate license agreement you may have executed
-      with Licensor regarding such Contributions.
-
-   6. Trademarks. This License does not grant permission to use the trade
-      names, trademarks, service marks, or product names of the Licensor,
-      except as required for reasonable and customary use in describing the
-      origin of the Work and reproducing the content of the NOTICE file.
-
-   7. Disclaimer of Warranty. Unless required by applicable law or
-      agreed to in writing, Licensor provides the Work (and each
-      Contributor provides its Contributions) on an "AS IS" BASIS,
-      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
-      implied, including, without limitation, any warranties or conditions
-      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
-      PARTICULAR PURPOSE. You are solely responsible for determining the
-      appropriateness of using or redistributing the Work and assume any
-      risks associated with Your exercise of permissions under this License.
-
-   8. Limitation of Liability. In no event and under no legal theory,
-      whether in tort (including negligence), contract, or otherwise,
-      unless required by applicable law (such as deliberate and grossly
-      negligent acts) or agreed to in writing, shall any Contributor be
-      liable to You for damages, including any direct, indirect, special,
-      incidental, or consequential damages of any character arising as a
-      result of this License or out of the use or inability to use the
-      Work (including but not limited to damages for loss of goodwill,
-      work stoppage, computer failure or malfunction, or any and all
-      other commercial damages or losses), even if such Contributor
-      has been advised of the possibility of such damages.
-
-   9. Accepting Warranty or Additional Liability. While redistributing
-      the Work or Derivative Works thereof, You may choose to offer,
-      and charge a fee for, acceptance of support, warranty, indemnity,
-      or other liability obligations and/or rights consistent with this
-      License. However, in accepting such obligations, You may act only
-      on Your own behalf and on Your sole responsibility, not on behalf
-      of any other Contributor, and only if You agree to indemnify,
-      defend, and hold each Contributor harmless for any liability
-      incurred by, or claims asserted against, such Contributor by reason
-      of your accepting any such warranty or additional liability.
-
-   END OF TERMS AND CONDITIONS
-
-   APPENDIX: How to apply the Apache License to your work.
-
-      To apply the Apache License to your work, attach the following
-      boilerplate notice, with the fields enclosed by brackets "[]"
-      replaced with your own identifying information. (Don't include
-      the brackets!)  The text should be enclosed in the appropriate
-      comment syntax for the file format. We also recommend that a
-      file or class name and description of purpose be included on the
-      same "printed page" as the copyright notice for easier
-      identification within third-party archives.
-
-   Copyright [yyyy] [name of copyright owner]
-
-   Licensed under the Apache License, Version 2.0 (the "License");
-   you may not use this file except in compliance with the License.
-   You may obtain a copy of the License at
-
-       http://www.apache.org/licenses/LICENSE-2.0
-
-   Unless required by applicable law or agreed to in writing, software
-   distributed under the License is distributed on an "AS IS" BASIS,
-   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-   See the License for the specific language governing permissions and
-   limitations under the License.
diff --git a/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/METADATA b/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/METADATA
deleted file mode 100644
index 1d7ca6ac..00000000
--- a/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/METADATA
+++ /dev/null
@@ -1,279 +0,0 @@
-Metadata-Version: 2.1
-Name: googleapis-storage-testbench
-Version: 0.33.0
-Summary: A testbench for Google Cloud Storage client libraries
-Home-page: https://github.com/googleapis/storage-testbench
-Author: Google LLC
-Author-email: googleapis-packages@google.com
-Project-URL: Bug Tracker, https://github.com/googleapis/storage-testbench/issues
-Classifier: Programming Language :: Python :: 3
-Classifier: License :: OSI Approved :: Apache-2.0
-Classifier: Operating System :: OS Independent
-Requires-Python: >=3.6
-Description-Content-Type: text/markdown
-License-File: LICENSE
-Requires-Dist: grpcio (==1.48.2)
-Requires-Dist: googleapis-common-protos (==1.59.0)
-Requires-Dist: protobuf (==4.23.2)
-Requires-Dist: flask (==2.3.2)
-Requires-Dist: requests-toolbelt (==1.0.0)
-Requires-Dist: scalpl (==0.4.2)
-Requires-Dist: crc32c (==2.3)
-Requires-Dist: gunicorn (==20.1.0)
-Requires-Dist: waitress (==2.1.2)
-
-# Storage Testbench
-
-**This is not an officially supported Google product**
-
-This repository is used by Storage Client libraries to test integration tests locally
-and reproduce Storage API transient errors. The testbench emulates the Storage API and
-is expected to be used by Storage library maintainers.
-
-
-## Table of Contents
-- [Storage Testbench](#storage-testbench)
-  - [Table of Contents](#table-of-contents)
-  - [Issue Policy](#issue-policy)
-  - [What is this testbench?](#what-is-this-testbench)
-  - [When to use this testbench](#when-to-use-this-testbench)
-  - [How to use this testbench](#how-to-use-this-testbench)
-    - [Initial set up](#initial-set-up)
-    - [Run the testbench](#run-the-testbench)
-    - [Start the gRPC server](#start-the-gRPC-server)
-    - [Check that the testbench is running](#check-that-the-testbench-is-running)
-  - [Updating Proto Files](#updating-proto-files)
-  - [Force Failures](#force-failures)
-    - [return-broken-stream](#return-broken-stream)
-    - [return-corrupted-data](#return-corrupted-data)
-    - [stall-always](#stall-always)
-    - [stall-at-256KiB](#stall-at-256kib)
-    - [return-503-after-256K](#return-503-after-256k)
-    - [return-503-after-256K/retry-N](#return-503-after-256kretry-n)
-  - [Retry Test API](#retry-test-api)
-    - [Creating a new Retry Test](#creating-a-new-retry-test)
-    - [Get a Retry Test resource](#get-a-retry-test-resource)
-    - [Delete a Retry Test resource](#delete-a-retry-test-resource)
-    - [Causing a failure using x-retry-test-id header](#causing-a-failure-using-x-retry-test-id-header)
-    - [Forced Failures Supported](#forced-failures-supported)
-
-## Issue Policy
-
-Repository provides no dedicated support for issues filed.
-Issues will be addressed when time permits.
-
-## What is this testbench?
-
-This testbench fakes the Google Cloud Storage (GCS) API. You can configure the GCS client libraries to make calls to this fake rather than to the actual API.
-* The testbench fakes the JSON API, both over REST and gRPC. It has limited support for the XML API.
-* Generally, the error codes are similar to the ones generated by GCS, but the error messages are not.
-* The testbench performs far fewer error checks, and no permission checks (ACL/IAM).
-
-## When to use this testbench
-
-In general, this testbench is best suited for integration tests that are hard (or just annoying) to reliably run against production. The primary example of this are errors that make the client library go through its retry path.
-
-This testbench can be useful to test HMAC keys, which are really hard to test against production due to quota restrictions.
-
-It is useful as well to test features that are not yet deployed to production: you can implement them in the testbench and then write the library code before production is "ready".
-
-## How to use this testbench
-
-### Initial set up
-
-1. [Set up python if you haven't already](https://cloud.google.com/python/docs/setup)
-2. [Clone this repository](https://docs.github.com/en/github/creating-cloning-and-archiving-repositories/cloning-a-repository-from-github/cloning-a-repository#cloning-a-repository)
-
-   From the terminal:
-   ```bash
-   git clone https://github.com/googleapis/storage-testbench.git
-   ```
-3. Switch to the cloned directory:
-   ```bash
-   cd storage-testbench
-    ```
-4. [Create a virtual environment](https://cloud.google.com/python/docs/setup#installing_and_using_virtualenv)
-    * keep this virtual environment active whenever you run the testbench
-5. Install dependencies:
-    ```bash
-    pip install -e .
-    ```
-
-### Run the testbench
-
-To start the testbench, run this command from a terminal:
-
- On Non-Windows
-```bash
-python3 testbench_run.py localhost 9000 10
-```
-
-On Windows
-```bash
-py testbench_run.py localhost 9000 10
-```
-
-> ⚠️ Ensure that the virtual environment you created to install the dependencies is active.
-
-
-### Start the gRPC server
-
-If you want to test the gRPC API, you must make a request which will start
-the testbench's gRPC server. Also, the gRPC server must run on a different port
-than the HTTP server. To serve gRPC requests on port 8888, run:
-
-```bash
-curl -s --retry 5 --retry-max-time 40 "http://localhost:9000/start_grpc?port=8888"
-```
-
-### Check that the testbench is running
-
-Ensure the testbench is running by sending it a request from a different terminal, such as:
-
-```bash
-curl -X GET localhost:9000
-```
-
-The response you get should be: `OK`
-
-Now you can use the testbench (while it's running) with the client libraries.
-
-## Updating Proto Files
-
-From time to time you may need to update the files generated by protobuf and
-gRPC. To do so, clone the [protos](https://github.com/googleapis/googleapis) and
-run the grpc_tools generator:
-
-```shell
-cd $HOME/storage-testbench
-
-# This creates a new directory with the protos from `googleapis`.  If
-# the clone already exists use:
-#    git -C $HOME/googleapis pull
-git -C $HOME clone https://github.com/googleapis/googleapis
-
-pip install grpcio-tools
-python -m grpc_tools.protoc -I$HOME/googleapis \
-    --python_out=. --grpc_python_out=. \
-    $HOME/googleapis/google/iam/v1/iam_policy.proto
-python -m grpc_tools.protoc -I$HOME/googleapis \
-    --python_out=. --grpc_python_out=. \
-    $HOME/googleapis/google/iam/v1/options.proto
-python -m grpc_tools.protoc -I$HOME/googleapis \
-    --python_out=. --grpc_python_out=. \
-    $HOME/googleapis/google/iam/v1/policy.proto
-python -m grpc_tools.protoc -I$HOME/googleapis \
-    --python_out=. --grpc_python_out=. \
-    $HOME/googleapis/google/storage/v2/storage.proto
-```
-
-Then commit the files generated in `google/**`:
-
-```shell
-git commit -m"chore: update protos" google
-```
-
-## Force Failures
-
-You can force the following failures by using the `x-goog-emulator-instructions` header.
-The `x-goog-testbench-instructions` header is deprecated, but supported for
-backwards compatibility and provides the same functionality as
-`x-goog-emulator-instructions`, please change your code to use `x-goog-emulator-instructions` instead.
-
-### return-broken-stream
-
-Set request headers with `x-goog-emulator-instructions: return-broken-stream`.
-Testbench will fail after sending 1024*1024 bytes.
-
-### return-corrupted-data
-
-Set request headers with `x-goog-emulator-instructions: return-corrupted-data`.
-Testbench will return corrupted data.
-
-### stall-always
-
-Set request headers with `x-goog-emulator-instructions: stall-always`.
-Testbench will stall at the beginning.
-
-### stall-at-256KiB
-
-Set request headers with `x-goog-emulator-instructions: stall-at-256KiB`.
-Testbench will stall at 256KiB bytes.
-
-### return-503-after-256K
-
-Set request headers with `x-goog-emulator-instructions: return-503-after-256K`.
-Testbench will return a `HTTP 503` after sending 256KiB bytes.
-
-### return-503-after-256K/retry-N
-
-Set request headers with `x-goog-emulator-instructions: return-503-after-256K/retry-1` up to `x-goog-emulator-instructions: return-503-after-256K/retry-N`.
-
-For N==1 and N==2 behave like `return-305-after-256K`, for `N>=3` ignore the
-failure instruction and return successfully. This is used to test failures during
-retry, the client cooperates by sending the retry counter in the failure
-instructions.
-
-
-## Retry Test API
-
-The "Retry Test API" offers a mechanism to describe more complex retry scenarios
-while sending a single, constant header through all the HTTP requests from a
-test program. Retry Test provides accounting of failures used to validate
-the expected failures were experienced by the testbench and not accidentally missed.
-
-Previous versions of the GCS testbench used a custom header in the RPC to
-control the behavior of each RPC, for some test scenarios this required sending
-different header with the first retry attempt vs. subsequent attempts. Producing
-different headers in each attempt is not easy to implement with some client libraries.
-
-Sending a constant header with all RPCs can be implemented across all client libraries,
-and to some degree decouples the test setup from the test execution.
-
-### Creating a new Retry Test
-
-The following cURL request will create a Retry Test resource which emits a 503
-when a buckets list operation is received by the testbench with the returned
-retry test ID.
-
-```bash
-curl -X POST "http://localhost:9000/retry_test" -H 'Content-Type: application/json' \
-     -d '{"instructions":{"storage.buckets.list": ["return-503"]}}'
-```
-
-### Get a Retry Test resource
-
-Get Retry Test resource by id "1d05c20627844214a9ff7cbcf696317d".
-
-```bash
-curl -X GET "http://localhost:9000/retry_test/1d05c20627844214a9ff7cbcf696317d"
-```
-
-### Delete a Retry Test resource
-
-Delete Retry Test resource by id "1d05c20627844214a9ff7cbcf696317d".
-
-```bash
-curl -X DELETE "http://localhost:9000/retry_test/1d05c20627844214a9ff7cbcf696317d"
-```
-
-### Causing a failure using x-retry-test-id header
-
-The following cURL request will attempt to list buckets and the testbench will emit
-a `503` error once based on the Retry Test created above. Subsequent list buckets
-operations will succeed.
-
-```bash
-curl -H "x-retry-test-id: 1d05c20627844214a9ff7cbcf696317d" "http://localhost:9100/storage/v1/b?project=test"
-```
-
-### Forced Failures Supported
-
-| Failure Id              | Description
-| ----------------------- | ---
-| return-X                                  | Testbench will fail with HTTP code provided for `X`, e.g. return-503 returns a 503
-| return-X-after-YK                         | Testbench will return X after YKiB of uploaded data
-| return-broken-stream-final-chunk-after-YB | Testbench will break connection on final chunk of a resumable upload after Y bytes
-| return-broken-stream                      | Testbench will fail after a few bytes
-| return-broken-stream-after-YK             | Testbench will fail after YKiB of downloaded data
-| return-reset-connection                   | Testbench will fail with a reset connection
diff --git a/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/top_level.txt b/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/top_level.txt
deleted file mode 100644
index 01d339b0..00000000
--- a/pip-wheel-metadata/googleapis_storage_testbench-0.33.0.dist-info/top_level.txt
+++ /dev/null
@@ -1,5 +0,0 @@
-gcs
-google/iam/v1
-google/storage/v2
-testbench
-testbench/servers
diff --git a/testbench/rest_server.py b/testbench/rest_server.py
index 16a496f1..b89e72e0 100644
--- a/testbench/rest_server.py
+++ b/testbench/rest_server.py
@@ -1009,7 +1009,6 @@ def resumable_upload_chunk(bucket_name):
         if chunk_last_byte == "*":
             x_upload_content_length = len(upload.media)
             chunk_last_byte = len(upload.media) - 1
-            # chunk_last_byte = (int(chunk_first_byte) + len(data) - 1) if chunk_first_byte != "*" else last_byte_persisted + len(data)
         else:
             chunk_last_byte = int(chunk_last_byte)
         total_object_size = (

From 02ab95a20f7dffaefad32475eadc11fea551dc1c Mon Sep 17 00:00:00 2001
From: Cathy Ouyang <cathyo@google.com>
Date: Mon, 10 Jul 2023 16:21:48 -0700
Subject: [PATCH 12/14] update tests and coverage

---
 testbench/common.py           |  4 +--
 tests/test_testbench_retry.py | 58 +++++++++++++++++++++++++++++++----
 2 files changed, 53 insertions(+), 9 deletions(-)

diff --git a/testbench/common.py b/testbench/common.py
index a2a2b1a1..ab5fd72c 100644
--- a/testbench/common.py
+++ b/testbench/common.py
@@ -859,10 +859,8 @@ def handle_retry_uploads_error_after_bytes(
     if after_bytes > last_byte_persisted and after_bytes <= (chunk_last_byte + 1):
         range_start = 0
         # Ignore request bytes that have already been persisted.
-        if int(chunk_first_byte) < last_byte_persisted:
+        if last_byte_persisted != 0 and int(chunk_first_byte) <= last_byte_persisted:
             range_start = last_byte_persisted - int(chunk_first_byte) + 1
-        elif int(chunk_first_byte) == last_byte_persisted and last_byte_persisted != 0:
-            range_start = int(chunk_first_byte) + 1
         range_end = len(data)
         # Only partial data will be commited due to the instructed interruption.
         if after_bytes <= chunk_last_byte:
diff --git a/tests/test_testbench_retry.py b/tests/test_testbench_retry.py
index d84a96f7..0a6ff7bb 100644
--- a/tests/test_testbench_retry.py
+++ b/tests/test_testbench_retry.py
@@ -439,11 +439,20 @@ def test_retry_test_return_error_after_bytes(self):
         )
         self.assertEqual(response.status_code, 200)
 
-        # Setup a failure for uploading an object.
+        # Setup two after-bytes errors to test injecting failures in
+        # resumable uploads, both multiple chunks and a single chunk.
+        error_after_300K = 300 * 1024
         response = self.client.post(
             "/retry_test",
             data=json.dumps(
-                {"instructions": {"storage.objects.insert": ["return-504-after-256K"]}}
+                {
+                    "instructions": {
+                        "storage.objects.insert": [
+                            "return-504-after-256K",
+                            "return-504-after-300K",
+                        ]
+                    }
+                }
             ),
         )
         self.assertEqual(response.status_code, 200)
@@ -466,6 +475,7 @@ def test_retry_test_return_error_after_bytes(self):
         self.assertIsNotNone(match, msg=location)
         upload_id = match.group(1)
 
+        # Upload the first 256KiB chunk of data and trigger error.
         chunk = self._create_block(UPLOAD_QUANTUM)
         self.assertEqual(len(chunk), UPLOAD_QUANTUM)
 
@@ -497,20 +507,56 @@ def test_retry_test_return_error_after_bytes(self):
             response.headers.get("range"), "bytes=0-%d" % (UPLOAD_QUANTUM - 1)
         )
 
+        # Send a full object upload here to verify testbench can
+        # (1) trigger error_after_bytes instructions,
+        # (2) ignore duplicate request bytes and
+        # (3) return a forced failure with partial data.
+        chunk = self._create_block(2 * UPLOAD_QUANTUM)
+        self.assertEqual(len(chunk), 2 * UPLOAD_QUANTUM)
+        response = self.client.put(
+            "/upload/storage/v1/b/bucket-name/o",
+            query_string={"upload_id": upload_id},
+            headers={
+                "content-range": "bytes 0-{len:d}/{obj_size:d}".format(
+                    len=2 * UPLOAD_QUANTUM - 1, obj_size=2 * UPLOAD_QUANTUM
+                ),
+                "x-retry-test-id": id,
+            },
+            data=chunk,
+        )
+        self.assertEqual(response.status_code, 504, msg=response.data)
+
+        # Check the status of a resumable upload.
         response = self.client.put(
             "/upload/storage/v1/b/bucket-name/o",
             query_string={"upload_id": upload_id},
             headers={
-                "content-range": "bytes {beg:d}-{end:d}/{obj_size:d}".format(
-                    beg=UPLOAD_QUANTUM,
-                    end=2 * UPLOAD_QUANTUM - 1,
-                    obj_size=2 * UPLOAD_QUANTUM,
+                "content-range": "bytes */*",
+                "x-retry-test-id": id,
+            },
+        )
+        self.assertEqual(response.status_code, 308, msg=response.data)
+        self.assertIn("range", response.headers)
+        self.assertEqual(
+            response.headers.get("range"), "bytes=0-%d" % (error_after_300K - 1)
+        )
+
+        # Finally to complete the upload, resend a full object upload again.
+        response = self.client.put(
+            "/upload/storage/v1/b/bucket-name/o",
+            query_string={"upload_id": upload_id},
+            headers={
+                "content-range": "bytes 0-{len:d}/{obj_size:d}".format(
+                    len=2 * UPLOAD_QUANTUM - 1, obj_size=2 * UPLOAD_QUANTUM
                 ),
                 "x-retry-test-id": id,
             },
             data=chunk,
         )
         self.assertEqual(response.status_code, 200, msg=response.data)
+        create_rest = json.loads(response.data)
+        self.assertIn("size", create_rest)
+        self.assertEqual(int(create_rest.get("size")), 2 * UPLOAD_QUANTUM)
 
 
 if __name__ == "__main__":

From 9d803736ba4967fb6d16f3305530572532ff7e12 Mon Sep 17 00:00:00 2001
From: Cathy Ouyang <cathyo@google.com>
Date: Thu, 13 Jul 2023 12:39:53 -0700
Subject: [PATCH 13/14] update chunk last byte in PUT

---
 testbench/rest_server.py      | 27 ++++++++++++++++-----------
 tests/test_testbench_retry.py |  2 +-
 2 files changed, 17 insertions(+), 12 deletions(-)

diff --git a/testbench/rest_server.py b/testbench/rest_server.py
index b89e72e0..0d6856b2 100644
--- a/testbench/rest_server.py
+++ b/testbench/rest_server.py
@@ -1007,8 +1007,14 @@ def resumable_upload_chunk(bucket_name):
             upload.request.headers.get("x-upload-content-length", 0)
         )
         if chunk_last_byte == "*":
-            x_upload_content_length = len(upload.media)
-            chunk_last_byte = len(upload.media) - 1
+            x_upload_content_length = (
+                len(data) if not x_upload_content_length else x_upload_content_length
+            )
+            chunk_last_byte = (
+                len(data) - 1
+                if chunk_first_byte == "*"
+                else int(chunk_first_byte) + len(data) - 1
+            )
         else:
             chunk_last_byte = int(chunk_last_byte)
         total_object_size = (
@@ -1052,16 +1058,15 @@ def resumable_upload_chunk(bucket_name):
         # Thus we validate chunk_first_byte against last_byte_persisted.
         range_start = 0
         if chunk_first_byte != "*":
-            if int(chunk_first_byte) < last_byte_persisted:
-                range_start = last_byte_persisted - int(chunk_first_byte) + 1
-            elif (
-                int(chunk_first_byte) == last_byte_persisted
-                and last_byte_persisted != 0
+            if (
+                last_byte_persisted != 0
+                and int(chunk_first_byte) <= last_byte_persisted
             ):
-                range_start = int(chunk_first_byte) + 1
-        data = testbench.common.partial_media(
-            data, range_end=(chunk_last_byte + 1), range_start=range_start
-        )
+                range_start = last_byte_persisted - int(chunk_first_byte) + 1
+        if range_start:
+            data = testbench.common.partial_media(
+                data, range_end=(chunk_last_byte + 1), range_start=range_start
+            )
         upload.media += data
         upload.complete = total_object_size == len(upload.media) or (
             chunk_last_byte + 1 == total_object_size
diff --git a/tests/test_testbench_retry.py b/tests/test_testbench_retry.py
index 0a6ff7bb..b3530645 100644
--- a/tests/test_testbench_retry.py
+++ b/tests/test_testbench_retry.py
@@ -324,7 +324,7 @@ def test_retry_test_return_no_metadata_on_resumable_multi_chunk_complete(self):
         response = self.client.put(
             location,
             headers={
-                "content-range": "bytes {last:d}-*/*".format(last=len(chunk) - 1),
+                "content-range": "bytes {last:d}-*/*".format(last=len(chunk)),
                 "x-retry-test-id": id,
             },
             data=chunk,

From e8c4ef1c3ea06d56f7009208a8b451b7f43dd69e Mon Sep 17 00:00:00 2001
From: Cathy Ouyang <cathyo@google.com>
Date: Thu, 13 Jul 2023 14:15:51 -0700
Subject: [PATCH 14/14] only handle after-bytes testbench instructions

---
 testbench/rest_server.py              |  2 +-
 tests/test_testbench_object_upload.py | 63 +++++++++++++++++++++++++++
 2 files changed, 64 insertions(+), 1 deletion(-)

diff --git a/testbench/rest_server.py b/testbench/rest_server.py
index 0d6856b2..166ccbd8 100644
--- a/testbench/rest_server.py
+++ b/testbench/rest_server.py
@@ -1038,7 +1038,7 @@ def resumable_upload_chunk(bucket_name):
             after_bytes,
             test_id,
         ) = testbench.common.get_retry_uploads_error_after_bytes(db, request)
-        if instruction or error_code:
+        if error_code or instruction == "return-503-after-256K":
             if instruction == "return-503-after-256K":
                 error_code = 503
                 after_bytes = 262144
diff --git a/tests/test_testbench_object_upload.py b/tests/test_testbench_object_upload.py
index 87ab6eaf..267b1d7b 100644
--- a/tests/test_testbench_object_upload.py
+++ b/tests/test_testbench_object_upload.py
@@ -617,6 +617,69 @@ def test_upload_pre_conditions_failure(self):
             )
             self.assertEqual(response.status_code, 412, msg=name)
 
+    def test_upload_resumable_w_fault_injection(self):
+        # Test fault injection "return-503-after-256K"
+        media = self._create_valid_chunk()
+        response = self.client.post(
+            "/upload/storage/v1/b/bucket-name/o",
+            query_string={"uploadType": "resumable", "name": "fox"},
+            content_type="application/json",
+            headers={
+                "x-goog-testbench-instructions": "return-503-after-256K",
+            },
+        )
+        self.assertEqual(response.status_code, 200)
+        location = response.headers.get("location")
+        self.assertIn("upload_id=", location)
+        match = re.search("[&?]upload_id=([^&]+)", location)
+        self.assertIsNotNone(match, msg=location)
+        upload_id = match.group(1)
+
+        response = self.client.put(
+            "/upload/storage/v1/b/bucket-name/o",
+            query_string={"upload_id": upload_id},
+            data=media,
+            headers={
+                "content-range": "bytes 0-{last:d}/{object_size:d}".format(
+                    last=UPLOAD_QUANTUM - 1, object_size=UPLOAD_QUANTUM
+                ),
+                "x-goog-testbench-instructions": "return-503-after-256K",
+            },
+        )
+        self.assertEqual(response.status_code, 503)
+
+        # Test fault injection "inject-upload-data-error"
+        response = self.client.post(
+            "/upload/storage/v1/b/bucket-name/o",
+            query_string={"uploadType": "resumable", "name": "zebra"},
+            content_type="application/json",
+            headers={
+                "x-goog-testbench-instructions": "inject-upload-data-error",
+            },
+        )
+        self.assertEqual(response.status_code, 200)
+        location = response.headers.get("location")
+        self.assertIn("upload_id=", location)
+        match = re.search("[&?]upload_id=([^&]+)", location)
+        self.assertIsNotNone(match, msg=location)
+        upload_id = match.group(1)
+
+        response = self.client.put(
+            "/upload/storage/v1/b/bucket-name/o",
+            query_string={"upload_id": upload_id},
+            data=media,
+            headers={
+                "x-goog-testbench-instructions": "inject-upload-data-error",
+            },
+        )
+        self.assertEqual(response.status_code, 200)
+
+        response = self.client.get(
+            "/download/storage/v1/b/bucket-name/o/zebra", query_string={"alt": "media"}
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertNotEqual(response.data.decode("utf-8"), media)
+
 
 if __name__ == "__main__":
     unittest.main()