forked from TrueCloudLab/frostfs-testcases
139 lines
5.2 KiB
Python
139 lines
5.2 KiB
Python
import allure
|
|
import pytest
|
|
from frostfs_testlib import reporter
|
|
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: WalletInfo, 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: WalletInfo, 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,
|
|
object_size: ObjectSize,
|
|
client_shell: Shell,
|
|
cluster: Cluster,
|
|
) -> list[StorageObjectInfo]:
|
|
epoch = get_epoch(client_shell, cluster)
|
|
storage_objects: list[StorageObjectInfo] = []
|
|
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 (obj_size={object_size})")
|
|
@pytest.mark.parametrize(
|
|
"user_container",
|
|
[SINGLE_PLACEMENT_RULE],
|
|
indirect=True,
|
|
)
|
|
def test_delete_object_with_s3_wallet_bearer(
|
|
self,
|
|
storage_objects: list[StorageObjectInfo],
|
|
bearer_token_file_all_allow: str,
|
|
):
|
|
s3_gate_wallet = WalletInfo.from_node(self.cluster.s3_gates[0])
|
|
with reporter.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,
|
|
storage_object.cid,
|
|
storage_object.oid,
|
|
self.shell,
|
|
endpoint=self.cluster.default_rpc_endpoint,
|
|
bearer=bearer_token_file_all_allow,
|
|
)
|
|
|
|
@allure.title("Object can be fetched from any node using s3gate wallet with bearer token (obj_size={object_size})")
|
|
@pytest.mark.parametrize(
|
|
"user_container",
|
|
[REP_2_FOR_3_NODES_PLACEMENT_RULE],
|
|
indirect=True,
|
|
)
|
|
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 = WalletInfo.from_node(self.cluster.s3_gates[0])
|
|
with reporter.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 reporter.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,
|
|
storage_object.cid,
|
|
storage_object.oid,
|
|
self.shell,
|
|
endpoint=node.get_rpc_endpoint(),
|
|
bearer=bearer_token_file_all_allow,
|
|
)
|