localstack · bryansan-local · May 14, 2025 · May 12, 2025 · May 13, 2025 · May 13, 2025
diff --git a/localstack-core/localstack/services/s3/provider.py b/localstack-core/localstack/services/s3/provider.py
@@ -581,14 +581,45 @@ def list_buckets(
         bucket_region: BucketRegion = None,
         **kwargs,
     ) -> ListBucketsOutput:
-        # TODO add support for max_buckets, continuation_token, prefix, and bucket_region
         owner = get_owner_for_account_id(context.account_id)
         store = self.get_store(context.account_id, context.region)
-        buckets = [
-            Bucket(Name=bucket.name, CreationDate=bucket.creation_date)
-            for bucket in store.buckets.values()
-        ]
-        return ListBucketsOutput(Owner=owner, Buckets=buckets)
+
+        decoded_continuation_token = (
+            to_str(base64.urlsafe_b64decode(continuation_token.encode()))
+            if continuation_token
+            else None
+        )
+
+        count = 0
+        buckets: list[Bucket] = []
+        next_continuation_token = None
+
+        # Comparing strings with case sensitivity since AWS is case-sensitive
+        for bucket in sorted(store.buckets.values(), key=lambda r: r.name):
+            if continuation_token and bucket.name < decoded_continuation_token:
+                continue
+
+            if prefix and not bucket.name.startswith(prefix):
+                continue
+
+            if bucket_region and not bucket.bucket_region == bucket_region:
+                continue
+
+            if max_buckets and count >= max_buckets:
+                next_continuation_token = to_str(base64.urlsafe_b64encode(bucket.name.encode()))
+                break
+
+            output_bucket = Bucket(
+                Name=bucket.name,
+                CreationDate=bucket.creation_date,
+                BucketRegion=bucket.bucket_region,
+            )
+            buckets.append(output_bucket)
+            count += 1
+
+        return ListBucketsOutput(
+            Owner=owner, Buckets=buckets, Prefix=prefix, ContinuationToken=next_continuation_token
+        )
 
     def head_bucket(
         self,

diff --git a/tests/aws/services/s3/test_s3_list_operations.py b/tests/aws/services/s3/test_s3_list_operations.py
@@ -17,6 +17,7 @@
 from localstack.constants import AWS_REGION_US_EAST_1, LOCALHOST_HOSTNAME
 from localstack.testing.aws.util import is_aws_cloud
 from localstack.testing.pytest import markers
+from localstack.utils.strings import short_uid
 
 
 def _bucket_url(bucket_name: str, region: str = "", localstack_host: str = None) -> str:
@@ -46,6 +47,103 @@ def assert_timestamp_is_iso8061_s3_format(timestamp: str):
     assert parsed_ts.microsecond == 0
 
 
+class TestS3ListBuckets:
+    @markers.aws.validated
+    def test_list_buckets_by_prefix_with_case_sensitivity(
+        self, s3_create_bucket, aws_client, snapshot
+    ):
+        snapshot.add_transformer(snapshot.transform.s3_api())
+
+        bucket_name = f"test-bucket-{short_uid()}"
+        s3_create_bucket(Bucket=bucket_name)
+        s3_create_bucket(Bucket=f"ignored-bucket-{short_uid()}")
+
+        response = aws_client.s3.list_buckets(Prefix=bucket_name.upper())
+        assert len(response["Buckets"]) == 0
+
+        response = aws_client.s3.list_buckets(Prefix=bucket_name)
+        assert len(response["Buckets"]) == 1
+
+        returned_bucket = response["Buckets"][0]
+        assert returned_bucket["Name"] == bucket_name
+
+        snapshot.match("list-objects-by-prefix", response)
+
+    @markers.aws.validated
+    def test_list_buckets_with_max_buckets(self, s3_create_bucket, aws_client, snapshot):
+        snapshot.add_transformer(snapshot.transform.s3_api())
+        snapshot.add_transformer(snapshot.transform.key_value("ContinuationToken"))
+
+        s3_create_bucket()
+        s3_create_bucket()
+
+        response = aws_client.s3.list_buckets(MaxBuckets=1)
+        assert len(response["Buckets"]) == 1
+
+        snapshot.match("list-objects-with-max-buckets", response)
+
+    @markers.aws.validated
+    def test_list_buckets_when_continuation_token_is_empty(
+        self, s3_create_bucket, aws_client, snapshot
+    ):
+        snapshot.add_transformer(snapshot.transform.s3_api())
+        snapshot.add_transformer(snapshot.transform.key_value("ContinuationToken"))
+
+        s3_create_bucket()
+        s3_create_bucket()
+
+        response = aws_client.s3.list_buckets(ContinuationToken="", MaxBuckets=1)
+        assert len(response["Buckets"]) == 1
+
+        snapshot.match("list-objects-with-empty-continuation-token", response)
+
+    @markers.aws.validated
+    def test_list_buckets_by_bucket_region(
+        self, s3_create_bucket, s3_create_bucket_with_client, aws_client_factory, snapshot
+    ):
+        region_us_west_2 = "us-west-2"
+        snapshot.add_transformer(snapshot.transform.s3_api())
+        snapshot.add_transformer(snapshot.transform.key_value("ContinuationToken"))
+        snapshot.add_transformer(snapshot.transform.regex(region_us_west_2, "<region>"))
+
+        s3_create_bucket()
+
+        client_us_west_2 = aws_client_factory(region_name="us-west-2").s3
+        s3_create_bucket_with_client(
+            client_us_west_2,
+            Bucket=f"test-bucket-{short_uid()}",
+            CreateBucketConfiguration={"LocationConstraint": region_us_west_2},
+        )
+
+        response = client_us_west_2.list_buckets(BucketRegion=region_us_west_2)
+
+        returned_bucket = response["Buckets"][0]
+        assert returned_bucket["BucketRegion"] == region_us_west_2
+
+        snapshot.match("list-objects-by-bucket-region", response)
+
+    @markers.aws.validated
+    def test_list_buckets_with_continuation_token(self, s3_create_bucket, aws_client, snapshot):
+        snapshot.add_transformer(snapshot.transform.s3_api())
+        snapshot.add_transformer(snapshot.transform.key_value("ContinuationToken"))
+
+        s3_create_bucket()
+        s3_create_bucket()
+        s3_create_bucket()
+        response = aws_client.s3.list_buckets(MaxBuckets=1)
+        assert "ContinuationToken" in response
+
+        first_returned_bucket = response["Buckets"][0]
+        continuation_token = response["ContinuationToken"]
+
+        response = aws_client.s3.list_buckets(MaxBuckets=1, ContinuationToken=continuation_token)
+
+        continuation_returned_bucket = response["Buckets"][0]
+        assert first_returned_bucket["Name"] != continuation_returned_bucket["Name"]
+
+        snapshot.match("list-objects-with-continuation", response)
+
+
 class TestS3ListObjects:
     @markers.aws.validated
     @pytest.mark.parametrize("delimiter", ["", "/", "%2F"])

diff --git a/tests/aws/services/s3/test_s3_list_operations.snapshot.json b/tests/aws/services/s3/test_s3_list_operations.snapshot.json
@@ -3085,5 +3085,119 @@
         }
       }
     }
