From: Casey Bodley Date: Thu, 3 Jul 2025 21:19:07 +0000 (-0400) Subject: s3: test delete_object() with IfMatchLastModifiedTime X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=4b0421c1da5ed21e29ab5dda2de1828fe561650b;p=s3-tests.git s3: test delete_object() with IfMatchLastModifiedTime Signed-off-by: Casey Bodley (cherry picked from commit d67d70d0f11aacfdf01a1b80d9bc23d21b041367) --- diff --git a/s3tests_boto3/functional/test_s3.py b/s3tests_boto3/functional/test_s3.py index 09f29244..eacb0070 100644 --- a/s3tests_boto3/functional/test_s3.py +++ b/s3tests_boto3/functional/test_s3.py @@ -18072,3 +18072,155 @@ def test_put_object_current_if_match(): assert (404, 'NoSuchKey') == _get_status_and_error_code(e.response) client.put_object(Bucket=bucket, Key=key, IfNoneMatch=etag) + +@pytest.mark.fails_on_aws # only supported for directory buckets +@pytest.mark.conditional_write +def test_delete_object_if_match(): + client = get_client() + bucket = get_new_bucket(client) + key = 'obj' + + etag = client.put_object(Bucket=bucket, Key=key)['ETag'] + + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, IfMatch='badetag') + assert (412, 'PreconditionFailed') == _get_status_and_error_code(e.response) + + client.delete_object(Bucket=bucket, Key=key, IfMatch=etag) + + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, IfMatch='*') + assert (404, 'NoSuchKey') == _get_status_and_error_code(e.response) + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, IfMatch='badetag') + assert (404, 'NoSuchKey') == _get_status_and_error_code(e.response) + + # recreate to test IfMatch='*' + client.put_object(Bucket=bucket, Key=key) + client.delete_object(Bucket=bucket, Key=key, IfMatch='*') + +@pytest.mark.fails_on_aws # only supported for directory buckets +@pytest.mark.conditional_write +def test_delete_object_current_if_match(): + client = get_client() + bucket = get_new_bucket(client) + check_configure_versioning_retry(bucket, "Enabled", "Enabled") + key = 'obj' + + # on nonexistent object + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, IfMatch='*') + assert (404, 'NoSuchKey') == _get_status_and_error_code(e.response) + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, IfMatch='badetag') + assert (404, 'NoSuchKey') == _get_status_and_error_code(e.response) + + response = client.put_object(Bucket=bucket, Key=key) + version = response['VersionId'] + etag = response['ETag'] + + # on current version + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, IfMatch='badetag') + assert (412, 'PreconditionFailed') == _get_status_and_error_code(e.response) + + response = client.delete_object(Bucket=bucket, Key=key, IfMatch=etag) + assert response['DeleteMarker'] + + # on current delete marker + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, IfMatch='*') + assert (404, 'NoSuchKey') == _get_status_and_error_code(e.response) + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, IfMatch='badetag') + assert (404, 'NoSuchKey') == _get_status_and_error_code(e.response) + + # remove delete marker to retest IfMatch='*' + client.delete_object(Bucket=bucket, Key=key, VersionId=version) + client.delete_object(Bucket=bucket, Key=key, IfMatch='*') + +@pytest.mark.fails_on_aws # only supported for directory buckets +@pytest.mark.conditional_write +def test_delete_object_version_if_match(): + client = get_client() + bucket = get_new_bucket(client) + check_configure_versioning_retry(bucket, "Enabled", "Enabled") + key = 'obj' + + response = client.put_object(Bucket=bucket, Key=key) + version = response['VersionId'] + etag = response['ETag'] + + # on specific version + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, VersionId=version, IfMatch='badetag') + assert (412, 'PreconditionFailed') == _get_status_and_error_code(e.response) + + response = client.delete_object(Bucket=bucket, Key=key, VersionId=version, IfMatch=etag) + assert not response['DeleteMarker'] + + # on nonexistent version + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, VersionId=version, IfMatch='*') + assert (404, 'NoSuchKey') == _get_status_and_error_code(e.response) + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, VersionId=version, IfMatch='badetag') + assert (404, 'NoSuchKey') == _get_status_and_error_code(e.response) + + # recreate to test IfMatch='*' + response = client.put_object(Bucket=bucket, Key=key) + version = response['VersionId'] + client.delete_object(Bucket=bucket, Key=key, VersionId=version, IfMatch='*') + +@pytest.mark.fails_on_aws # only supported for directory buckets +@pytest.mark.conditional_write +def test_delete_object_if_match_last_modified_time(): + client = get_client() + bucket = get_new_bucket(client) + key = 'obj' + + badmtime = datetime.datetime(2015, 1, 1) + + client.put_object(Bucket=bucket, Key=key) + mtime = client.head_object(Bucket=bucket, Key=key)['LastModified'] + + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, IfMatchLastModifiedTime=badmtime) + assert (412, 'PreconditionFailed') == _get_status_and_error_code(e.response) + + client.delete_object(Bucket=bucket, Key=key, IfMatchLastModifiedTime=mtime) + + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, IfMatchLastModifiedTime=badmtime) + assert (404, 'NoSuchKey') == _get_status_and_error_code(e.response) + +@pytest.mark.fails_on_aws # only supported for directory buckets +@pytest.mark.conditional_write +def test_delete_object_current_if_match_last_modified_time(): + client = get_client() + bucket = get_new_bucket(client) + check_configure_versioning_retry(bucket, "Enabled", "Enabled") + key = 'obj' + + badmtime = datetime.datetime(2015, 1, 1) + + client.put_object(Bucket=bucket, Key=key) + mtime = client.head_object(Bucket=bucket, Key=key)['LastModified'] + + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, IfMatchLastModifiedTime=badmtime) + assert (412, 'PreconditionFailed') == _get_status_and_error_code(e.response) + + response = client.delete_object(Bucket=bucket, Key=key, IfMatchLastModifiedTime=mtime) + assert response['DeleteMarker'] + + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, IfMatchLastModifiedTime=badmtime) + assert (404, 'NoSuchKey') == _get_status_and_error_code(e.response) + +@pytest.mark.fails_on_aws # only supported for directory buckets +@pytest.mark.conditional_write +def test_delete_object_version_if_match_last_modified_time(): + client = get_client() + bucket = get_new_bucket(client) + check_configure_versioning_retry(bucket, "Enabled", "Enabled") + key = 'obj' + + badmtime = datetime.datetime(2015, 1, 1) + + version = client.put_object(Bucket=bucket, Key=key)['VersionId'] + mtime = client.head_object(Bucket=bucket, Key=key)['LastModified'] + + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, VersionId=version, IfMatchLastModifiedTime=badmtime) + assert (412, 'PreconditionFailed') == _get_status_and_error_code(e.response) + + response = client.delete_object(Bucket=bucket, Key=key, VersionId=version, IfMatchLastModifiedTime=mtime) + assert not response['DeleteMarker'] + + e = assert_raises(ClientError, client.delete_object, Bucket=bucket, Key=key, VersionId=version, IfMatchLastModifiedTime=badmtime) + assert (404, 'NoSuchKey') == _get_status_and_error_code(e.response)