From: Matthew N. Heler Date: Wed, 4 Mar 2026 12:50:44 +0000 (-0600) Subject: s3tests: add tests for RestoreStatus in object listings X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=284738f1a20b1cad575c857ff60a564d96dfcd0a;p=s3-tests.git s3tests: add tests for RestoreStatus in object listings Add tests to validate the x-amz-optional-object-attributes RestoreStatus support in listing responses. New tests: - test_list_objects_restore_status: validates RestoreStatus appears in list_objects and list_objects_v2 responses for temporarily restored objects, and is absent for normal objects or when the header is omitted - test_list_object_versions_restore_status: validates RestoreStatus appears for restored non-current versions while unrestored versions remain unaffected Signed-off-by: Matthew N. Heler (cherry picked from commit 48494a1b454c1bcf985f7e07445cf0b839f8f22b) --- diff --git a/s3tests/functional/test_s3.py b/s3tests/functional/test_s3.py index 2f20ad75..ecac8b50 100644 --- a/s3tests/functional/test_s3.py +++ b/s3tests/functional/test_s3.py @@ -10481,6 +10481,165 @@ def test_restore_noncur_obj(): response = client.head_object(Bucket=bucket, Key=key, VersionId=version_ids[num]) assert response['ContentLength'] == 0 +@pytest.mark.cloud_restore +@pytest.mark.fails_on_aws +@pytest.mark.fails_on_dbstore +def test_list_objects_restore_status(): + cloud_sc = get_cloud_storage_class() + if cloud_sc is None: + pytest.skip('[s3 cloud] section missing cloud_storage_class') + + bucket = get_new_bucket() + client = get_client() + temp_key = 'test_restore_status_temp' + perm_key = 'test_restore_status_perm' + data = 'restore status listing data' + + client.put_object(Bucket=bucket, Key=temp_key, Body=data) + client.put_object(Bucket=bucket, Key=perm_key, Body=data) + + response = client.list_objects_v2( + Bucket=bucket, + OptionalObjectAttributes=['RestoreStatus']) + objs = response['Contents'] + assert len(objs) == 2 + for o in objs: + assert 'RestoreStatus' not in o + + response = client.list_objects( + Bucket=bucket, + OptionalObjectAttributes=['RestoreStatus']) + objs = response['Contents'] + assert len(objs) == 2 + for o in objs: + assert 'RestoreStatus' not in o + + rules = [{'ID': 'rule1', 'Transitions': [{'Days': 1, 'StorageClass': cloud_sc}], 'Prefix': '', 'Status': 'Enabled'}] + lifecycle = {'Rules': rules} + client.put_bucket_lifecycle_configuration(Bucket=bucket, LifecycleConfiguration=lifecycle) + + lc_interval = get_lc_debug_interval() + restore_period = get_restore_processor_period() + time.sleep(10 * lc_interval) + + verify_transition(client, bucket, temp_key, cloud_sc) + verify_transition(client, bucket, perm_key, cloud_sc) + + # delete lifecycle to prevent re-transition before restore check + client.delete_bucket_lifecycle(Bucket=bucket) + + client.restore_object(Bucket=bucket, Key=temp_key, RestoreRequest={'Days': 20}) + # permanent restore omits 'Days' + client.restore_object(Bucket=bucket, Key=perm_key, RestoreRequest={}) + time.sleep(3 * restore_period) + + response = client.list_objects_v2( + Bucket=bucket, + OptionalObjectAttributes=['RestoreStatus']) + objs = {o['Key']: o for o in response['Contents']} + assert len(objs) == 2 + assert 'RestoreStatus' in objs[temp_key] + assert objs[temp_key]['RestoreStatus']['IsRestoreInProgress'] == False + assert 'RestoreExpiryDate' in objs[temp_key]['RestoreStatus'] + assert 'RestoreStatus' in objs[perm_key] + assert objs[perm_key]['RestoreStatus']['IsRestoreInProgress'] == False + assert 'RestoreExpiryDate' not in objs[perm_key]['RestoreStatus'] + + response = client.list_objects( + Bucket=bucket, + OptionalObjectAttributes=['RestoreStatus']) + objs = {o['Key']: o for o in response['Contents']} + assert len(objs) == 2 + assert 'RestoreStatus' in objs[temp_key] + assert objs[temp_key]['RestoreStatus']['IsRestoreInProgress'] == False + assert 'RestoreExpiryDate' in objs[temp_key]['RestoreStatus'] + assert 'RestoreStatus' in objs[perm_key] + assert objs[perm_key]['RestoreStatus']['IsRestoreInProgress'] == False + assert 'RestoreExpiryDate' not in objs[perm_key]['RestoreStatus'] + + # without the header, RestoreStatus should not appear + response = client.list_objects_v2(Bucket=bucket) + objs = response['Contents'] + assert len(objs) == 2 + for o in objs: + assert 'RestoreStatus' not in o + +@pytest.mark.cloud_restore +@pytest.mark.fails_on_aws +@pytest.mark.fails_on_dbstore +def test_list_object_versions_restore_status(): + cloud_sc = get_cloud_storage_class() + if cloud_sc is None: + pytest.skip('[s3 cloud] section missing cloud_storage_class') + + bucket = get_new_bucket() + client = get_client() + + key = 'test1/a' + data = 'restore status versioned data' + + # create initial version before enabling versioning + client.put_object(Bucket=bucket, Key=key, Body=data) + check_configure_versioning_retry(bucket, "Enabled", "Enabled") + + # create a new version so the original becomes non-current + (version_ids, contents) = create_multiple_versions(client, bucket, key, 1) + + response = client.list_object_versions( + Bucket=bucket, + OptionalObjectAttributes=['RestoreStatus']) + for v in response['Versions']: + assert 'RestoreStatus' not in v + + # transition non-current versions to cloud + rules = [{ + 'ID': 'rule1', + 'Prefix': 'test1/', + 'Status': 'Enabled', + 'NoncurrentVersionTransitions': [{ + 'NoncurrentDays': 2, + 'StorageClass': cloud_sc + }], + }] + lifecycle = {'Rules': rules} + client.put_bucket_lifecycle_configuration(Bucket=bucket, LifecycleConfiguration=lifecycle) + + lc_interval = get_lc_debug_interval() + restore_period = get_restore_processor_period() + time.sleep(7 * lc_interval) + + # find the non-current version + response = client.list_object_versions(Bucket=bucket) + noncur_id = None + for v in response['Versions']: + if not v['IsLatest']: + noncur_id = v['VersionId'] + break + assert noncur_id is not None + + verify_transition(client, bucket, key, cloud_sc, noncur_id) + + # delete lifecycle to prevent re-transition before restore check + client.delete_bucket_lifecycle(Bucket=bucket) + + client.restore_object(Bucket=bucket, Key=key, VersionId=noncur_id, RestoreRequest={'Days': 20}) + time.sleep(3 * restore_period) + + response = client.list_object_versions( + Bucket=bucket, + OptionalObjectAttributes=['RestoreStatus']) + for v in response['Versions']: + if v['VersionId'] == noncur_id: + assert 'RestoreStatus' in v + assert v['RestoreStatus']['IsRestoreInProgress'] == False + assert 'RestoreExpiryDate' in v['RestoreStatus'] + else: + assert 'RestoreStatus' not in v + + response = client.list_object_versions(Bucket=bucket) + for v in response['Versions']: + assert 'RestoreStatus' not in v + @pytest.mark.encryption @pytest.mark.fails_on_dbstore def test_encrypted_transfer_1b():