From e75c573f7d501227e0a625f5ee7c5de22b3e45c2 Mon Sep 17 00:00:00 2001 From: Andrei Aaron Date: Mon, 12 Oct 2026 10:18:14 +0000 Subject: [PATCH] fix: parse list-object bool query flags as true/false strings - Clients such as the Go storage HTTP client send versions=false and includeTrailingDelimiter=false as nonempty query values on Objects.List - Parsing with type=bool (or bare truthiness) treats any nonempty string as true, so "false" incorrectly enabled those modes - That made live-only lists return every generation, and trailing-delimiter placeholder objects appear in items when they should stay in prefixes - Parse versions, includeTrailingDelimiter, and softDeleted as enabled only when the value is the string "true"; add unit coverage for both resolves: https://github.com/googleapis/storage-testbench/issues/829 Signed-off-by: Andrei Aaron --- testbench/database.py | 10 ++- tests/test_database.py | 150 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 157 insertions(+), 3 deletions(-) diff --git a/testbench/database.py b/testbench/database.py index de345961..53bc6508 100644 --- a/testbench/database.py +++ b/testbench/database.py @@ -253,12 +253,16 @@ def __extract_list_object_request(cls, request, context): return cls.__extract_list_object_request_grpc(request) delimiter = request.args.get("delimiter", "") prefix = request.args.get("prefix", "") - versions = request.args.get("versions", False, type=bool) + # Do not use type=bool: bool("false") is True in Python, so the Go client's + # versions=false (and similar query flags) were treated as enabled. + versions = str(request.args.get("versions", "false")).lower() == "true" start_offset = request.args.get("startOffset", "") end_offset = request.args.get("endOffset") - include_trailing_delimiter = request.args.get("includeTrailingDelimiter", False) + include_trailing_delimiter = ( + str(request.args.get("includeTrailingDelimiter", "false")).lower() == "true" + ) match_glob = request.args.get("matchGlob", None) - soft_deleted = request.args.get("softDeleted", False) + soft_deleted = str(request.args.get("softDeleted", "false")).lower() == "true" return ( delimiter, prefix, diff --git a/tests/test_database.py b/tests/test_database.py index 70927ccf..f86a115a 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -334,6 +334,156 @@ def test_object_crud(self): ) self.assertEqual(set(), {o.name for o in items}) + def test_list_object_versions_false_query_string(self): + # The Go storage client sends versions=false as a nonempty query value. + # Parsing with type=bool treats that as True; only the live generation + # must be returned when versions is explicitly false. + self.bucket.metadata.versioning.enabled = True + bodies = (b"one", b"two", b"three") + inserted = [] + for body in bodies: + request = testbench.common.FakeRequest( + args={"name": "versioned-obj"}, data=body, headers={}, environ={} + ) + blob, _ = gcs.object.Object.init_media(request, self.bucket.metadata) + self.database.insert_object("bucket-name", blob, context=None) + inserted.append((blob.metadata.generation, body)) + + live_only, _ = self.database.list_object( + Request(create_environ(query_string={"versions": "false"})), + "bucket-name", + None, + ) + live_gens = [o.generation for o in live_only if o.name == "versioned-obj"] + self.assertEqual(live_gens, [inserted[-1][0]]) + live = self.database.get_object( + "bucket-name", "versioned-obj", context=None, generation=live_gens[0] + ) + self.assertEqual(live.media, bodies[-1]) + + all_versions, _ = self.database.list_object( + Request(create_environ(query_string={"versions": "true"})), + "bucket-name", + None, + ) + all_gens = [o.generation for o in all_versions if o.name == "versioned-obj"] + self.assertCountEqual(all_gens, [gen for gen, _ in inserted]) + for gen, body in inserted: + blob = self.database.get_object( + "bucket-name", "versioned-obj", context=None, generation=gen + ) + self.assertEqual(blob.media, body) + + def test_list_object_include_trailing_delimiter_false_query_string(self): + # includeTrailingDelimiter=false must not be treated as enabled. + names = ( + "a/a/", + "a/a/test-0", + "a/b/", + "a/b/test-1", + "a/c/test-2", + "a/test-3", + ) + for name in names: + request = testbench.common.FakeRequest( + args={"name": name}, data=b"media", headers={}, environ={} + ) + blob, _ = gcs.object.Object.init_media(request, self.bucket.metadata) + self.database.insert_object("bucket-name", blob, context=None) + + without_trailing, prefixes = self.database.list_object( + Request( + create_environ( + query_string={ + "prefix": "a/", + "delimiter": "/", + "includeTrailingDelimiter": "false", + } + ) + ), + "bucket-name", + None, + ) + self.assertEqual(prefixes, ["a/a/", "a/b/", "a/c/"]) + self.assertEqual([o.name for o in without_trailing], ["a/test-3"]) + + with_trailing, prefixes = self.database.list_object( + Request( + create_environ( + query_string={ + "prefix": "a/", + "delimiter": "/", + "includeTrailingDelimiter": "true", + } + ) + ), + "bucket-name", + None, + ) + self.assertEqual(prefixes, ["a/a/", "a/b/", "a/c/"]) + self.assertEqual([o.name for o in with_trailing], ["a/a/", "a/b/", "a/test-3"]) + + def test_list_object_soft_deleted_false_query_string(self): + # softDeleted=false must not be treated as true (bool("false") is True). + # Without a soft-delete policy that misparse would reject the list call. + request = testbench.common.FakeRequest( + args={"name": "live-obj"}, data=b"live", headers={}, environ={} + ) + blob, _ = gcs.object.Object.init_media(request, self.bucket.metadata) + self.database.insert_object("bucket-name", blob, context=None) + + live_items, _ = self.database.list_object( + Request(create_environ(query_string={"softDeleted": "false"})), + "bucket-name", + None, + ) + self.assertEqual([o.name for o in live_items], ["live-obj"]) + + with self.assertRaises(testbench.error.RestException) as rest: + self.database.list_object( + Request(create_environ(query_string={"softDeleted": "true"})), + "bucket-name", + None, + ) + self.assertEqual(rest.exception.code, 400) + + def test_list_object_soft_deleted_true_query_string(self): + request = testbench.common.FakeRequest( + args={}, + data=json.dumps( + { + "name": "sd-bucket-name", + "softDeletePolicy": {"retentionDurationSeconds": 7 * 24 * 60 * 60}, + } + ), + ) + sd_bucket, _ = gcs.bucket.Bucket.init(request, None) + self.database.insert_bucket(sd_bucket, None) + + request = testbench.common.FakeRequest( + args={"name": "soft-obj"}, data=b"soft", headers={}, environ={} + ) + blob, _ = gcs.object.Object.init_media(request, sd_bucket.metadata) + self.database.insert_object("sd-bucket-name", blob, context=None) + soft_gen = blob.metadata.generation + self.database.delete_object("sd-bucket-name", "soft-obj", context=None) + + after_delete, _ = self.database.list_object( + Request(create_environ(query_string={"softDeleted": "false"})), + "sd-bucket-name", + None, + ) + self.assertEqual([o.name for o in after_delete], []) + + soft_items, _ = self.database.list_object( + Request(create_environ(query_string={"softDeleted": "true"})), + "sd-bucket-name", + None, + ) + soft_only = [o for o in soft_items if o.name == "soft-obj"] + self.assertEqual(len(soft_only), 1) + self.assertEqual(soft_only[0].generation, soft_gen) + def test_get_object_not_found(self): with self.assertRaises(testbench.error.RestException) as rest: _ = self.database.get_object("bucket-name", "object-name", context=None)