test ingestion of non versioned objets

Kerkesni · Kerkesni · commit e762c3f598fc · 2025-02-20T11:54:56.000+01:00
Issue: ZENKO-4330
diff --git a/.github/workflows/end2end.yaml b/.github/workflows/end2end.yaml
@@ -81,6 +81,9 @@ env:
   RING_S3C_ENDPOINT: http://s3c.local:8000
   RING_S3C_BACKEND_SOURCE_LOCATION: rings3cbackendingestion
   RING_S3C_INGESTION_SRC_BUCKET_NAME: ingestion-test-src-bucket-${{ github.run_id }}
+  RING_S3C_BACKEND_SOURCE_NON_VERSIONED_LOCATION: rings3cbackendingestionnonversioned
+  RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME: ingestion-test-src-non-versioned-bucket-${{ github.run_id }}
+  RING_S3C_INGESTION_NON_VERSIONED_OBJECT_COUNT_PER_TYPE: 2
   # CTST end2end tests
   NOTIF_DEST_NAME: "destination1"
   NOTIF_DEST_TOPIC: "destination-topic-1"
diff --git a/tests/zenko_tests/create_buckets.py b/tests/zenko_tests/create_buckets.py
@@ -4,6 +4,7 @@
 from azure.core.credentials import AzureNamedKeyCredential
 from azure.core.exceptions import ResourceExistsError
 from azure.storage.queue import QueueServiceClient
+import math
 import os
 import logging
 
@@ -25,13 +26,60 @@ def bucket_safe_create(bucket):
         _log.info('Error creating bucket %s - %s' % (bucket.name, str(exp)))
         raise exp
 
+def put_file(bucket, object_name, body):
+    """
+    Put an object in a bucket
+    """
+    try:
+        _log.info('Putting object %s' % object_name)
+        obj = bucket.Object(object_name)
+        obj.put(Body=body)
+    except Exception as exp:  # pylint: disable=broad-except
+        _log.info('Error putting object %s - %s' % (bucket.name, object_name, str(exp)))
+        raise exp
+    
+def put_singlepart_mpu(bucket, object_name, body):
+    """
+    Put an object in a bucket using a single part multipart upload
+    """
+    try:
+        _log.info('Putting object %s using multipart upload' % object_name)
+        mpu = bucket.meta.client.create_multipart_upload(
+            Bucket=bucket.name,
+            Key=object_name
+        )
+        response = bucket.meta.client.upload_part(
+            Body=body,
+            Bucket=bucket.name,
+            Key=object_name,
+            PartNumber=1,
+            UploadId=mpu['UploadId']
+        )
+        bucket.meta.client.complete_multipart_upload(
+            Bucket=bucket.name,
+            Key=object_name,
+            UploadId=mpu['UploadId'],
+            MultipartUpload={
+                'Parts': [{
+                    'PartNumber': 1,
+                    'ETag': response['ETag']
+                }]
+            }
+        )
+    except Exception as exp:
+        _log.error('Error in multipart upload for object %s - %s' % (object_name, str(exp)))
+        raise exp
+
 def create_ring_buckets():
     VERIFY_CERTIFICATES = get_env('VERIFY_CERTIFICATES', False)
     RING_S3C_ACCESS_KEY = get_env('RING_S3C_ACCESS_KEY')
     RING_S3C_SECRET_KEY = get_env('RING_S3C_SECRET_KEY')
-    RING_S3C_INGESTION_SRC_BUCKET_NAME = get_env('RING_S3C_INGESTION_SRC_BUCKET_NAME')
     RING_S3C_ENDPOINT = get_env('RING_S3C_ENDPOINT')
     ENABLE_RING_TESTS = get_env('ENABLE_RING_TESTS')
+    ## test bucket names
+    RING_S3C_INGESTION_SRC_BUCKET_NAME = get_env('RING_S3C_INGESTION_SRC_BUCKET_NAME')
+    RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME = get_env('RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME')
+    RING_S3C_INGESTION_NON_VERSIONED_OBJECT_COUNT_PER_TYPE = get_env('RING_S3C_INGESTION_NON_VERSIONED_OBJECT_COUNT_PER_TYPE')
 
     # Disable if Ring is not enabled
     if ENABLE_RING_TESTS == "false":
@@ -42,10 +90,28 @@ def create_ring_buckets():
     ring_s3c_client = s3c.resource('s3', endpoint_url=RING_S3C_ENDPOINT,
                       verify=VERIFY_CERTIFICATES)
 
+    versioned_bucket = ring_s3c_client.Bucket(RING_S3C_INGESTION_SRC_BUCKET_NAME)
+    non_versioned_bucket = ring_s3c_client.Bucket(RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME)
+
     ## Creating S3C buckets
     _log.info('Creating S3C buckets...')
