[#340] add iam ape test #340
2 changed files with 64 additions and 0 deletions
34
pytest_tests/testsuites/ape/test_ape_iam.py
Normal file
34
pytest_tests/testsuites/ape/test_ape_iam.py
Normal file
|
@ -0,0 +1,34 @@
|
||||||
|
import allure
|
||||||
|
import pytest
|
||||||
|
from frostfs_testlib import reporter
|
||||||
|
from frostfs_testlib.resources.error_patterns import ADD_CHAIN_ERROR
|
||||||
|
from frostfs_testlib.storage.grpc_operations.client_wrappers import CliClientWrapper
|
||||||
|
from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.ape
|
||||||
|
@pytest.mark.ape_iam
|
||||||
|
class TestApeIAM(ClusterTestBase):
|
||||||
|
@allure.title("[NEGATIVE] Create policy using another user wallet")
|
||||||
|
def test_negative_policy_another_user_wallet_frostfs_cli(
|
||||||
|
self,
|
||||||
|
grpc_client: CliClientWrapper,
|
||||||
|
grpc_second_client: CliClientWrapper,
|
||||||
|
):
|
||||||
|
|
||||||
|
with reporter.step("Create container for first user"):
|
||||||
|
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,
|
||||||
|
)
|
||||||
|
|
||||||
|
with reporter.step("[NEGATIVE] Create policy using another user wallet"):
|
||||||
|
with pytest.raises(RuntimeError, match=ADD_CHAIN_ERROR):
|
||||||
|
grpc_second_client.cli.ape_manager.add(
|
||||||
|
rpc_endpoint=self.cluster.default_rpc_endpoint,
|
||||||
|
target_type="container",
|
||||||
|
target_name=f"{cid}",
|
||||||
|
chain_id="allowDeleteObject",
|
||||||
|
rule=f"allow Object.Head Object.Delete kapusta/{cid}/*",
|
||||||
|
)
|
|
@ -210,12 +210,24 @@ def frostfs_cli(client_shell: Shell, default_wallet: WalletInfo) -> FrostfsCli:
|
||||||
return FrostfsCli(client_shell, FROSTFS_CLI_EXEC, default_wallet.config_path)
|
return FrostfsCli(client_shell, FROSTFS_CLI_EXEC, default_wallet.config_path)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="session")
|
||||||
|
@allure.title("Init Frostfs CLI")
|
||||||
|
def frostfs_second_cli(client_shell: Shell, default_second_wallet: WalletInfo) -> FrostfsCli:
|
||||||
|
return FrostfsCli(client_shell, FROSTFS_CLI_EXEC, default_second_wallet.config_path)
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="session")
|
@pytest.fixture(scope="session")
|
||||||
@allure.title("Init GrpcClientWrapper with local Frostfs CLI")
|
@allure.title("Init GrpcClientWrapper with local Frostfs CLI")
|
||||||
def grpc_client(frostfs_cli: FrostfsCli) -> GrpcClientWrapper:
|
def grpc_client(frostfs_cli: FrostfsCli) -> GrpcClientWrapper:
|
||||||
return CliClientWrapper(frostfs_cli)
|
return CliClientWrapper(frostfs_cli)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="session")
|
||||||
|
@allure.title("Init GrpcClientWrapper with local Frostfs CLI")
|
||||||
|
def grpc_second_client(frostfs_second_cli: FrostfsCli) -> GrpcClientWrapper:
|
||||||
|
return CliClientWrapper(frostfs_second_cli)
|
||||||
|
|
||||||
|
|
||||||
# By default we want all tests to be executed with both storage policies.
|
# By default we want all tests to be executed with both storage policies.
|
||||||
# This can be overriden in choosen tests if needed.
|
# This can be overriden in choosen tests if needed.
|
||||||
@pytest.fixture(scope="session", params=[pytest.param("rep", marks=pytest.mark.rep), pytest.param("ec", marks=pytest.mark.ec)])
|
@pytest.fixture(scope="session", params=[pytest.param("rep", marks=pytest.mark.rep), pytest.param("ec", marks=pytest.mark.ec)])
|
||||||
|
@ -443,12 +455,30 @@ def default_user(credentials_provider: CredentialsProvider, cluster: Cluster) ->
|
||||||
return user
|
return user
|
||||||
|
|
||||||
|
|
||||||
|
@reporter.step("Prepare second data user with wallet")
|
||||||
|
@pytest.fixture(scope="session")
|
||||||
|
@cached_fixture(optionals.OPTIONAL_CACHE_FIXTURES)
|
||||||
|
def default_second_user(credentials_provider: CredentialsProvider, cluster: Cluster) -> User:
|
||||||
|
|||||||
|
user = User(string_utils.unique_name("user-"))
|
||||||
|
node = cluster.cluster_nodes[0]
|
||||||
|
|
||||||
|
credentials_provider.GRPC.provide(user, node)
|
||||||
|
|
||||||
|
return user
|
||||||
|
|
||||||
|
|
||||||
@reporter.step("Get wallet for default user")
|
@reporter.step("Get wallet for default user")
|
||||||
@pytest.fixture(scope="session")
|
@pytest.fixture(scope="session")
|
||||||
def default_wallet(default_user: User) -> WalletInfo:
|
def default_wallet(default_user: User) -> WalletInfo:
|
||||||
return default_user.wallet
|
return default_user.wallet
|
||||||
|
|
||||||
|
|
||||||
|
@reporter.step("Get wallet for default user")
|
||||||
|
@pytest.fixture(scope="session")
|
||||||
|
def default_second_wallet(default_second_user: User) -> WalletInfo:
|
||||||
|
return default_second_user.wallet
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="session")
|
@pytest.fixture(scope="session")
|
||||||
@cached_fixture(optionals.OPTIONAL_CACHE_FIXTURES)
|
@cached_fixture(optionals.OPTIONAL_CACHE_FIXTURES)
|
||||||
def wallets_pool(credentials_provider: CredentialsProvider, cluster: Cluster) -> list[WalletInfo]:
|
def wallets_pool(credentials_provider: CredentialsProvider, cluster: Cluster) -> list[WalletInfo]:
|
||||||
|
|
Loading…
Reference in a new issue
remove
default
word