[4.18 system test] Test bucket replication with object versioning

ocs_ci/ocs/bucket_utils.py

@@ -2115,6 +2115,15 @@ def update_replication_policy(bucket_name, replication_policy_dict):
     ).patch(params=json.dumps(replication_policy_patch_dict), format_type="merge")
 
 
+def get_replication_policy(bucket_name):
+
+    return OCP(
+        kind="obc",
+        namespace=config.ENV_DATA["cluster_namespace"],
+        resource_name=bucket_name,
+    ).get()["spec"]["additionalConfig"]["replicationPolicy"]
+
+
 def patch_replication_policy_to_bucketclass(
     bucketclass_name, rule_id, destination_bucket_name
 ):
@@ -2716,6 +2725,42 @@ def bulk_s3_put_bucket_lifecycle_config(mcg_obj, buckets, lifecycle_config):
     logger.info("Applied lifecyle rule on all the buckets")
 
 
+def upload_random_objects_to_source_and_wait_for_replication(
+    mcg_obj,
+    source_bucket,
+    target_bucket,
+    mockup_logger,
+    file_dir,
+    pattern="ObjKey-",
+    amount=1,
+    prefix=None,
+    timeout=600,
+):
+    """
+    Upload randomly generated objects to the source bucket and wait until the
+    replication happens
+
+    """
+
+    logger.info(f"Randomly generating {amount} object/s")
+    obj_list = write_random_objects_in_pod(
+        io_pod=mockup_logger.awscli_pod,
+        file_dir=file_dir,
+        amount=amount,
+        pattern=pattern,
+    )
+
+    mockup_logger.upload_random_objects_and_log(
+        source_bucket.name, file_dir=file_dir, obj_list=obj_list, prefix=prefix
+    )
+    assert compare_bucket_object_list(
+        mcg_obj,
+        source_bucket.name,
+        target_bucket.name,
+        timeout=timeout,
+    ), f"Standard replication failed to complete in {timeout} seconds"
+
+
 def upload_test_objects_to_source_and_wait_for_replication(
     mcg_obj, source_bucket, target_bucket, mockup_logger, timeout
 ):
@@ -2945,7 +2990,9 @@ def put_bucket_versioning_via_awscli(
     )
 
 
-def upload_obj_versions(mcg_obj, awscli_pod, bucket_name, obj_key, amount=1, size="1M"):
+def upload_obj_versions(
+    mcg_obj, awscli_pod, bucket_name, obj_key, prefix=None, amount=1, size="1M"
+):
     """
     Upload multiple random data versions to a given object key and return their ETag values
 
@@ -2960,6 +3007,10 @@ def upload_obj_versions(mcg_obj, awscli_pod, bucket_name, obj_key, amount=1, siz
     Returns:
         list: List of ETag values of versions in latest to oldest order
     """
+    full_object_path = (
+        f"s3://{bucket_name}/{prefix}" if prefix else f"s3://{bucket_name}/"
+    )
+
     file_dir = os.path.join("/tmp", str(uuid4()))
     awscli_pod.exec_cmd_on_pod(f"mkdir {file_dir}")
 
