import allure import pytest from frostfs_testlib.resources.wellknown_acl import EACL_PUBLIC_READ_WRITE from frostfs_testlib.shell import Shell from frostfs_testlib.steps.acl import form_bearertoken_file from frostfs_testlib.steps.cli.container import ( REP_2_FOR_3_NODES_PLACEMENT_RULE, SINGLE_PLACEMENT_RULE, StorageContainer, StorageContainerInfo, create_container, ) from frostfs_testlib.steps.cli.object import delete_object, get_object from frostfs_testlib.steps.epoch import get_epoch from frostfs_testlib.steps.storage_object import StorageObjectInfo from frostfs_testlib.storage.cluster import Cluster from frostfs_testlib.storage.dataclasses.acl import EACLAccess, EACLOperation, EACLRole, EACLRule from frostfs_testlib.storage.dataclasses.object_size import ObjectSize from frostfs_testlib.storage.dataclasses.wallet import WalletInfo from frostfs_testlib.testing.cluster_test_base import ClusterTestBase from frostfs_testlib.testing.test_control import expect_not_raises from pytest import FixtureRequest @pytest.fixture(scope="module") @allure.title("Create bearer token for OTHERS with all operations allowed for all containers") def bearer_token_file_all_allow(default_wallet: str, client_shell: Shell, cluster: Cluster) -> str: bearer = form_bearertoken_file( default_wallet, "", [ EACLRule(operation=op, access=EACLAccess.ALLOW, role=EACLRole.OTHERS) for op in EACLOperation ], shell=client_shell, endpoint=cluster.default_rpc_endpoint, ) return bearer @pytest.fixture(scope="module") @allure.title("Create user container for bearer token usage") def user_container( default_wallet: str, client_shell: Shell, cluster: Cluster, request: FixtureRequest ) -> StorageContainer: container_id = create_container( default_wallet, shell=client_shell, rule=request.param, basic_acl=EACL_PUBLIC_READ_WRITE, endpoint=cluster.default_rpc_endpoint, ) # Deliberately using s3gate wallet here to test bearer token s3gate = cluster.s3_gates[0] return StorageContainer( StorageContainerInfo(container_id, WalletInfo.from_node(s3gate)), client_shell, cluster, ) @pytest.fixture() def storage_objects( user_container: StorageContainer, bearer_token_file_all_allow: str, request: FixtureRequest, client_shell: Shell, cluster: Cluster, ) -> list[StorageObjectInfo]: epoch = get_epoch(client_shell, cluster) storage_objects: list[StorageObjectInfo] = [] object_size: ObjectSize = request.param for node in cluster.storage_nodes: storage_objects.append( user_container.generate_object( object_size.value, epoch + 3, bearer_token=bearer_token_file_all_allow, endpoint=node.get_rpc_endpoint(), ) ) return storage_objects @pytest.mark.smoke @pytest.mark.bearer class TestObjectApiWithBearerToken(ClusterTestBase): @allure.title( "Object can be deleted from any node using s3gate wallet with bearer token for {storage_objects}" ) @pytest.mark.parametrize( "storage_objects,user_container", [ (pytest.lazy_fixture("simple_object_size"), SINGLE_PLACEMENT_RULE), (pytest.lazy_fixture("complex_object_size"), SINGLE_PLACEMENT_RULE), ], ids=["simple object size", "complex object size"], indirect=True, ) def test_delete_object_with_s3_wallet_bearer( self, storage_objects: list[StorageObjectInfo], bearer_token_file_all_allow: str, request: FixtureRequest, ): allure.dynamic.title( f"Object can be deleted from any node using s3gate wallet with bearer token for {request.node.callspec.id}" ) s3_gate_wallet = self.cluster.s3_gates[0] with allure.step("Try to delete each object from first storage node"): for storage_object in storage_objects: with expect_not_raises(): delete_object( s3_gate_wallet.get_wallet_path(), storage_object.cid, storage_object.oid, self.shell, endpoint=self.cluster.default_rpc_endpoint, bearer=bearer_token_file_all_allow, wallet_config=s3_gate_wallet.get_wallet_config_path(), ) @allure.title( "Object can be fetched from any node using s3gate wallet with bearer token for {object_size}" ) @pytest.mark.parametrize( "object_size, user_container", [ (pytest.lazy_fixture("simple_object_size"), REP_2_FOR_3_NODES_PLACEMENT_RULE), (pytest.lazy_fixture("complex_object_size"), REP_2_FOR_3_NODES_PLACEMENT_RULE), ], ids=["simple object size", "complex object size"], indirect=["user_container"], ) def test_get_object_with_s3_wallet_bearer_from_all_nodes( self, user_container: StorageContainer, object_size: ObjectSize, bearer_token_file_all_allow: str, ): s3_gate_wallet = self.cluster.s3_gates[0] with allure.step("Put one object to container"): epoch = self.get_epoch() storage_object = user_container.generate_object( object_size.value, epoch + 3, bearer_token=bearer_token_file_all_allow ) with allure.step("Try to fetch object from each storage node"): for node in self.cluster.storage_nodes: with expect_not_raises(): get_object( s3_gate_wallet.get_wallet_path(), storage_object.cid, storage_object.oid, self.shell, endpoint=node.get_rpc_endpoint(), bearer=bearer_token_file_all_allow, wallet_config=s3_gate_wallet.get_wallet_config_path(), )