+  },
+  "tests/aws/services/s3/test_s3_list_operations.py::TestS3ListBuckets::test_list_buckets_with_max_buckets": {
+    "recorded-date": "13-05-2025, 14:36:23",
+    "recorded-content": {
+      "list-objects-with-max-buckets": {
+        "Buckets": [
+          {
+            "BucketRegion": "<region>",
+            "CreationDate": "datetime",
+            "Name": "<bucket-name:1>"
+          }
+        ],
+        "ContinuationToken": "<continuation-token:1>",
+        "Owner": {
+          "DisplayName": "<display-name>",
+          "ID": "<owner-id>"
+        },
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      }
+    }
+  },
+  "tests/aws/services/s3/test_s3_list_operations.py::TestS3ListBuckets::test_list_buckets_by_bucket_region": {
+    "recorded-date": "13-05-2025, 14:36:27",
+    "recorded-content": {
+      "list-objects-by-bucket-region": {
+        "Buckets": [
+          {
+            "BucketRegion": "<region>",
+            "CreationDate": "datetime",
+            "Name": "<bucket-name:1>"
+          }
+        ],
+        "Owner": {
+          "DisplayName": "<display-name>",
+          "ID": "<owner-id>"
+        },
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      }
+    }
+  },
+  "tests/aws/services/s3/test_s3_list_operations.py::TestS3ListBuckets::test_list_buckets_with_continuation_token": {
+    "recorded-date": "13-05-2025, 14:46:32",
+    "recorded-content": {
+      "list-objects-with-continuation": {
+        "Buckets": [
+          {
+            "BucketRegion": "<region>",
+            "CreationDate": "datetime",
+            "Name": "<bucket-name:1>"
+          }
+        ],
+        "ContinuationToken": "<continuation-token:1>",
+        "Owner": {
+          "DisplayName": "<display-name>",
+          "ID": "<owner-id>"
+        },
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      }
+    }
+  },
+  "tests/aws/services/s3/test_s3_list_operations.py::TestS3ListBuckets::test_list_buckets_by_prefix_with_case_sensitivity": {
+    "recorded-date": "13-05-2025, 14:36:20",
+    "recorded-content": {
+      "list-objects-by-prefix": {
+        "Buckets": [
+          {
+            "BucketRegion": "<region>",
+            "CreationDate": "datetime",
+            "Name": "<bucket-name:1>"
+          }
+        ],
+        "Owner": {
+          "DisplayName": "<display-name>",
+          "ID": "<owner-id>"
+        },
+        "Prefix": "<bucket-name:1>",
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      }
+    }
+  },
+  "tests/aws/services/s3/test_s3_list_operations.py::TestS3ListBuckets::test_list_buckets_when_continuation_token_is_empty": {
+    "recorded-date": "13-05-2025, 15:06:12",
+    "recorded-content": {
+      "list-objects-with-empty-continuation-token": {
+        "Buckets": [
+          {
+            "BucketRegion": "<region>",
+            "CreationDate": "datetime",
+            "Name": "<bucket-name:1>"
+          }
+        ],
+        "ContinuationToken": "<continuation-token:1>",
+        "Owner": {
+          "DisplayName": "<display-name>",
+          "ID": "<owner-id>"
+        },
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      }
+    }
   }
 }
