red-hat-storage · mashetty330 · Jan 6, 2025 · Jan 27, 2025 · Mar 17, 2025 · Mar 18, 2025
@@ -2143,6 +2143,24 @@ 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):
+    """
+    Get the replication policy on a bucket
+
+    Args:
+        bucket_name (str): Name of the bucket
+
+    Returns:
+        Dict: replication policy
+
+    """
+    return OCP(
+        kind="obc",
+        namespace=config.ENV_DATA["cluster_namespace"],
+        resource_name=bucket_name
8000
,
+    ).get()["spec"]["additionalConfig"]["replicationPolicy"]
+
+
 def patch_replication_policy_to_bucketclass(
     bucketclass_name, rule_id, destination_bucket_name
 ):
@@ -2855,6 +2873,56 @@ 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,
+    num_versions=1,
+    prefix=None,
+    timeout=600,
+):
+    """
+    Upload randomly generated objects to the source bucket and wait until the
+    replication happens
+
+    Args:
+        mcg_obj (MCG): MCG object
+        source_bucket (OBC): OBC object
+        target_bucket (OBC): OBC object
+        mockup_logger (MockupLogger): MockupLogger object
+        file_dir (str): File directory where to generate objects
+        pattern (str): Prefix for object name
+        amount (int): Number of objects
+        num_verions (int): Number of versions of each object
+        prefix (str): Prefix under bucket where objects need to be uploaded
+        timeout (int): Timeout to wait until the replication
+
+    """
+
+    logger.info(f"Randomly generating {amount} object/s")
+    for _ in range(num_versions):
+        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
 ):
@@ -3147,7 +3215,6 @@ 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('"')
-
     return versions_dicts
 
 
@@ -3356,3 +3423,34 @@ def delete_all_objects_in_batches(
         batch_deleter.delete_in_parallel()
     else:
         batch_deleter.delete_sequentially()
+
+
+def verify_soft_deletion(mcg_obj, awscli_pod, bucket_name, object_key):
+    """
+    Verify if deletion marker exists and IsLatest 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:
+        delete_markers = json.loads(resp).get("DeleteMarkers")[0]
+        logger.info(f"{bucket_name}:\n{delete_markers}")
+        if delete_markers.get("IsLatest"):
+            return True
+    return False
@@ -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
 

@@ -101,6 +101,34 @@ 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
+
+        Args:
+            bucket_name (str): Name of the bucket
+            file_dir (str): File directory where the objects are present
+            obj_list (list): List of the objects
+            prefix (str): Prefix under which object needs to be uploaded
+
+        """
+
+        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, prefix=""):
         """
         Delete list of objects from the MCG bucket and write

@@ -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.cnv.virtual_machine import VirtualMachine, VMCloner
 from ocs_ci.ocs.dr.dr_workload import (
@@ -8488,6 +8490,29 @@ def factory(new_lifecycle_batch_size):
     return factory
 
 
+@pytest.fixture()
+def reduce_replication_delay(add_env_vars_to_noobaa_core_class):
+    """
+    Fixture to reduce the replication cycle delay
+
+    """
+
+    def factory(interval=1):
+        """
+        Factory function to reduce the replication
+        cycle delay
+
+        """
+        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():
     """
@@ -8815,21 +8840,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 = {
@@ -8842,27 +8880,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=enable_versioning,
         )
 
         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=enable_versioning,
+            )
+            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