@@ -2974,7 +3025,7 @@ def upload_obj_versions(mcg_obj, awscli_pod, bucket_name, obj_key, amount=1, siz
         # the uploaded object's ETag in the response
         resp = awscli_pod.exec_cmd_on_pod(
             command=craft_s3_command(
-                f"cp {file_path} s3://{bucket_name}/{obj_key} --debug 2>&1",
+                f"cp {file_path} {full_object_path}/{obj_key} --debug 2>&1",
                 mcg_obj=mcg_obj,
             ),
             out_yaml_format=False,
@@ -3030,5 +3081,34 @@ def get_obj_versions(mcg_obj, awscli_pod, bucket_name, obj_key):
         # Remove quotes from the ETag values for easier usage
         for d in versions_dicts:
             d["ETag"] = d["ETag"].strip('"')
-
+    logger.info(f"{bucket_name} version dict: {versions_dicts}")
     return versions_dicts
+
+
+def verify_deletion_marker(mcg_obj, awscli_pod, bucket_name, object_key):
+    """
+    Verify if deletion marker exists for the given object key
+
+    Args:
+        mcg_obj (MCG): MCG object
+        awscli_pod (Pod): Pod object where AWS CLI is installed
+        bucket_name (str): Name of the bucket
+        object_key (str): Object key
+
+    Returns:
+        True if DeletionMarkers exists else False
+
+    """
+    resp = awscli_pod.exec_cmd_on_pod(
+        command=craft_s3_command(
+            f"list-object-versions --bucket {bucket_name} --prefix {object_key}",
+            mcg_obj=mcg_obj,
+            api=True,
+        ),
+        out_yaml_format=False,
+    )
+
+    if resp and "DeleteMarkers" in resp:
+        if json.loads(resp).get("DeleteMarkers")[0].get("IsLatest") == "true":
+            return True
+    return False

ocs_ci/ocs/resources/mcg_replication_policy.py

@@ -38,15 +38,18 @@ def __init__(
         self,
         destination_bucket,
         sync_deletions=False,
+        sync_versions=False,
         prefix="",
     ):
         super().__init__(destination_bucket, prefix)
         self.sync_deletions = sync_deletions
+        self.sync_versions = sync_versions
 
     @abstractmethod
     def to_dict(self):
         dict = super().to_dict()
         dict["rules"][0]["sync_deletions"] = self.sync_deletions
+        dict["rules"][0]["sync_versions"] = self.sync_versions
         dict["log_replication_info"] = {}
 
         return dict
@@ -65,8 +68,9 @@ def __init__(
         logs_bucket="",
         prefix="",
         logs_location_prefix="",
+        sync_versions=False,
     ):
-        super().__init__(destination_bucket, sync_deletions, prefix)
+        super().__init__(destination_bucket, sync_deletions, sync_versions, prefix)
         self.logs_bucket = logs_bucket
         self.logs_location_prefix = logs_location_prefix
 

ocs_ci/ocs/resources/mockup_bucket_logger.py

@@ -101,6 +101,28 @@ def upload_arbitrary_object_and_log(self, bucket_name):
 
         self._upload_mockup_logs(bucket_name, [obj_name], "PUT")
 
+    def upload_random_objects_and_log(
+        self, bucket_name, file_dir, obj_list, prefix=None
+    ):
+        """
+        Uploads randomly generated objects to the bucket and upload a matching
+        mockup log
+
+        """
+
+        logger.info(
+            f"Uploading randomly generated objects from {file_dir} to {bucket_name}"
+        )
+        prefix = prefix if prefix else ""
+        sync_object_directory(
+            self.awscli_pod,
+            file_dir,
+            f"s3://{bucket_name}/{prefix}",
+            self.mcg_obj,
+        )
+
+        self._upload_mockup_logs(bucket_name=bucket_name, obj_list=obj_list, op="PUT")
+
     def delete_objs_and_log(self, bucket_name, objs):
         """
         Delete list of objects from the MCG bucket and write

tests/conftest.py

@@ -46,6 +46,8 @@
 from ocs_ci.ocs.bucket_utils import (
     craft_s3_command,
     put_bucket_policy,
+    update_replication_policy,
+    put_bucket_versioning_via_awscli,
 )
 from ocs_ci.ocs.constants import FUSION_CONF_DIR
 from ocs_ci.ocs.cnv.virtual_machine import VirtualMachine, VMCloner
@@ -8138,6 +8140,21 @@ def factory(interval):
     return factory
 
 
+@pytest.fixture()
+def reduce_replication_delay(add_env_vars_to_noobaa_core_class):
+
+    def factory(interval=1):
+
+        new_delay_in_milliseconfs = interval * 60 * 1000
+        new_env_var_touples = [
+            (constants.BUCKET_REPLICATOR_DELAY_PARAM, new_delay_in_milliseconfs),
+            (constants.BUCKET_LOG_REPLICATOR_DELAY_PARAM, new_delay_in_milliseconfs),
+        ]
+        add_env_vars_to_noobaa_core_class(new_env_var_touples)
+
+    return factory
+
+
 @pytest.fixture()
 def reset_conn_score():
     """
@@ -8465,21 +8482,34 @@ def aws_log_based_replication_setup(
     """
     A fixture to set up standard log-based replication with deletion sync.
 
-    Args:
-        awscli_pod_session(Pod): A pod running the AWS CLI
-        mcg_obj_session(MCG): An MCG object
-        bucket_factory: A bucket factory fixture
-
-    Returns:
-        MockupBucketLogger: A MockupBucketLogger object
-        Bucket: The source bucket
-        Bucket: The target bucket
-
     """
 
     reduce_replication_delay_setup()
 
-    def factory(bucketclass_dict=None):
+    def factory(
+        bucketclass_dict=None,
+        prefix_source="",
+        prefix_target="",
+        bidirectional=False,
+        deletion_sync=True,
+        enable_versioning=False,
+    ):
+        """
+        A fixture to set up standard log-based replication with deletion sync.
+
+        Args:
+            bucketclass_dict (Dict): Dictionary representing bucketclass parameters
+            bidirectional (Bool): True if you want to setup bi-directional replication
+                                  otherwise False
+            deletion_sync (Bool): True if you want to setup deletion sync otherwise False
+
+        Returns:
+            MockupBucketLogger: A MockupBucketLogger object
+            Bucket: The source bucket
+            Bucket: The target bucket
+
+        """
+
         log.info("Starting log-based replication setup")
         if bucketclass_dict is None:
             bucketclass_dict = {
@@ -8492,27 +8522,60 @@ def factory(bucketclass_dict=None):
                 },
             }
         target_bucket = bucket_factory(bucketclass=bucketclass_dict)[0]
+        if enable_versioning:
+            put_bucket_versioning_via_awscli(
+                mcg_obj_session, awscli_pod_session, target_bucket.name
+            )
 
-        mockup_logger = MockupBucketLogger(
+        mockup_logger_source = MockupBucketLogger(
             awscli_pod=awscli_pod_session,
             mcg_obj=mcg_obj_session,
             bucket_factory=bucket_factory,
             platform=constants.AWS_PLATFORM,
             region=constants.DEFAULT_AWS_REGION,
         )
-        replication_policy = AwsLogBasedReplicationPolicy(
+        replication_policy_source = AwsLogBasedReplicationPolicy(
             destination_bucket=target_bucket.name,
-            sync_deletions=True,
-            logs_bucket=mockup_logger.logs_bucket_uls_name,
+            sync_deletions=deletion_sync,
+            logs_bucket=mockup_logger_source.logs_bucket_uls_name,
+            prefix=prefix_source,
+            sync_versions=True,
         )
 
         source_bucket = bucket_factory(
-            1, bucketclass=bucketclass_dict, replication_policy=replication_policy
+            1,
+            bucketclass=bucketclass_dict,
+            replication_policy=replication_policy_source,
         )[0]
+        if enable_versioning:
+            put_bucket_versioning_via_awscli(
+                mcg_obj_session, awscli_pod_session, source_bucket.name
+            )
+
+        mockup_logger_target = None
+        if bidirectional:
+            mockup_logger_target = MockupBucketLogger(
+                awscli_pod=awscli_pod_session,
+                mcg_obj=mcg_obj_session,
+                bucket_factory=bucket_factory,
+                platform=constants.AWS_PLATFORM,
+                region=constants.DEFAULT_AWS_REGION,
+            )
+
+            replication_policy_target = AwsLogBasedReplicationPolicy(
+                destination_bucket=source_bucket.name,
+                sync_deletions=deletion_sync,
+                logs_bucket=mockup_logger_target.logs_bucket_uls_name,
+                prefix=prefix_target,
+                sync_versions=True,
+            )
+            update_replication_policy(
+                target_bucket.name, replication_policy_target.to_dict()
+            )
 
         log.info("log-based replication setup complete")
 
-        return mockup_logger, source_bucket, target_bucket
+        return mockup_logger_source, mockup_logger_target, source_bucket, target_bucket
 
     return factory
 

tests/cross_functional/conftest.py

@@ -362,6 +362,15 @@ def factory(noobaa_pvc_obj):
         )
         return restore_pvc_objs, snap_obj
 
+    def teardown():
+        """
+        Teardown code to delete the restore pvc objects
+
+        """
+        for pvc_obj in restore_pvc_objs:
+            pvc_obj.delete()
+
+    request.addfinalizer(teardown)
     return factory