@@ -1,4 +1,19 @@
 {
+  "tests/aws/services/s3/test_s3_list_operations.py::TestS3ListBuckets::test_list_buckets_by_bucket_region": {
+    "last_validated_date": "2025-05-13T14:37:07+00:00"
+  },
+  "tests/aws/services/s3/test_s3_list_operations.py::TestS3ListBuckets::test_list_buckets_by_prefix_with_case_sensitivity": {
+    "last_validated_date": "2025-05-13T14:37:02+00:00"
+  },
+  "tests/aws/services/s3/test_s3_list_operations.py::TestS3ListBuckets::test_list_buckets_when_continuation_token_is_empty": {
+    "last_validated_date": "2025-05-13T15:06:47+00:00"
+  },
+  "tests/aws/services/s3/test_s3_list_operations.py::TestS3ListBuckets::test_list_buckets_with_continuation_token": {
+    "last_validated_date": "2025-05-13T14:46:44+00:00"
+  },
+  "tests/aws/services/s3/test_s3_list_operations.py::TestS3ListBuckets::test_list_buckets_with_max_buckets": {
+    "last_validated_date": "2025-05-13T14:37:05+00:00"
+  },
   "tests/aws/services/s3/test_s3_list_operations.py::TestS3ListMultipartUploads::test_list_multipart_uploads_marker_common_prefixes": {
     "last_validated_date": "2025-01-21T18:15:14+00:00"
   },