abereziny/frostfs-testcases
1
0
Fork 0
forked from TrueCloudLab/frostfs-testcases
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity

[#286] Add APE tests with objectID filter

Browse source 
Signed-off-by: a.berezin <a.berezin@yadro.com>
This commit is contained in:
Andrey Berezin 2024-08-08 18:35:59 +03:00
parent 3806185c74
commit d08aacd421
1 changed files with 62 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

65
pytest_tests/testsuites/access/ape/test_ape_filters.py
View file

@ -57,7 +57,7 @@ class TestApeFilters(ClusterTestBase):
return cid, objects_with_header, objects_with_other_header, objects_without_header, file_path
@pytest.fixture(scope="function")
def container_with_objects(self, default_wallet: WalletInfo, file_path: TestFile, frostfs_cli: FrostfsCli, cluster: Cluster):
def private_container(self, default_wallet: WalletInfo, frostfs_cli: FrostfsCli, cluster: Cluster):
with reporter.step("Create private container"):
cid = create_container(default_wallet, self.shell, self.cluster.default_rpc_endpoint, basic_acl="0")
@ -76,9 +76,14 @@ class TestApeFilters(ClusterTestBase):
with reporter.step("Wait for one block"):
self.wait_for_blocks()
objects_with_header, objects_with_other_header, objects_without_header = self._fill_container(default_wallet, file_path, cid)
return cid
return cid, objects_with_header, objects_with_other_header, objects_without_header, file_path
@pytest.fixture(scope="function")
def container_with_objects(self, private_container: str, default_wallet: WalletInfo, file_path: TestFile):
objects_with_header, objects_with_other_header, objects_without_header = self._fill_container(
default_wallet, file_path, private_container
)
return private_container, objects_with_header, objects_with_other_header, objects_without_header, file_path
@reporter.step("Add objects to container")
def _fill_container(self, wallet: WalletInfo, test_file: TestFile, cid: str):
@ -372,3 +377,57 @@ class TestApeFilters(ClusterTestBase):
with expect_not_raises():
put_object_to_random_node(other_wallet, file_path, cid, self.shell, self.cluster, bearer, attributes=allow_attribute)
@allure.title("PUT and GET object using bearer with objectID in filter (obj_size={object_size}, match_type=NOT_EQUAL)")
def test_ape_filter_object_id_not_equals(
self,
frostfs_cli: FrostfsCli,
default_wallet: WalletInfo,
other_wallet: WalletInfo,
private_container: str,
temp_directory: str,
file_path: TestFile,
):
with reporter.step("Put object to container"):
oid = put_object_to_random_node(default_wallet, file_path, private_container, self.shell, self.cluster)
with reporter.step("Create bearer token with objectID filter"):
role_condition = ape.Condition.by_role(ape.Role.OTHERS)
object_condition = ape.Condition.by_object_id(oid, ape.ConditionType.RESOURCE, ape.MatchType.NOT_EQUAL)
rule = ape.Rule(ape.Verb.ALLOW, ALL_OBJECT_OPERATIONS, [role_condition, object_condition])
bearer = create_bearer_token(frostfs_cli, temp_directory, private_container, rule, self.cluster.default_rpc_endpoint)
with reporter.step("Others should be able to put object using bearer token"):
with expect_not_raises():
put_object_to_random_node(other_wallet, file_path, private_container, self.shell, self.cluster, bearer)
with reporter.step("Others should not be able to get object matching the filter"):
with pytest.raises(Exception, match=OBJECT_NO_ACCESS):
get_object_from_random_node(other_wallet, private_container, oid, self.shell, self.cluster, bearer)
@allure.title("PUT and GET object using bearer with objectID in filter (obj_size={object_size}, match_type=EQUAL)")
def test_ape_filter_object_id_equals(
self,
frostfs_cli: FrostfsCli,
default_wallet: WalletInfo,
other_wallet: WalletInfo,
private_container: str,
temp_directory: str,
file_path: TestFile,
):
with reporter.step("Put object to container"):
oid = put_object_to_random_node(default_wallet, file_path, private_container, self.shell, self.cluster)
with reporter.step("Create bearer token with objectID filter"):
role_condition = ape.Condition.by_role(ape.Role.OTHERS)
object_condition = ape.Condition.by_object_id(oid, ape.ConditionType.RESOURCE, ape.MatchType.EQUAL)
rule = ape.Rule(ape.Verb.ALLOW, ALL_OBJECT_OPERATIONS, [role_condition, object_condition])
bearer = create_bearer_token(frostfs_cli, temp_directory, private_container, rule, self.cluster.default_rpc_endpoint)
with reporter.step("Others should not be able to put object using bearer token"):
with pytest.raises(Exception, match=OBJECT_NO_ACCESS):
put_object_to_random_node(other_wallet, file_path, private_container, self.shell, self.cluster, bearer)
with reporter.step("Others should be able to get object matching the filter"):
with expect_not_raises():
get_object_from_random_node(other_wallet, private_container, oid, self.shell, self.cluster, bearer)