forked from TrueCloudLab/s3-tests
[#XX] Add test_frostfs.py
Signed-off-by: Denis Kirillov <d.kirillov@yadro.com>
This commit is contained in:
parent
08df9352f9
commit
f8120b10f7
1 changed files with 177 additions and 0 deletions
177
s3tests_boto3/functional/test_frostfs.py
Normal file
177
s3tests_boto3/functional/test_frostfs.py
Normal file
|
|
@ -0,0 +1,177 @@
|
|||
from botocore.exceptions import ClientError
|
||||
import json
|
||||
import time
|
||||
|
||||
from .utils import assert_raises
|
||||
from .utils import _get_status_and_error_code
|
||||
|
||||
from . import (
|
||||
configfile, setup_teardown, # we need this to parse config
|
||||
get_client,
|
||||
get_unauthenticated_client,
|
||||
get_new_bucket,
|
||||
)
|
||||
|
||||
|
||||
def test_bucket_policy_frostfs_deny():
|
||||
bucket_name = get_new_bucket()
|
||||
client = get_client()
|
||||
key = 'tmp'
|
||||
|
||||
resource1 = "arn:aws:s3:::" + bucket_name
|
||||
resource2 = "arn:aws:s3:::" + bucket_name + "/*"
|
||||
policy_document = json.dumps(
|
||||
{
|
||||
"Version": "2012-10-17",
|
||||
"Statement": [{
|
||||
"Effect": "Deny",
|
||||
"Principal": "*",
|
||||
"Action": "s3:PutObject",
|
||||
"Resource": [
|
||||
"{}".format(resource1),
|
||||
"{}".format(resource2)
|
||||
],
|
||||
"Condition": {
|
||||
"StringNotEquals": {
|
||||
"s3:RequestObjectTag/environment": "production"
|
||||
}
|
||||
}
|
||||
}]
|
||||
}
|
||||
)
|
||||
client.put_bucket_policy(Bucket=bucket_name, Policy=policy_document)
|
||||
|
||||
# TEST 7
|
||||
client.put_object(Bucket=bucket_name, Key=key, Tagging='environment=production')
|
||||
|
||||
# TEST 8
|
||||
e = assert_raises(ClientError, client.put_object, Bucket=bucket_name, Key=key, Tagging='environment=development')
|
||||
status, error_code = _get_status_and_error_code(e.response)
|
||||
assert status == 403
|
||||
assert error_code == 'AccessDenied'
|
||||
|
||||
# TEST 9
|
||||
e = assert_raises(ClientError, client.put_object, Bucket=bucket_name, Key=key)
|
||||
status, error_code = _get_status_and_error_code(e.response)
|
||||
assert status == 403
|
||||
assert error_code == 'AccessDenied'
|
||||
|
||||
|
||||
|
||||
policy_document = json.dumps(
|
||||
{
|
||||
"Version": "2012-10-17",
|
||||
"Statement": [{
|
||||
"Effect": "Deny",
|
||||
"Principal": "*",
|
||||
"Action": "s3:PutObject",
|
||||
"Resource": [
|
||||
"{}".format(resource1),
|
||||
"{}".format(resource2)
|
||||
],
|
||||
"Condition": {
|
||||
"StringEquals": {
|
||||
"s3:RequestObjectTag/environment": "production"
|
||||
}
|
||||
}
|
||||
}]
|
||||
}
|
||||
)
|
||||
client.put_bucket_policy(Bucket=bucket_name, Policy=policy_document)
|
||||
|
||||
# TEST 10
|
||||
e = assert_raises(ClientError, client.put_object, Bucket=bucket_name, Key=key, Tagging='environment=production')
|
||||
status, error_code = _get_status_and_error_code(e.response)
|
||||
assert status == 403
|
||||
assert error_code == 'AccessDenied'
|
||||
|
||||
# TEST 11
|
||||
client.put_object(Bucket=bucket_name, Key=key, Tagging='environment=development')
|
||||
|
||||
# TEST 12
|
||||
client.put_object(Bucket=bucket_name, Key=key)
|
||||
|
||||
def test_bucket_policy_frostfs_allow():
|
||||
bucket_name = get_new_bucket()
|
||||
client = get_client()
|
||||
key = 'tmp'
|
||||
|
||||
client.put_object(Bucket=bucket_name, Key=key)
|
||||
|
||||
resource1 = "arn:aws:s3:::" + bucket_name
|
||||
resource2 = "arn:aws:s3:::" + bucket_name + "/*"
|
||||
policy_document = json.dumps(
|
||||
{
|
||||
"Version": "2012-10-17",
|
||||
"Statement": [{
|
||||
"Effect": "Allow",
|
||||
"Principal": "*",
|
||||
"Action": "s3:PutObjectTagging",
|
||||
"Resource": [
|
||||
"{}".format(resource1),
|
||||
"{}".format(resource2)
|
||||
],
|
||||
"Condition": {
|
||||
"StringNotEquals": {
|
||||
"s3:RequestObjectTag/environment": "production"
|
||||
}
|
||||
}
|
||||
}]
|
||||
}
|
||||
)
|
||||
client.put_bucket_policy(Bucket=bucket_name, Policy=policy_document)
|
||||
time.sleep(3)
|
||||
|
||||
alt_client = get_unauthenticated_client()
|
||||
|
||||
# TEST 1
|
||||
e = assert_raises(ClientError, alt_client.put_object_tagging, Bucket=bucket_name, Key=key, Tagging={'TagSet':[{'Key':'environment','Value':'production'}]})
|
||||
status, error_code = _get_status_and_error_code(e.response)
|
||||
assert status == 403
|
||||
assert error_code == 'AccessDenied'
|
||||
|
||||
# TEST 2
|
||||
alt_client.put_object_tagging(Bucket=bucket_name, Key=key, Tagging={'TagSet':[{'Key':'environment','Value':'development'}]})
|
||||
|
||||
# TEST 3
|
||||
alt_client.put_object_tagging(Bucket=bucket_name, Key=key, Tagging={'TagSet':[]})
|
||||
|
||||
|
||||
|
||||
policy_document2 = json.dumps(
|
||||
{
|
||||
"Version": "2012-10-17",
|
||||
"Statement": [{
|
||||
"Effect": "Allow",
|
||||
"Principal": "*",
|
||||
"Action": "s3:PutObjectTagging",
|
||||
"Resource": [
|
||||
"{}".format(resource1),
|
||||
"{}".format(resource2)
|
||||
],
|
||||
"Condition": {
|
||||
"StringEquals": {
|
||||
"s3:RequestObjectTag/environment": "production"
|
||||
}
|
||||
}
|
||||
}]
|
||||
}
|
||||
)
|
||||
client.put_bucket_policy(Bucket=bucket_name, Policy=policy_document2)
|
||||
time.sleep(15) # probably we can reduce this
|
||||
|
||||
# TEST 4
|
||||
alt_client.put_object_tagging(Bucket=bucket_name, Key=key, Tagging={'TagSet':[{'Key':'environment','Value':'production'}]})
|
||||
|
||||
# TEST 5
|
||||
e = assert_raises(ClientError, alt_client.put_object_tagging, Bucket=bucket_name, Key=key, Tagging={'TagSet':[{'Key':'environment','Value':'development'}]})
|
||||
status, error_code = _get_status_and_error_code(e.response)
|
||||
assert status == 403
|
||||
assert error_code == 'AccessDenied'
|
||||
|
||||
# TEST 6
|
||||
e = assert_raises(ClientError, alt_client.put_object_tagging, Bucket=bucket_name, Key=key, Tagging={'TagSet':[]})
|
||||
status, error_code = _get_status_and_error_code(e.response)
|
||||
assert status == 403
|
||||
assert error_code == 'AccessDenied'
|
||||
|
||||
Loading…
Reference in a new issue