diff --git a/pytest_tests/testsuites/ape/test_ape_bearer.py b/pytest_tests/testsuites/ape/test_ape_bearer.py new file mode 100644 index 00000000..e435d9d9 --- /dev/null +++ b/pytest_tests/testsuites/ape/test_ape_bearer.py @@ -0,0 +1,80 @@ +import os +import time + +import allure +import pytest +from frostfs_testlib import reporter +from frostfs_testlib.credentials.interfaces import User +from frostfs_testlib.resources.common import MORPH_BLOCK_TIME +from frostfs_testlib.resources.error_patterns import NO_RULE_FOUND_OBJECT, RULE_ACCESS_DENIED_OBJECT +from frostfs_testlib.steps.cli.object import put_object +from frostfs_testlib.storage.dataclasses.ape import Operations +from frostfs_testlib.storage.dataclasses.object_size import ObjectSize +from frostfs_testlib.storage.grpc_operations.client_wrappers import CliClientWrapper +from frostfs_testlib.testing.cluster_test_base import ClusterTestBase +from frostfs_testlib.testing.test_control import expect_not_raises +from frostfs_testlib.utils import datetime_utils +from frostfs_testlib.utils.file_utils import generate_file +from frostfs_testlib.utils.string_utils import unique_name +from frostfs_testlib_plugin_to.storage.capi.capi_client import CApiClient + + +@pytest.mark.ape +@pytest.mark.ape_bearer +class TestApeBearer(ClusterTestBase): + @allure.title("BT with container basic-acl=0 root") + def test_BT_with_container_basic_acl_0_root( + self, + grpc_client: CliClientWrapper, + simple_object_size: ObjectSize, + temp_directory: str, + default_user: User, + ): + test_file = generate_file(simple_object_size.value) + + with reporter.step("Create container"): + cid = grpc_client.container.create( + endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(), + policy="REP 2 IN X CBF 1 SELECT 2 FROM * AS X", + await_mode=True, + basic_acl="0", + ) + + chain_file = os.path.join(temp_directory, "serialized_chain.json") + bt_file = os.path.join(temp_directory, "bt.json") + bt_sign_file = os.path.join(temp_directory, "bt-sign.json") + + with reporter.step("Generate APE override by target and APE chains"): + grpc_client.cli.bearer.generate_ape_override( + rule="allow Object.Put *", + chain_id="allowPutObjBT", + cid=cid, + output=chain_file, + ) + + with reporter.step("Create bearer token"): + grpc_client.cli.bearer.create( + rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(), + ape=chain_file, + issued_at=2, + expire_at=1000, + out=bt_file, + ) + + with reporter.step("Sign bearer token"): + grpc_client.cli.util.sign_bearer_token(from_file=bt_file, to_file=bt_sign_file) + + with reporter.step("[NEGATIVE] Put object without token"): + with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT.format(operation=Operations.PUT_OBJECT.value)): + put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()) + + with reporter.step("Put object with bearer token"): + with expect_not_raises(): + put_object( + default_user.wallet, + test_file, + cid, + self.shell, + self.cluster.storage_nodes[0].get_rpc_endpoint(), + bearer=bt_sign_file, + )