[CEPH-83604857]: Tags added during multipart upload should be retriev…

…able

Signed-off-by: Chaithra <[email protected]>

rgw/v2/tests/s3_swift/configs/test_multipart_upload_with_tagging.yaml

@@ -0,0 +1,23 @@
+# script: test_Mbuckets_with_Nobjects.py
+# upload type: Multipart
+# Polarion ID: CEPH-83604857
+# BZ: 2323604
+config:
+  user_count: 1
+  bucket_count: 1
+  objects_count: 1
+  objects_size_range:
+    min: 1G
+    max: 2G
+  test_ops:
+    create_bucket: true
+    create_object: true
+    download_object: false
+    delete_bucket_object: false
+    multipart_upload_with_tag: true
+    sharding:
+      enable: false
+      max_shards: 0
+    compression:
+      enable: false
+      type: zlib

rgw/v2/tests/s3_swift/reusable.py

@@ -459,6 +459,8 @@ def upload_object_with_tagging(
     obj_tag,
     append_data=False,
     append_msg=None,
+    verify_tag_retrieval=False,
+    s3_client=None,
 ):
     log.info("s3 object name: %s" % s3_object_name)
     s3_object_path = os.path.join(TEST_DATA_PATH, s3_object_name)
@@ -499,6 +501,19 @@ def upload_object_with_tagging(
         raise TestExecError("Resource execution failed: object upload failed")
     if object_uploaded_status is None:
         log.info("object uploaded")
+    if verify_tag_retrieval:
+        log.info("Verify Tag applied is retirievable")
+        log.info(f"object: {s3_object_name}")
+        get_obj_tagging_result = s3lib.resource_op(
+            {
+                "obj": s3_client,
+                "resource": "get_object_tagging",
+                "kwargs": dict(Bucket=bucket.name, Key=s3_object_name),
+            }
+        )
+        log.info(f"get_obj_tagging_result: {get_obj_tagging_result}")
+        if not get_obj_tagging_result.get("TagSet"):
+            raise AssertionError("Tag retrival is failed for multipart objects!")
 
 
 def upload_mutipart_object(

rgw/v2/tests/s3_swift/test_Mbuckets_with_Nobjects.py

@@ -966,7 +966,19 @@ def test_exec(config, ssh_con):
             if final_op != -1:
                 test_info.failed_status("test failed")
                 sys.exit(1)
-
+        if config.test_ops.get("multipart_upload_with_tag", False):
+            log.info("Testing tag retrival post multipart upload")
+            obj_tag = "mpupload=mpupload"
+            reusable.upload_object_with_tagging(
+                s3_object_name,
+                bucket,
+                TEST_DATA_PATH,
+                config,
+                each_user,
+                obj_tag,
+                verify_tag_retrieval=True,
+                s3_client=rgw_conn2,
+            )
     # test async rgw_data_notify_interval_msec=0 does not disable async data notifications
     if config.test_aync_data_notifications:
         log.info("Testing async data notifications")