forked from TrueCloudLab/frostfs-testcases
Add assert_s3_acl
Signed-off-by: Elizaveta Chichindaeva <elizaveta@nspcc.ru>
This commit is contained in:
parent
c7a69b89e3
commit
9395a8003f
4 changed files with 66 additions and 101 deletions
|
@ -1,4 +1,5 @@
|
||||||
import datetime
|
import datetime
|
||||||
|
import logging
|
||||||
import os
|
import os
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
@ -8,6 +9,8 @@ import s3_gate_bucket
|
||||||
import s3_gate_object
|
import s3_gate_object
|
||||||
from dateutil.parser import parse
|
from dateutil.parser import parse
|
||||||
|
|
||||||
|
logger = logging.getLogger("NeoLogger")
|
||||||
|
|
||||||
|
|
||||||
@allure.step("Expected all objects are presented in the bucket")
|
@allure.step("Expected all objects are presented in the bucket")
|
||||||
def check_objects_in_bucket(
|
def check_objects_in_bucket(
|
||||||
|
@ -127,3 +130,31 @@ def assert_object_lock_mode(
|
||||||
assert (
|
assert (
|
||||||
retain_date - last_modify + timedelta(seconds=1)
|
retain_date - last_modify + timedelta(seconds=1)
|
||||||
).days == retain_period, f"Expected retention period is {retain_period} days"
|
).days == retain_period, f"Expected retention period is {retain_period} days"
|
||||||
|
|
||||||
|
|
||||||
|
def assert_s3_acl(acl_grants: list, permitted_users: str):
|
||||||
|
if permitted_users == "AllUsers":
|
||||||
|
grantees = {"AllUsers": 0, "CanonicalUser": 0}
|
||||||
|
for acl_grant in acl_grants:
|
||||||
|
if acl_grant.get("Grantee", {}).get("Type") == "Group":
|
||||||
|
uri = acl_grant.get("Grantee", {}).get("URI")
|
||||||
|
permission = acl_grant.get("Permission")
|
||||||
|
assert (uri, permission) == (
|
||||||
|
"http://acs.amazonaws.com/groups/global/AllUsers",
|
||||||
|
"FULL_CONTROL",
|
||||||
|
), "All Groups should have FULL_CONTROL"
|
||||||
|
grantees["AllUsers"] += 1
|
||||||
|
if acl_grant.get("Grantee", {}).get("Type") == "CanonicalUser":
|
||||||
|
permission = acl_grant.get("Permission")
|
||||||
|
assert permission == "FULL_CONTROL", "Canonical User should have FULL_CONTROL"
|
||||||
|
grantees["CanonicalUser"] += 1
|
||||||
|
assert grantees["AllUsers"] >= 1, "All Users should have FULL_CONTROL"
|
||||||
|
assert grantees["CanonicalUser"] >= 1, "Canonical User should have FULL_CONTROL"
|
||||||
|
|
||||||
|
if permitted_users == "CanonicalUser":
|
||||||
|
for acl_grant in acl_grants:
|
||||||
|
if acl_grant.get("Grantee", {}).get("Type") == "CanonicalUser":
|
||||||
|
permission = acl_grant.get("Permission")
|
||||||
|
assert permission == "FULL_CONTROL", "Only CanonicalUser should have FULL_CONTROL"
|
||||||
|
else:
|
||||||
|
logger.error("FULL_CONTROL is given to All Users")
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
import allure
|
import allure
|
||||||
import pytest
|
import pytest
|
||||||
from file_helper import generate_file
|
from file_helper import generate_file
|
||||||
from s3_helper import object_key_from_file_path
|
from s3_helper import assert_s3_acl, object_key_from_file_path
|
||||||
|
|
||||||
from steps import s3_gate_bucket, s3_gate_object
|
from steps import s3_gate_bucket, s3_gate_object
|
||||||
from steps.s3_gate_base import TestS3GateBase
|
from steps.s3_gate_base import TestS3GateBase
|
||||||
|
@ -29,19 +29,12 @@ class TestS3GateACL(TestS3GateBase):
|
||||||
with allure.step("Put object ACL = public-read"):
|
with allure.step("Put object ACL = public-read"):
|
||||||
s3_gate_object.put_object_acl_s3(self.s3_client, bucket, file_name, "public-read")
|
s3_gate_object.put_object_acl_s3(self.s3_client, bucket, file_name, "public-read")
|
||||||
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name)
|
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name)
|
||||||
obj_permission = [permission.get("Permission") for permission in obj_acl]
|
assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers")
|
||||||
assert obj_permission == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for all groups is FULL_CONTROL"
|
|
||||||
|
|
||||||
with allure.step("Put object ACL = private"):
|
with allure.step("Put object ACL = private"):
|
||||||
s3_gate_object.put_object_acl_s3(self.s3_client, bucket, file_name, "private")
|
s3_gate_object.put_object_acl_s3(self.s3_client, bucket, file_name, "private")
|
||||||
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name)
|
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name)
|
||||||
obj_permission = [permission.get("Permission") for permission in obj_acl]
|
assert_s3_acl(acl_grants=obj_acl, permitted_users="CanonicalUser")
|
||||||
assert obj_permission == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for Canonical User is FULL_CONTROL"
|
|
||||||
|
|
||||||
with allure.step(
|
with allure.step(
|
||||||
"Put object with grant-read uri=http://acs.amazonaws.com/groups/global/AllUsers"
|
"Put object with grant-read uri=http://acs.amazonaws.com/groups/global/AllUsers"
|
||||||
|
@ -53,30 +46,19 @@ class TestS3GateACL(TestS3GateBase):
|
||||||
grant_read="uri=http://acs.amazonaws.com/groups/global/AllUsers",
|
grant_read="uri=http://acs.amazonaws.com/groups/global/AllUsers",
|
||||||
)
|
)
|
||||||
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name)
|
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name)
|
||||||
obj_permission = [permission.get("Permission") for permission in obj_acl]
|
assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers")
|
||||||
assert obj_permission == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for all groups is FULL_CONTROL"
|
|
||||||
|
|
||||||
@allure.title("Test S3: Bucket ACL")
|
@allure.title("Test S3: Bucket ACL")
|
||||||
def test_s3_bucket_ACL(self):
|
def test_s3_bucket_ACL(self):
|
||||||
with allure.step("Create bucket with ACL = public-read-write"):
|
with allure.step("Create bucket with ACL = public-read-write"):
|
||||||
bucket = s3_gate_bucket.create_bucket_s3(self.s3_client, True, acl="public-read-write")
|
bucket = s3_gate_bucket.create_bucket_s3(self.s3_client, True, acl="public-read-write")
|
||||||
bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket)
|
bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket)
|
||||||
bucket_permission = [permission.get("Permission") for permission in bucket_acl]
|
assert_s3_acl(acl_grants=bucket_acl, permitted_users="AllUsers")
|
||||||
assert bucket_permission == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for all groups is FULL_CONTROL"
|
|
||||||
|
|
||||||
with allure.step("Change bucket ACL to private"):
|
with allure.step("Change bucket ACL to private"):
|
||||||
s3_gate_bucket.put_bucket_acl_s3(self.s3_client, bucket, acl="private")
|
s3_gate_bucket.put_bucket_acl_s3(self.s3_client, bucket, acl="private")
|
||||||
bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket)
|
bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket)
|
||||||
bucket_permission = [permission.get("Permission") for permission in bucket_acl]
|
assert_s3_acl(acl_grants=bucket_acl, permitted_users="CanonicalUser")
|
||||||
assert bucket_permission == [
|
|
||||||
"FULL_CONTROL"
|
|
||||||
], "Permission for CanonicalUser is FULL_CONTROL"
|
|
||||||
|
|
||||||
with allure.step(
|
with allure.step(
|
||||||
"Change bucket acl to --grant-write uri=http://acs.amazonaws.com/groups/global/AllUsers"
|
"Change bucket acl to --grant-write uri=http://acs.amazonaws.com/groups/global/AllUsers"
|
||||||
|
@ -87,8 +69,4 @@ class TestS3GateACL(TestS3GateBase):
|
||||||
grant_write="uri=http://acs.amazonaws.com/groups/global/AllUsers",
|
grant_write="uri=http://acs.amazonaws.com/groups/global/AllUsers",
|
||||||
)
|
)
|
||||||
bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket)
|
bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket)
|
||||||
bucket_permission = [permission.get("Permission") for permission in bucket_acl]
|
assert_s3_acl(acl_grants=bucket_acl, permitted_users="AllUsers")
|
||||||
assert bucket_permission == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for all groups is FULL_CONTROL"
|
|
||||||
|
|
|
@ -3,7 +3,12 @@ from datetime import datetime, timedelta
|
||||||
import allure
|
import allure
|
||||||
import pytest
|
import pytest
|
||||||
from file_helper import generate_file
|
from file_helper import generate_file
|
||||||
from s3_helper import assert_object_lock_mode, check_objects_in_bucket, object_key_from_file_path
|
from s3_helper import (
|
||||||
|
assert_object_lock_mode,
|
||||||
|
assert_s3_acl,
|
||||||
|
check_objects_in_bucket,
|
||||||
|
object_key_from_file_path,
|
||||||
|
)
|
||||||
|
|
||||||
from steps import s3_gate_bucket, s3_gate_object
|
from steps import s3_gate_bucket, s3_gate_object
|
||||||
from steps.s3_gate_base import TestS3GateBase
|
from steps.s3_gate_base import TestS3GateBase
|
||||||
|
@ -24,41 +29,26 @@ class TestS3GateBucket(TestS3GateBase):
|
||||||
with allure.step("Create bucket with ACL private"):
|
with allure.step("Create bucket with ACL private"):
|
||||||
bucket = s3_gate_bucket.create_bucket_s3(self.s3_client, True, acl="private")
|
bucket = s3_gate_bucket.create_bucket_s3(self.s3_client, True, acl="private")
|
||||||
bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket)
|
bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket)
|
||||||
bucket_permission = [permission.get("Permission") for permission in bucket_acl]
|
assert_s3_acl(acl_grants=bucket_acl, permitted_users="CanonicalUser")
|
||||||
assert bucket_permission == [
|
|
||||||
"FULL_CONTROL"
|
|
||||||
], "Permission for CanonicalUser is FULL_CONTROL"
|
|
||||||
|
|
||||||
with allure.step("Create bucket with ACL = public-read"):
|
with allure.step("Create bucket with ACL = public-read"):
|
||||||
bucket_1 = s3_gate_bucket.create_bucket_s3(self.s3_client, True, acl="public-read")
|
bucket_1 = s3_gate_bucket.create_bucket_s3(self.s3_client, True, acl="public-read")
|
||||||
bucket_acl_1 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_1)
|
bucket_acl_1 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_1)
|
||||||
bucket_permission_1 = [permission.get("Permission") for permission in bucket_acl_1]
|
assert_s3_acl(acl_grants=bucket_acl_1, permitted_users="AllUsers")
|
||||||
assert bucket_permission_1 == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for all groups is FULL_CONTROL"
|
|
||||||
|
|
||||||
with allure.step("Create bucket with ACL public-read-write"):
|
with allure.step("Create bucket with ACL public-read-write"):
|
||||||
bucket_2 = s3_gate_bucket.create_bucket_s3(
|
bucket_2 = s3_gate_bucket.create_bucket_s3(
|
||||||
self.s3_client, True, acl="public-read-write"
|
self.s3_client, True, acl="public-read-write"
|
||||||
)
|
)
|
||||||
bucket_acl_2 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_2)
|
bucket_acl_2 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_2)
|
||||||
bucket_permission_2 = [permission.get("Permission") for permission in bucket_acl_2]
|
assert_s3_acl(acl_grants=bucket_acl_2, permitted_users="AllUsers")
|
||||||
assert bucket_permission_2 == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for CanonicalUser is FULL_CONTROL"
|
|
||||||
|
|
||||||
with allure.step("Create bucket with ACL = authenticated-read"):
|
with allure.step("Create bucket with ACL = authenticated-read"):
|
||||||
bucket_3 = s3_gate_bucket.create_bucket_s3(
|
bucket_3 = s3_gate_bucket.create_bucket_s3(
|
||||||
self.s3_client, True, acl="authenticated-read"
|
self.s3_client, True, acl="authenticated-read"
|
||||||
)
|
)
|
||||||
bucket_acl_3 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_3)
|
bucket_acl_3 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_3)
|
||||||
bucket_permission_3 = [permission.get("Permission") for permission in bucket_acl_3]
|
assert_s3_acl(acl_grants=bucket_acl_3, permitted_users="AllUsers")
|
||||||
assert bucket_permission_3 == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for all groups is FULL_CONTROL"
|
|
||||||
|
|
||||||
@allure.title("Test S3: Create Bucket with different ACL by grand")
|
@allure.title("Test S3: Create Bucket with different ACL by grand")
|
||||||
def test_s3_create_bucket_with_grands(self):
|
def test_s3_create_bucket_with_grands(self):
|
||||||
|
@ -70,11 +60,7 @@ class TestS3GateBucket(TestS3GateBase):
|
||||||
grant_read="uri=http://acs.amazonaws.com/groups/global/AllUsers",
|
grant_read="uri=http://acs.amazonaws.com/groups/global/AllUsers",
|
||||||
)
|
)
|
||||||
bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket)
|
bucket_acl = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket)
|
||||||
bucket_permission = [permission.get("Permission") for permission in bucket_acl]
|
assert_s3_acl(acl_grants=bucket_acl, permitted_users="AllUsers")
|
||||||
assert bucket_permission == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for CanonicalUser is FULL_CONTROL"
|
|
||||||
|
|
||||||
with allure.step("Create bucket with --grant-wtite"):
|
with allure.step("Create bucket with --grant-wtite"):
|
||||||
bucket_1 = s3_gate_bucket.create_bucket_s3(
|
bucket_1 = s3_gate_bucket.create_bucket_s3(
|
||||||
|
@ -83,11 +69,7 @@ class TestS3GateBucket(TestS3GateBase):
|
||||||
grant_write="uri=http://acs.amazonaws.com/groups/global/AllUsers",
|
grant_write="uri=http://acs.amazonaws.com/groups/global/AllUsers",
|
||||||
)
|
)
|
||||||
bucket_acl_1 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_1)
|
bucket_acl_1 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_1)
|
||||||
bucket_permission_1 = [permission.get("Permission") for permission in bucket_acl_1]
|
assert_s3_acl(acl_grants=bucket_acl_1, permitted_users="AllUsers")
|
||||||
assert bucket_permission_1 == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for all groups is FULL_CONTROL"
|
|
||||||
|
|
||||||
with allure.step("Create bucket with --grant-full-control"):
|
with allure.step("Create bucket with --grant-full-control"):
|
||||||
bucket_2 = s3_gate_bucket.create_bucket_s3(
|
bucket_2 = s3_gate_bucket.create_bucket_s3(
|
||||||
|
@ -96,11 +78,7 @@ class TestS3GateBucket(TestS3GateBase):
|
||||||
grant_full_control="uri=http://acs.amazonaws.com/groups/global/AllUsers",
|
grant_full_control="uri=http://acs.amazonaws.com/groups/global/AllUsers",
|
||||||
)
|
)
|
||||||
bucket_acl_2 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_2)
|
bucket_acl_2 = s3_gate_bucket.get_bucket_acl(self.s3_client, bucket_2)
|
||||||
bucket_permission_2 = [permission.get("Permission") for permission in bucket_acl_2]
|
assert_s3_acl(acl_grants=bucket_acl_2, permitted_users="AllUsers")
|
||||||
assert bucket_permission_2 == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for CanonicalUser is FULL_CONTROL"
|
|
||||||
|
|
||||||
@allure.title("Test S3: create bucket with object lock")
|
@allure.title("Test S3: create bucket with object lock")
|
||||||
def test_s3_bucket_object_lock(self, simple_object_size):
|
def test_s3_bucket_object_lock(self, simple_object_size):
|
||||||
|
|
|
@ -12,7 +12,12 @@ from data_formatters import get_wallet_public_key
|
||||||
from file_helper import concat_files, generate_file, generate_file_with_content, get_file_hash
|
from file_helper import concat_files, generate_file, generate_file_with_content, get_file_hash
|
||||||
from neofs_testlib.utils.wallet import init_wallet
|
from neofs_testlib.utils.wallet import init_wallet
|
||||||
from python_keywords.payment_neogo import deposit_gas, transfer_gas
|
from python_keywords.payment_neogo import deposit_gas, transfer_gas
|
||||||
from s3_helper import assert_object_lock_mode, check_objects_in_bucket, set_bucket_versioning
|
from s3_helper import (
|
||||||
|
assert_object_lock_mode,
|
||||||
|
assert_s3_acl,
|
||||||
|
check_objects_in_bucket,
|
||||||
|
set_bucket_versioning,
|
||||||
|
)
|
||||||
|
|
||||||
from steps import s3_gate_bucket, s3_gate_object
|
from steps import s3_gate_bucket, s3_gate_object
|
||||||
from steps.s3_gate_base import TestS3GateBase
|
from steps.s3_gate_base import TestS3GateBase
|
||||||
|
@ -131,10 +136,7 @@ class TestS3GateObject(TestS3GateBase):
|
||||||
self.s3_client, bucket, obj_key, ACL="public-read-write"
|
self.s3_client, bucket, obj_key, ACL="public-read-write"
|
||||||
)
|
)
|
||||||
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, copy_obj_path)
|
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, copy_obj_path)
|
||||||
for control in obj_acl:
|
assert_s3_acl(acl_grants=obj_acl, permitted_users="CanonicalUser")
|
||||||
assert (
|
|
||||||
control.get("Permission") == "FULL_CONTROL"
|
|
||||||
), "Permission for all groups is FULL_CONTROL"
|
|
||||||
|
|
||||||
@allure.title("Test S3: Copy object with metadata")
|
@allure.title("Test S3: Copy object with metadata")
|
||||||
def test_s3_copy_metadate(self, bucket, simple_object_size):
|
def test_s3_copy_metadate(self, bucket, simple_object_size):
|
||||||
|
@ -703,8 +705,7 @@ class TestS3GateObject(TestS3GateBase):
|
||||||
with allure.step("Put object with acl private"):
|
with allure.step("Put object with acl private"):
|
||||||
s3_gate_object.put_object_s3(self.s3_client, bucket, file_path_1, ACL="private")
|
s3_gate_object.put_object_s3(self.s3_client, bucket, file_path_1, ACL="private")
|
||||||
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name)
|
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name)
|
||||||
obj_permission = [permission.get("Permission") for permission in obj_acl]
|
assert_s3_acl(acl_grants=obj_acl, permitted_users="CanonicalUser")
|
||||||
assert obj_permission == ["FULL_CONTROL"], "Permission for all groups is FULL_CONTROL"
|
|
||||||
object_1 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name)
|
object_1 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name)
|
||||||
assert get_file_hash(file_path_1) == get_file_hash(object_1), "Hashes must be the same"
|
assert get_file_hash(file_path_1) == get_file_hash(object_1), "Hashes must be the same"
|
||||||
|
|
||||||
|
@ -712,11 +713,7 @@ class TestS3GateObject(TestS3GateBase):
|
||||||
file_path_2 = generate_file_with_content(simple_object_size, file_path=file_path_1)
|
file_path_2 = generate_file_with_content(simple_object_size, file_path=file_path_1)
|
||||||
s3_gate_object.put_object_s3(self.s3_client, bucket, file_path_2, ACL="public-read")
|
s3_gate_object.put_object_s3(self.s3_client, bucket, file_path_2, ACL="public-read")
|
||||||
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name)
|
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name)
|
||||||
obj_permission = [permission.get("Permission") for permission in obj_acl]
|
assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers")
|
||||||
assert obj_permission == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for all groups is FULL_CONTROL"
|
|
||||||
object_2 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name)
|
object_2 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name)
|
||||||
assert get_file_hash(file_path_2) == get_file_hash(object_2), "Hashes must be the same"
|
assert get_file_hash(file_path_2) == get_file_hash(object_2), "Hashes must be the same"
|
||||||
|
|
||||||
|
@ -726,11 +723,7 @@ class TestS3GateObject(TestS3GateBase):
|
||||||
self.s3_client, bucket, file_path_3, ACL="public-read-write"
|
self.s3_client, bucket, file_path_3, ACL="public-read-write"
|
||||||
)
|
)
|
||||||
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name)
|
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name)
|
||||||
obj_permission = [permission.get("Permission") for permission in obj_acl]
|
assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers")
|
||||||
assert obj_permission == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for all groups is FULL_CONTROL"
|
|
||||||
object_3 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name)
|
object_3 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name)
|
||||||
assert get_file_hash(file_path_3) == get_file_hash(object_3), "Hashes must be the same"
|
assert get_file_hash(file_path_3) == get_file_hash(object_3), "Hashes must be the same"
|
||||||
|
|
||||||
|
@ -740,11 +733,7 @@ class TestS3GateObject(TestS3GateBase):
|
||||||
self.s3_client, bucket, file_path_4, ACL="authenticated-read"
|
self.s3_client, bucket, file_path_4, ACL="authenticated-read"
|
||||||
)
|
)
|
||||||
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name)
|
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name)
|
||||||
obj_permission = [permission.get("Permission") for permission in obj_acl]
|
assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers")
|
||||||
assert obj_permission == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for all groups is FULL_CONTROL"
|
|
||||||
object_4 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name)
|
object_4 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name)
|
||||||
assert get_file_hash(file_path_4) == get_file_hash(object_4), "Hashes must be the same"
|
assert get_file_hash(file_path_4) == get_file_hash(object_4), "Hashes must be the same"
|
||||||
|
|
||||||
|
@ -760,11 +749,7 @@ class TestS3GateObject(TestS3GateBase):
|
||||||
GrantFullControl=f"id={self.other_public_key}",
|
GrantFullControl=f"id={self.other_public_key}",
|
||||||
)
|
)
|
||||||
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name_5)
|
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name_5)
|
||||||
obj_permission = [permission.get("Permission") for permission in obj_acl]
|
assert_s3_acl(acl_grants=obj_acl, permitted_users="CanonicalUser")
|
||||||
assert obj_permission == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for all groups is FULL_CONTROL"
|
|
||||||
object_4 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name_5)
|
object_4 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name_5)
|
||||||
assert get_file_hash(file_path_5) == get_file_hash(object_4), "Hashes must be the same"
|
assert get_file_hash(file_path_5) == get_file_hash(object_4), "Hashes must be the same"
|
||||||
|
|
||||||
|
@ -779,11 +764,7 @@ class TestS3GateObject(TestS3GateBase):
|
||||||
GrantRead="uri=http://acs.amazonaws.com/groups/global/AllUsers",
|
GrantRead="uri=http://acs.amazonaws.com/groups/global/AllUsers",
|
||||||
)
|
)
|
||||||
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name_5)
|
obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, file_name_5)
|
||||||
obj_permission = [permission.get("Permission") for permission in obj_acl]
|
assert_s3_acl(acl_grants=obj_acl, permitted_users="AllUsers")
|
||||||
assert obj_permission == [
|
|
||||||
"FULL_CONTROL",
|
|
||||||
"FULL_CONTROL",
|
|
||||||
], "Permission for all groups is FULL_CONTROL"
|
|
||||||
object_7 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name_5)
|
object_7 = s3_gate_object.get_object_s3(self.s3_client, bucket, file_name_5)
|
||||||
assert get_file_hash(file_path_7) == get_file_hash(object_7), "Hashes must be the same"
|
assert get_file_hash(file_path_7) == get_file_hash(object_7), "Hashes must be the same"
|
||||||
|
|
||||||
|
@ -913,12 +894,9 @@ class TestS3GateObject(TestS3GateBase):
|
||||||
assert (
|
assert (
|
||||||
obj_head.get("Metadata") == object_metadata
|
obj_head.get("Metadata") == object_metadata
|
||||||
), f"Metadata of object is {object_metadata}"
|
), f"Metadata of object is {object_metadata}"
|
||||||
|
# Uncomment after https://github.com/nspcc-dev/neofs-s3-gw/issues/685 is solved
|
||||||
# obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, obj_key)
|
# obj_acl = s3_gate_object.get_object_acl_s3(self.s3_client, bucket, obj_key)
|
||||||
# obj_permission = [permission.get("Permission") for permission in obj_acl]
|
# assert_s3_acl(acl_grants = obj_acl, permitted_users = "AllUsers")
|
||||||
# assert obj_permission == [
|
|
||||||
# "FULL_CONTROL",
|
|
||||||
# "FULL_CONTROL",
|
|
||||||
# ], "Permission for all groups is FULL_CONTROL"
|
|
||||||
|
|
||||||
@allure.title("Test S3 Put 10 nested level object")
|
@allure.title("Test S3 Put 10 nested level object")
|
||||||
def test_s3_put_10_folder(self, bucket, temp_directory, simple_object_size):
|
def test_s3_put_10_folder(self, bucket, temp_directory, simple_object_size):
|
||||||
|
|
Loading…
Reference in a new issue