split getting array from listing out to _extract_listing

lib/reindex/s3_bucketd.py

@@ -232,16 +232,12 @@ def get_next_upload_id(p):
                     upload_id=key['value']['UploadId']))
         return keys
 
-    def _sum_objects(self, bucket, listing):
+    def _sum_objects(self, bucket, listing, only_latest_when_locked = False):
         count = 0
         total_size = 0
         last_key = None
-        for status_code, payload in listing:
-            contents = payload['Versions'] if isinstance(payload, dict) else payload
-            if contents is None:
-                _log.error('Invalid contents in listing. bucket:%s status_code:%s'%(bucket.name, status_code))
-                raise InvalidListing(bucket.name)
-            for obj in contents:
+        try:
+            for obj in listing:
                 if isinstance(obj['value'], dict):
                     # bucketd v6 returns a dict:
                     data = obj.get('value', {})
@@ -254,16 +250,26 @@ def _sum_objects(self, bucket, listing):
                 is_latest = obj['key'] != last_key
                 last_key = obj['key']
 
-                if self._only_latest_when_locked and bucket.object_lock_enabled and not is_latest:
+                if only_latest_when_locked and bucket.object_lock_enabled and not is_latest:
                     _log.debug('Skipping versioned key: %s'%obj['key'])
                     continue
 
                 count += 1
                 total_size += size
 
-
+        except InvalidListing:
+            _log.error('Invalid contents in listing. bucket:%s status_code:%s'%(bucket.name, status_code))
+            raise InvalidListing(bucket.name)
         return count, total_size
 
+    def _extract_listing(self, key, listing):
+        for status_code, payload in listing:
+            contents = payload[key] if isinstance(payload, dict) else payload
+            if contents is None:
+                raise InvalidListing('')
+            for obj in contents:
+                yield obj
+
     def count_bucket_contents(self, bucket):
 
         def get_key_marker(p):
@@ -283,7 +289,8 @@ def get_vid_marker(p):
             'versionIdMarker': get_vid_marker,
         }
 
-        count, total_size = self._sum_objects(bucket, self._list_bucket(bucket.name, **params))
+        listing = self._list_bucket(bucket.name, **params)
+        count, total_size = self._sum_objects(bucket, self._extract_listing('Versions', listing), self._only_latest_when_locked)
         return BucketContents(
             bucket=bucket,
             obj_count=count,
@@ -312,7 +319,8 @@ def get_next_marker(p):
             'listingType': 'Delimiter',
         }
 
-        count, total_size = self._sum_objects(shadow_bucket, self._list_bucket(shadow_bucket_name, **params))
+        listing = self._list_bucket(shadow_bucket_name, **params)
+        count, total_size = self._sum_objects(shadow_bucket, self._extract_listing('Contents', listing))
         return BucketContents(
             bucket=shadow_bucket,
             obj_count=0, # MPU parts are not counted towards numberOfObjects

tests/functional/cron/testReindex.js

@@ -231,12 +231,12 @@ describe('UtapiReindex', () => {
                         bucketName: bucket,
                         contentLength: 1024,
                     })
-                        .setBucketContent({
-                            bucketName: MPUBucket,
-                            contentLength: 1024,
-                        })
-                        .setBucketCount(count)
-                        .createBuckets();
+                    .setBucketContent({
+                        bucketName: MPUBucket,
+                        contentLength: 1024,
+                    })
+                    .setBucketCount(count)
+                    .createBuckets();
 
                     function job() {
                         reindex._scheduleJob();