python: stress test for copy w/ --sync

Signed-off-by: Abhishek Gaikwad <[email protected]>

python/tests/integration/__init__.py

@@ -8,3 +8,4 @@
 TEST_TIMEOUT = 30
 TEST_TIMEOUT_LONG = 120
 OBJECT_COUNT = 10
+STRESS_TEST_OBJECT_COUNT = 500

python/tests/integration/sdk/remote_enabled_test.py

@@ -1,3 +1,7 @@
+#
+# Copyright (c) 2023-2024, NVIDIA CORPORATION. All rights reserved.
+#
+
 import unittest
 
 from aistore.sdk.const import PROVIDER_AIS
@@ -109,3 +113,27 @@ def _validate_objects_cached(self, objects, expected_cached):
                 self.assertTrue(obj.is_cached())
             else:
                 self.assertFalse(obj.is_cached())
+
+    def _verify_cached_objects(self, expected_object_count, cached_range):
+        """
+        List each of the objects and verify the correct count and that all objects matching
+        the cached range are cached and all others are not
+
+        Args:
+            expected_object_count: expected number of objects to list
+            cached_range: object indices that should be cached, all others should not
+        """
+        objects = self.bucket.list_objects(
+            props="name,cached", prefix=self.obj_prefix
+        ).entries
+        self.assertEqual(expected_object_count, len(objects))
+        cached_names = {self.obj_prefix + str(x) + "-suffix" for x in cached_range}
+        cached_objs = []
+        evicted_objs = []
+        for obj in objects:
+            if obj.name in cached_names:
+                cached_objs.append(obj)
+            else:
+                evicted_objs.append(obj)
+        self._validate_objects_cached(cached_objs, True)
+        self._validate_objects_cached(evicted_objs, False)

python/tests/integration/sdk/test_bucket_ops_stress.py

@@ -0,0 +1,67 @@
+#
+# Copyright (c) 2024, NVIDIA CORPORATION. All rights reserved.
+#
+import unittest
+import random
+import boto3
+
+from tests.integration import REMOTE_SET, TEST_TIMEOUT, STRESS_TEST_OBJECT_COUNT
+from tests.integration.sdk.remote_enabled_test import RemoteEnabledTest
+from tests import AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY
+from tests.integration.boto3 import AWS_REGION
+
+
+# pylint: disable=unused-variable,too-many-instance-attributes
+class TestBucketOpsStress(RemoteEnabledTest):
+    def setUp(self) -> None:
+        super().setUp()
+        if REMOTE_SET:
+            self.s3_client = boto3.client(
+                "s3",
+                region_name=AWS_REGION,
+                aws_access_key_id=AWS_ACCESS_KEY_ID,
+                aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
+            )
+
+    @unittest.skipIf(
+        not REMOTE_SET,
+        "Remote bucket is not set",
+    )
+    def test_stress_copy_objects_sync_flag(self):
+        obj_names = self._create_objects(
+            num_obj=STRESS_TEST_OBJECT_COUNT, suffix="-suffix"
+        )
+        to_bck_name = "dst-bck-cp-sync"
+        to_bck = self._create_bucket(to_bck_name)
+
+        obj_group = self.bucket.objects(obj_names=obj_names)
+        # self._evict_all_objects()
+
+        # cache and verify
+        job_id = obj_group.prefetch()
+        self.client.job(job_id).wait(timeout=TEST_TIMEOUT * 2)
+        self._verify_cached_objects(
+            STRESS_TEST_OBJECT_COUNT, range(STRESS_TEST_OBJECT_COUNT)
+        )
+
+        # copy objs to dst bck
+        copy_job = self.bucket.copy(prefix_filter=self.obj_prefix, to_bck=to_bck)
+        # copy_job = self.bucket.objects(obj_names=self.obj_names).copy(to_bck=to_bck)
+        self.client.job(job_id=copy_job).wait_for_idle(timeout=TEST_TIMEOUT)
+        self.assertEqual(STRESS_TEST_OBJECT_COUNT, len(to_bck.list_all_objects()))
+
+        # randomly delete 10% of the objects
+        to_delete = random.sample(obj_names, int(STRESS_TEST_OBJECT_COUNT * 0.1))
+
+        # out of band delete
+        for obj_name in to_delete:
+            self.s3_client.delete_object(Bucket=self.bucket.name, Key=obj_name)
+
+        # test --sync flag
+        copy_job = self.bucket.copy(
+            prefix_filter=self.obj_prefix, to_bck=to_bck, sync=True
+        )
+        self.client.job(job_id=copy_job).wait_for_idle(timeout=TEST_TIMEOUT * 3)
+        self.assertEqual(
+            int(STRESS_TEST_OBJECT_COUNT * 0.9), len(to_bck.list_all_objects())
+        )

python/tests/integration/sdk/test_object_group_ops.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2023, NVIDIA CORPORATION. All rights reserved.
+# Copyright (c) 2023-2024, NVIDIA CORPORATION. All rights reserved.
 #
 import hashlib
 import unittest
@@ -265,27 +265,3 @@ def _evict_all_objects(self):
         job_id = self.bucket.objects(obj_names=self.obj_names).evict()
         self.client.job(job_id).wait(timeout=TEST_TIMEOUT)
         self._check_all_objects_cached(OBJECT_COUNT, expected_cached=False)
-
-    def _verify_cached_objects(self, expected_object_count, cached_range):
-        """
-        List each of the objects and verify the correct count and that all objects matching
-        the cached range are cached and all others are not
-
-        Args:
-            expected_object_count: expected number of objects to list
-            cached_range: object indices that should be cached, all others should not
-        """
-        objects = self.bucket.list_objects(
-            props="name,cached", prefix=self.obj_prefix
-        ).entries
-        self.assertEqual(expected_object_count, len(objects))
-        cached_names = {self.obj_prefix + str(x) + "-suffix" for x in cached_range}
-        cached_objs = []
-        evicted_objs = []
-        for obj in objects:
-            if obj.name in cached_names:
-                cached_objs.append(obj)
-            else:
-                evicted_objs.append(obj)
-        self._validate_objects_cached(cached_objs, True)
-        self._validate_objects_cached(evicted_objs, False)