diff --git a/pytest_tests/testsuites/access/ape/test_ape_filters.py b/pytest_tests/testsuites/access/ape/test_ape_filters.py index f8d291e..ddbf709 100644 --- a/pytest_tests/testsuites/access/ape/test_ape_filters.py +++ b/pytest_tests/testsuites/access/ape/test_ape_filters.py @@ -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)