From e792b16f61f4a7629615ed5ada1ebfd4c7cf78a2 Mon Sep 17 00:00:00 2001 From: Alistair Coles Date: Thu, 31 Jul 2025 12:58:43 +0100 Subject: [PATCH] mpu: add compat test for listing sessions Change-Id: If2ebfaba902303774cefe3c5038b9c565f764b1d Signed-off-by: Alistair Coles --- test/s3api/test_mpu.py | 38 ++++++++++++++++++++++++++++++++++++-- 1 file changed, 36 insertions(+), 2 deletions(-) diff --git a/test/s3api/test_mpu.py b/test/s3api/test_mpu.py index f4ec903f62..5cd93bb639 100644 --- a/test/s3api/test_mpu.py +++ b/test/s3api/test_mpu.py @@ -186,9 +186,12 @@ class TestMultiPartUpload(BaseMultiPartUploadTestCase): 'ResponseMetadata']['HTTPStatusCode']) return create_mpu_resp['UploadId'] - def list_mpus(self): + def list_mpus(self, key_marker='', upload_id_marker=''): list_mpu_resp = self.client.list_multipart_uploads( - Bucket=self.bucket_name) + Bucket=self.bucket_name, + KeyMarker=key_marker, + UploadIdMarker=upload_id_marker + ) self.assertEqual(200, list_mpu_resp[ 'ResponseMetadata']['HTTPStatusCode']) mpus = list_mpu_resp.get('Uploads', []) @@ -906,6 +909,37 @@ class TestMultiPartUpload(BaseMultiPartUploadTestCase): self.assertEqual(400, status_from_error(cm.exception)) self.assertEqual('InvalidPartOrder', code_from_error(cm.exception)) + def test_create_list_mpus_utf8(self): + key_1 = self.create_name('key\N{SNOWMAN}x') + key_2 = self.create_name('key\N{SNOWMAN}y') + + uploaded = [(key, self.create_mpu(key)) + for key in (key_1, key_2, key_1)] + + found_uploads = self.list_mpus() + # sort by key then created time + expected = [uploaded[i] for i in (0, 2, 1)] + self.assertEqual(expected, found_uploads) + + # use key marker + found_uploads = self.list_mpus(key_marker=key_1) + expected = [uploaded[1]] + self.assertEqual(expected, found_uploads) + + found_uploads = self.list_mpus(key_marker=key_2) + expected = [] + self.assertEqual(expected, found_uploads) + + # use upload_id marker + found_uploads = self.list_mpus(key_marker=key_1, + upload_id_marker=uploaded[0][1]) + expected = [uploaded[i] for i in (2, 1)] + self.assertEqual(expected, found_uploads) + found_uploads = self.list_mpus(key_marker=key_1, + upload_id_marker=uploaded[2][1]) + expected = [uploaded[1]] + self.assertEqual(expected, found_uploads) + def test_create_list_mpus_abort_list_mpus(self): key_name = self.create_name('key') upload_id = self.create_mpu(key_name)