-    bucket_safe_create(ring_s3c_client.Bucket(RING_S3C_INGESTION_SRC_BUCKET_NAME))
-    ring_s3c_client.Bucket(RING_S3C_INGESTION_SRC_BUCKET_NAME).Versioning().enable()
+    bucket_safe_create(versioned_bucket)
+    bucket_safe_create(non_versioned_bucket)
+
+    ## Adding non versioned objects before test execution to avoid
+    ## having to create the location mid tests which might cause flakiness.
+    ## A RING location can only be created if the bucket is versioned, and
+    ## once versioning is enabled it cannot be disabled.
+    _log.info('Putting non versioned objects...')
+    for i in range(int(RING_S3C_INGESTION_NON_VERSIONED_OBJECT_COUNT_PER_TYPE)):
+        put_file(non_versioned_bucket, 'simple-%d' % i, b'data')
+        put_file(non_versioned_bucket, 'zerobyte-%d' % i, b'')
+        put_singlepart_mpu(non_versioned_bucket, 'mpu-singlepart-%d' % i, b'mpudata')
+
+    ## Enabling versioning
+    _log.info('Enabling versioning on buckets...')
+    versioned_bucket.Versioning().enable()
+    non_versioned_bucket.Versioning().enable()
 
 def create_aws_buckets():
     VERIFY_CERTIFICATES = get_env('VERIFY_CERTIFICATES', False)
@@ -86,7 +152,6 @@ def create_azure_containers():
         except ResourceExistsError:
             _log.info('Container %s already exists!' % bucket_name)
 
-
 def create_azure_queues():
     AZURE_BACKEND_QUEUE_ENDPOINT = get_env("AZURE_BACKEND_QUEUE_ENDPOINT")
     AZURE_ACCOUNT_NAME = get_env("AZURE_ACCOUNT_NAME")
diff --git a/tests/zenko_tests/e2e-config.yaml.template b/tests/zenko_tests/e2e-config.yaml.template
@@ -48,6 +48,15 @@ locations:
       secretKey: "${RING_S3C_SECRET_KEY}"
       bucketMatch: yes
       repoId: []
+  - name: "${RING_S3C_BACKEND_SOURCE_NON_VERSIONED_LOCATION}"
+    locationType: "location-scality-ring-s3-v1"
+    details:
+      bucketName: "${RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME}"
+      endpoint: "${RING_S3C_ENDPOINT}"
+      accessKey: "${RING_S3C_ACCESS_KEY}"
+      secretKey: "${RING_S3C_SECRET_KEY}"
+      bucketMatch: yes
+      repoId: []
   - name: "${GCP_BACKEND_DESTINATION_LOCATION}"
     locationType: "location-gcp-v1"
     details:
diff --git a/tests/zenko_tests/node_tests/backbeat/tests/ingestion/ingestionS3C.js b/tests/zenko_tests/node_tests/backbeat/tests/ingestion/ingestionS3C.js
@@ -9,6 +9,10 @@ const scalityUtils = new IngestionUtility(scalityS3Client, ringS3Client);
 const ringS3CUtils = new IngestionUtility(ringS3Client);
 const ingestionSrcBucket = process.env.RING_S3C_INGESTION_SRC_BUCKET_NAME;
 const srcLocation = process.env.RING_S3C_BACKEND_SOURCE_LOCATION;
+const ingestionNonVersionedSrcBucket = process.env.RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME;
+// Multiplying by 3 to account for the 3 types of objects: simple, zero-byte, and multipart
+const nonVersionedObjectCount = parseInt(process.RING_S3C_INGESTION_NON_VERSIONED_OBJECT_COUNT_PER_TYPE, 10) * 3;
+const srcNonVersionedLocation = process.env.RING_S3C_BACKEND_SOURCE_NON_VERSIONED_LOCATION;
 const location = srcLocation;
 let INGESTION_DEST_BUCKET;
 let KEY_PREFIX;
@@ -356,4 +360,34 @@ describe('Ingesting existing data from RING S3C bucket', () => {
             next,
         ),
     ], done));
+
+    // The source bucket used in this test has non versioned objects of 3 types: simple, zero-byte, and multipart
+    // These non versioned objects were put before creating the location as the bucket has to be versioned
+    // for the location to be created, and once it's versioned can't be set to non versioned again. This is
+    // done to avoid having to create the location during the tests which might impact other tests running
+    // in parallel.
+    it('should ingest all non versioned objects', done => async.waterfall([
+        next => scalityUtils.createIngestionBucket(
+            INGESTION_DEST_BUCKET,
+            srcNonVersionedLocation,
+            next,
+        ),
+        next => ringS3CUtils.s3.listObjectVersions({
+            Bucket: ingestionNonVersionedSrcBucket,
+        }, next),
+        (data, next) => {
+            assert.strictEqual(data.Versions.length, nonVersionedObjectCount);
+            assert.strictEqual(data.DeleteMarkers.length, 0);
+            async.forEach(data.Versions, (version, cb) => {
+                scalityUtils.compareObjectsRINGS3C(
+                    ingestionNonVersionedSrcBucket,
+                    INGESTION_DEST_BUCKET,
+                    version.Key,
+                    'null',
+                    undefined,
+                    cb,
+                );
+            }, next);
+        },
+    ], done));
 });
