forked from TrueCloudLab/frostfs-testcases
Compare commits
1 commit
master
...
EliChin/fe
| Author | SHA1 | Date | |
|---|---|---|---|
| cabe3e7e84 |
3 changed files with 4 additions and 613 deletions
|
|
@ -1,473 +0,0 @@
|
|||
import logging
|
||||
import os
|
||||
import uuid
|
||||
from typing import Optional
|
||||
|
||||
import allure
|
||||
import pytest
|
||||
from frostfs_testlib.resources.common import OBJECT_ACCESS_DENIED, OBJECT_NOT_FOUND
|
||||
from frostfs_testlib.utils import wallet_utils
|
||||
|
||||
from pytest_tests.helpers.acl import (
|
||||
EACLAccess,
|
||||
EACLOperation,
|
||||
EACLRole,
|
||||
EACLRule,
|
||||
create_eacl,
|
||||
form_bearertoken_file,
|
||||
set_eacl,
|
||||
)
|
||||
from pytest_tests.helpers.container import create_container
|
||||
from pytest_tests.helpers.file_helper import generate_file
|
||||
from pytest_tests.helpers.frostfs_verbs import put_object_to_random_node
|
||||
from pytest_tests.helpers.payment_neogo import deposit_gas, transfer_gas
|
||||
from pytest_tests.helpers.storage_group import (
|
||||
delete_storagegroup,
|
||||
get_storagegroup,
|
||||
list_storagegroup,
|
||||
put_storagegroup,
|
||||
verify_get_storage_group,
|
||||
verify_list_storage_group,
|
||||
)
|
||||
from pytest_tests.resources.common import ASSETS_DIR, FREE_STORAGE, WALLET_PASS
|
||||
from pytest_tests.steps.cluster_test_base import ClusterTestBase
|
||||
|
||||
logger = logging.getLogger("NeoLogger")
|
||||
deposit = 30
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"object_size",
|
||||
[pytest.lazy_fixture("simple_object_size"), pytest.lazy_fixture("complex_object_size")],
|
||||
ids=["simple object", "complex object"],
|
||||
)
|
||||
@pytest.mark.sanity
|
||||
@pytest.mark.acl
|
||||
@pytest.mark.storage_group
|
||||
class TestStorageGroup(ClusterTestBase):
|
||||
@pytest.fixture(autouse=True)
|
||||
def prepare_two_wallets(self, default_wallet):
|
||||
self.main_wallet = default_wallet
|
||||
self.other_wallet = os.path.join(os.getcwd(), ASSETS_DIR, f"{str(uuid.uuid4())}.json")
|
||||
wallet_utils.init_wallet(self.other_wallet, WALLET_PASS)
|
||||
if not FREE_STORAGE:
|
||||
main_chain = self.cluster.main_chain_nodes[0]
|
||||
deposit = 30
|
||||
transfer_gas(
|
||||
shell=self.shell,
|
||||
amount=deposit + 1,
|
||||
main_chain=main_chain,
|
||||
wallet_to_path=self.other_wallet,
|
||||
wallet_to_password=WALLET_PASS,
|
||||
)
|
||||
deposit_gas(
|
||||
shell=self.shell,
|
||||
amount=deposit,
|
||||
main_chain=main_chain,
|
||||
wallet_from_path=self.other_wallet,
|
||||
wallet_from_password=WALLET_PASS,
|
||||
)
|
||||
|
||||
@allure.title("Test Storage Group in Private Container")
|
||||
def test_storagegroup_basic_private_container(self, object_size, max_object_size):
|
||||
cid = create_container(
|
||||
self.main_wallet, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint
|
||||
)
|
||||
file_path = generate_file(object_size)
|
||||
oid = put_object_to_random_node(self.main_wallet, file_path, cid, self.shell, self.cluster)
|
||||
objects = [oid]
|
||||
storage_group = put_storagegroup(
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
wallet=self.main_wallet,
|
||||
cid=cid,
|
||||
objects=objects,
|
||||
)
|
||||
|
||||
self.expect_success_for_storagegroup_operations(
|
||||
wallet=self.main_wallet,
|
||||
cid=cid,
|
||||
obj_list=objects,
|
||||
object_size=object_size,
|
||||
max_object_size=max_object_size,
|
||||
)
|
||||
self.expect_failure_for_storagegroup_operations(
|
||||
wallet=self.other_wallet,
|
||||
cid=cid,
|
||||
obj_list=objects,
|
||||
gid=storage_group,
|
||||
)
|
||||
self.storagegroup_operations_by_system_ro_container(
|
||||
wallet=self.main_wallet,
|
||||
cid=cid,
|
||||
obj_list=objects,
|
||||
object_size=object_size,
|
||||
max_object_size=max_object_size,
|
||||
)
|
||||
|
||||
@allure.title("Test Storage Group in Public Container")
|
||||
def test_storagegroup_basic_public_container(self, object_size, max_object_size):
|
||||
cid = create_container(
|
||||
self.main_wallet,
|
||||
basic_acl="public-read-write",
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
)
|
||||
file_path = generate_file(object_size)
|
||||
oid = put_object_to_random_node(
|
||||
self.main_wallet, file_path, cid, shell=self.shell, cluster=self.cluster
|
||||
)
|
||||
objects = [oid]
|
||||
self.expect_success_for_storagegroup_operations(
|
||||
wallet=self.main_wallet,
|
||||
cid=cid,
|
||||
obj_list=objects,
|
||||
object_size=object_size,
|
||||
max_object_size=max_object_size,
|
||||
)
|
||||
self.expect_success_for_storagegroup_operations(
|
||||
wallet=self.other_wallet,
|
||||
cid=cid,
|
||||
obj_list=objects,
|
||||
object_size=object_size,
|
||||
max_object_size=max_object_size,
|
||||
)
|
||||
self.storagegroup_operations_by_system_ro_container(
|
||||
wallet=self.main_wallet,
|
||||
cid=cid,
|
||||
obj_list=objects,
|
||||
object_size=object_size,
|
||||
max_object_size=max_object_size,
|
||||
)
|
||||
|
||||
@allure.title("Test Storage Group in Read-Only Container")
|
||||
def test_storagegroup_basic_ro_container(self, object_size, max_object_size):
|
||||
cid = create_container(
|
||||
self.main_wallet,
|
||||
basic_acl="public-read",
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
)
|
||||
file_path = generate_file(object_size)
|
||||
oid = put_object_to_random_node(
|
||||
self.main_wallet, file_path, cid, shell=self.shell, cluster=self.cluster
|
||||
)
|
||||
objects = [oid]
|
||||
self.expect_success_for_storagegroup_operations(
|
||||
wallet=self.main_wallet,
|
||||
cid=cid,
|
||||
obj_list=objects,
|
||||
object_size=object_size,
|
||||
max_object_size=max_object_size,
|
||||
)
|
||||
self.storagegroup_operations_by_other_ro_container(
|
||||
owner_wallet=self.main_wallet,
|
||||
other_wallet=self.other_wallet,
|
||||
cid=cid,
|
||||
obj_list=objects,
|
||||
object_size=object_size,
|
||||
max_object_size=max_object_size,
|
||||
)
|
||||
self.storagegroup_operations_by_system_ro_container(
|
||||
wallet=self.main_wallet,
|
||||
cid=cid,
|
||||
obj_list=objects,
|
||||
object_size=object_size,
|
||||
max_object_size=max_object_size,
|
||||
)
|
||||
|
||||
@allure.title("Test Storage Group with Bearer Allow")
|
||||
def test_storagegroup_bearer_allow(self, object_size, max_object_size):
|
||||
cid = create_container(
|
||||
self.main_wallet,
|
||||
basic_acl="eacl-public-read-write",
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
)
|
||||
file_path = generate_file(object_size)
|
||||
oid = put_object_to_random_node(
|
||||
self.main_wallet, file_path, cid, shell=self.shell, cluster=self.cluster
|
||||
)
|
||||
objects = [oid]
|
||||
self.expect_success_for_storagegroup_operations(
|
||||
wallet=self.main_wallet,
|
||||
cid=cid,
|
||||
obj_list=objects,
|
||||
object_size=object_size,
|
||||
max_object_size=max_object_size,
|
||||
)
|
||||
storage_group = put_storagegroup(
|
||||
self.shell, self.cluster.default_rpc_endpoint, self.main_wallet, cid, objects
|
||||
)
|
||||
eacl_deny = [
|
||||
EACLRule(access=EACLAccess.DENY, role=role, operation=op)
|
||||
for op in EACLOperation
|
||||
for role in EACLRole
|
||||
]
|
||||
set_eacl(
|
||||
self.main_wallet,
|
||||
cid,
|
||||
create_eacl(cid, eacl_deny, shell=self.shell),
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
)
|
||||
self.expect_failure_for_storagegroup_operations(
|
||||
self.main_wallet, cid, objects, storage_group
|
||||
)
|
||||
bearer_file = form_bearertoken_file(
|
||||
self.main_wallet,
|
||||
cid,
|
||||
[
|
||||
EACLRule(operation=op, access=EACLAccess.ALLOW, role=EACLRole.USER)
|
||||
for op in EACLOperation
|
||||
],
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
)
|
||||
self.expect_success_for_storagegroup_operations(
|
||||
wallet=self.main_wallet,
|
||||
cid=cid,
|
||||
obj_list=objects,
|
||||
object_size=object_size,
|
||||
max_object_size=max_object_size,
|
||||
bearer=bearer_file,
|
||||
)
|
||||
|
||||
@allure.title("Test to check Storage Group lifetime")
|
||||
def test_storagegroup_lifetime(self, object_size):
|
||||
cid = create_container(
|
||||
self.main_wallet, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint
|
||||
)
|
||||
file_path = generate_file(object_size)
|
||||
oid = put_object_to_random_node(
|
||||
self.main_wallet, file_path, cid, shell=self.shell, cluster=self.cluster
|
||||
)
|
||||
objects = [oid]
|
||||
storage_group = put_storagegroup(
|
||||
self.shell,
|
||||
self.cluster.default_rpc_endpoint,
|
||||
self.main_wallet,
|
||||
cid,
|
||||
objects,
|
||||
lifetime=1,
|
||||
)
|
||||
with allure.step("Tick two epochs"):
|
||||
for _ in range(2):
|
||||
self.tick_epoch()
|
||||
self.wait_for_epochs_align()
|
||||
with pytest.raises(Exception, match=OBJECT_NOT_FOUND):
|
||||
get_storagegroup(
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
wallet=self.main_wallet,
|
||||
cid=cid,
|
||||
gid=storage_group,
|
||||
)
|
||||
|
||||
@allure.step("Run Storage Group Operations And Expect Success")
|
||||
def expect_success_for_storagegroup_operations(
|
||||
self,
|
||||
wallet: str,
|
||||
cid: str,
|
||||
obj_list: list,
|
||||
object_size: int,
|
||||
max_object_size: int,
|
||||
bearer: Optional[str] = None,
|
||||
):
|
||||
"""
|
||||
This func verifies if the Object's owner is allowed to
|
||||
Put, List, Get and Delete the Storage Group which contains
|
||||
the Object.
|
||||
"""
|
||||
storage_group = put_storagegroup(
|
||||
self.shell, self.cluster.default_rpc_endpoint, wallet, cid, obj_list, bearer
|
||||
)
|
||||
verify_list_storage_group(
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
wallet=wallet,
|
||||
cid=cid,
|
||||
gid=storage_group,
|
||||
bearer=bearer,
|
||||
)
|
||||
verify_get_storage_group(
|
||||
shell=self.shell,
|
||||
cluster=self.cluster,
|
||||
wallet=wallet,
|
||||
cid=cid,
|
||||
gid=storage_group,
|
||||
obj_list=obj_list,
|
||||
object_size=object_size,
|
||||
max_object_size=max_object_size,
|
||||
bearer=bearer,
|
||||
)
|
||||
delete_storagegroup(
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
wallet=wallet,
|
||||
cid=cid,
|
||||
gid=storage_group,
|
||||
bearer=bearer,
|
||||
)
|
||||
|
||||
@allure.step("Run Storage Group Operations And Expect Failure")
|
||||
def expect_failure_for_storagegroup_operations(
|
||||
self, wallet: str, cid: str, obj_list: list, gid: str
|
||||
):
|
||||
"""
|
||||
This func verifies if the Object's owner isn't allowed to
|
||||
Put, List, Get and Delete the Storage Group which contains
|
||||
the Object.
|
||||
"""
|
||||
with pytest.raises(Exception, match=OBJECT_ACCESS_DENIED):
|
||||
put_storagegroup(
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
wallet=wallet,
|
||||
cid=cid,
|
||||
objects=obj_list,
|
||||
)
|
||||
with pytest.raises(Exception, match=OBJECT_ACCESS_DENIED):
|
||||
list_storagegroup(
|
||||
shell=self.shell, endpoint=self.cluster.default_rpc_endpoint, wallet=wallet, cid=cid
|
||||
)
|
||||
with pytest.raises(Exception, match=OBJECT_ACCESS_DENIED):
|
||||
get_storagegroup(
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
wallet=wallet,
|
||||
cid=cid,
|
||||
gid=gid,
|
||||
)
|
||||
with pytest.raises(Exception, match=OBJECT_ACCESS_DENIED):
|
||||
delete_storagegroup(
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
wallet=wallet,
|
||||
cid=cid,
|
||||
gid=gid,
|
||||
)
|
||||
|
||||
@allure.step("Run Storage Group Operations On Other's Behalf In RO Container")
|
||||
def storagegroup_operations_by_other_ro_container(
|
||||
self,
|
||||
owner_wallet: str,
|
||||
other_wallet: str,
|
||||
cid: str,
|
||||
obj_list: list,
|
||||
object_size: int,
|
||||
max_object_size: int,
|
||||
):
|
||||
storage_group = put_storagegroup(
|
||||
self.shell, self.cluster.default_rpc_endpoint, owner_wallet, cid, obj_list
|
||||
)
|
||||
with pytest.raises(Exception, match=OBJECT_ACCESS_DENIED):
|
||||
put_storagegroup(
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
wallet=other_wallet,
|
||||
cid=cid,
|
||||
objects=obj_list,
|
||||
)
|
||||
verify_list_storage_group(
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
wallet=other_wallet,
|
||||
cid=cid,
|
||||
gid=storage_group,
|
||||
)
|
||||
verify_get_storage_group(
|
||||
shell=self.shell,
|
||||
cluster=self.cluster,
|
||||
wallet=other_wallet,
|
||||
cid=cid,
|
||||
gid=storage_group,
|
||||
obj_list=obj_list,
|
||||
object_size=object_size,
|
||||
max_object_size=max_object_size,
|
||||
)
|
||||
with pytest.raises(Exception, match=OBJECT_ACCESS_DENIED):
|
||||
delete_storagegroup(
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
wallet=other_wallet,
|
||||
cid=cid,
|
||||
gid=storage_group,
|
||||
)
|
||||
|
||||
@allure.step("Run Storage Group Operations On Systems's Behalf In RO Container")
|
||||
def storagegroup_operations_by_system_ro_container(
|
||||
self,
|
||||
wallet: str,
|
||||
cid: str,
|
||||
obj_list: list,
|
||||
object_size: int,
|
||||
max_object_size: int,
|
||||
):
|
||||
"""
|
||||
In this func we create a Storage Group on Inner Ring's key behalf
|
||||
and include an Object created on behalf of some user. We expect
|
||||
that System key is granted to make all operations except PUT and DELETE.
|
||||
"""
|
||||
ir_node = self.cluster.ir_nodes[0]
|
||||
ir_wallet_path = ir_node.get_wallet_path()
|
||||
ir_wallet_password = ir_node.get_wallet_password()
|
||||
ir_wallet_config = ir_node.get_wallet_config_path()
|
||||
|
||||
if not FREE_STORAGE:
|
||||
main_chain = self.cluster.main_chain_nodes[0]
|
||||
deposit = 30
|
||||
transfer_gas(
|
||||
shell=self.shell,
|
||||
amount=deposit + 1,
|
||||
main_chain=main_chain,
|
||||
wallet_to_path=ir_wallet_path,
|
||||
wallet_to_password=ir_wallet_password,
|
||||
)
|
||||
deposit_gas(
|
||||
shell=self.shell,
|
||||
amount=deposit,
|
||||
main_chain=main_chain,
|
||||
wallet_from_path=ir_wallet_path,
|
||||
wallet_from_password=ir_wallet_password,
|
||||
)
|
||||
storage_group = put_storagegroup(
|
||||
self.shell, self.cluster.default_rpc_endpoint, wallet, cid, obj_list
|
||||
)
|
||||
with pytest.raises(Exception, match=OBJECT_ACCESS_DENIED):
|
||||
put_storagegroup(
|
||||
self.shell,
|
||||
self.cluster.default_rpc_endpoint,
|
||||
ir_wallet_path,
|
||||
cid,
|
||||
obj_list,
|
||||
wallet_config=ir_wallet_config,
|
||||
)
|
||||
verify_list_storage_group(
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
wallet=ir_wallet_path,
|
||||
cid=cid,
|
||||
gid=storage_group,
|
||||
wallet_config=ir_wallet_config,
|
||||
)
|
||||
verify_get_storage_group(
|
||||
shell=self.shell,
|
||||
cluster=self.cluster,
|
||||
wallet=ir_wallet_path,
|
||||
cid=cid,
|
||||
gid=storage_group,
|
||||
obj_list=obj_list,
|
||||
object_size=object_size,
|
||||
max_object_size=max_object_size,
|
||||
wallet_config=ir_wallet_config,
|
||||
)
|
||||
with pytest.raises(Exception, match=OBJECT_ACCESS_DENIED):
|
||||
delete_storagegroup(
|
||||
shell=self.shell,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
wallet=ir_wallet_path,
|
||||
cid=cid,
|
||||
gid=storage_group,
|
||||
wallet_config=ir_wallet_config,
|
||||
)
|
||||
|
|
@ -1,102 +0,0 @@
|
|||
import logging
|
||||
import os
|
||||
|
||||
import allure
|
||||
import pytest
|
||||
import yaml
|
||||
from frostfs_testlib.cli import FrostfsCli
|
||||
from frostfs_testlib.shell import CommandResult, Shell
|
||||
|
||||
from pytest_tests.helpers.wallet import WalletFactory, WalletFile
|
||||
from pytest_tests.resources.common import FREE_STORAGE, FROSTFS_CLI_EXEC, WALLET_CONFIG
|
||||
from pytest_tests.steps.cluster_test_base import ClusterTestBase
|
||||
|
||||
logger = logging.getLogger("NeoLogger")
|
||||
DEPOSIT_AMOUNT = 30
|
||||
|
||||
|
||||
@pytest.mark.sanity
|
||||
@pytest.mark.payments
|
||||
@pytest.mark.skipif(FREE_STORAGE, reason="Test only works on public network with paid storage")
|
||||
class TestBalanceAccounting(ClusterTestBase):
|
||||
@pytest.fixture(scope="class")
|
||||
def main_wallet(self, wallet_factory: WalletFactory) -> WalletFile:
|
||||
return wallet_factory.create_wallet()
|
||||
|
||||
@pytest.fixture(scope="class")
|
||||
def other_wallet(self, wallet_factory: WalletFactory) -> WalletFile:
|
||||
return wallet_factory.create_wallet()
|
||||
|
||||
@pytest.fixture(scope="class")
|
||||
def cli(self, client_shell: Shell) -> FrostfsCli:
|
||||
return FrostfsCli(client_shell, FROSTFS_CLI_EXEC, WALLET_CONFIG)
|
||||
|
||||
@allure.step("Check deposit amount")
|
||||
def check_amount(self, result: CommandResult) -> None:
|
||||
amount_str = result.stdout.rstrip()
|
||||
|
||||
try:
|
||||
amount = int(amount_str)
|
||||
except Exception as ex:
|
||||
pytest.fail(
|
||||
f"Amount parse error, should be parsable as int({DEPOSIT_AMOUNT}), but given {amount_str}: {ex}"
|
||||
)
|
||||
|
||||
assert amount == DEPOSIT_AMOUNT
|
||||
|
||||
@staticmethod
|
||||
@allure.step("Write config with API endpoint")
|
||||
def write_api_config(config_dir: str, endpoint: str, wallet: str) -> str:
|
||||
with open(WALLET_CONFIG, "r") as file:
|
||||
wallet_config = yaml.full_load(file)
|
||||
api_config = {
|
||||
**wallet_config,
|
||||
"rpc-endpoint": endpoint,
|
||||
"wallet": wallet,
|
||||
}
|
||||
api_config_file = os.path.join(config_dir, "frostfs-cli-api-config.yaml")
|
||||
with open(api_config_file, "w") as file:
|
||||
yaml.dump(api_config, file)
|
||||
return api_config_file
|
||||
|
||||
@allure.title("Test balance request with wallet and address")
|
||||
def test_balance_wallet_address(self, main_wallet: WalletFile, cli: FrostfsCli):
|
||||
result = cli.accounting.balance(
|
||||
wallet=main_wallet.path,
|
||||
rpc_endpoint=self.cluster.default_rpc_endpoint,
|
||||
address=main_wallet.get_address(),
|
||||
)
|
||||
|
||||
self.check_amount(result)
|
||||
|
||||
@allure.title("Test balance request with wallet only")
|
||||
def test_balance_wallet(self, main_wallet: WalletFile, cli: FrostfsCli):
|
||||
result = cli.accounting.balance(
|
||||
wallet=main_wallet.path, rpc_endpoint=self.cluster.default_rpc_endpoint
|
||||
)
|
||||
self.check_amount(result)
|
||||
|
||||
@allure.title("Test balance request with wallet and wrong address")
|
||||
def test_balance_wrong_address(
|
||||
self, main_wallet: WalletFile, other_wallet: WalletFile, cli: FrostfsCli
|
||||
):
|
||||
with pytest.raises(Exception, match="address option must be specified and valid"):
|
||||
cli.accounting.balance(
|
||||
wallet=main_wallet.path,
|
||||
rpc_endpoint=self.cluster.default_rpc_endpoint,
|
||||
address=other_wallet.get_address(),
|
||||
)
|
||||
|
||||
@allure.title("Test balance request with config file")
|
||||
def test_balance_api(self, temp_directory: str, main_wallet: WalletFile, client_shell: Shell):
|
||||
config_file = self.write_api_config(
|
||||
config_dir=temp_directory,
|
||||
endpoint=self.cluster.default_rpc_endpoint,
|
||||
wallet=main_wallet.path,
|
||||
)
|
||||
logger.info(f"Config with API endpoint: {config_file}")
|
||||
|
||||
cli = FrostfsCli(client_shell, FROSTFS_CLI_EXEC, config_file=config_file)
|
||||
result = cli.accounting.balance()
|
||||
|
||||
self.check_amount(result)
|
||||
|
|
@ -6,17 +6,10 @@ from random import choices, sample
|
|||
|
||||
import allure
|
||||
import pytest
|
||||
from frostfs_testlib.utils import wallet_utils
|
||||
|
||||
from pytest_tests.helpers.aws_cli_client import AwsCliClient
|
||||
from pytest_tests.helpers.file_helper import (
|
||||
concat_files,
|
||||
generate_file,
|
||||
generate_file_with_content,
|
||||
get_file_hash,
|
||||
)
|
||||
from pytest_tests.helpers.payment_neogo import deposit_gas, transfer_gas
|
||||
from pytest_tests.helpers.s3_helper import (
|
||||
from aws_cli_client import AwsCliClient
|
||||
from common import ASSETS_DIR
|
||||
from file_helper import concat_files, generate_file, generate_file_with_content, get_file_hash
|
||||
from s3_helper import (
|
||||
assert_object_lock_mode,
|
||||
assert_s3_acl,
|
||||
check_objects_in_bucket,
|
||||
|
|
@ -662,37 +655,10 @@ class TestS3GateObject(TestS3GateBase):
|
|||
{"Key": tag_key_3, "Value": str(tag_value_3)}
|
||||
], "Tags must be the same"
|
||||
|
||||
@pytest.fixture
|
||||
def prepare_two_wallets(self, default_wallet, client_shell):
|
||||
self.main_wallet = default_wallet
|
||||
self.main_public_key = wallet_utils.get_wallet_public_key(self.main_wallet, WALLET_PASS)
|
||||
self.other_wallet = os.path.join(os.getcwd(), ASSETS_DIR, f"{str(uuid.uuid4())}.json")
|
||||
wallet_utils.init_wallet(self.other_wallet, WALLET_PASS)
|
||||
self.other_public_key = wallet_utils.get_wallet_public_key(self.other_wallet, WALLET_PASS)
|
||||
|
||||
if not FREE_STORAGE:
|
||||
main_chain = self.cluster.main_chain_nodes[0]
|
||||
deposit = 30
|
||||
transfer_gas(
|
||||
shell=client_shell,
|
||||
amount=deposit + 1,
|
||||
main_chain=main_chain,
|
||||
wallet_to_path=self.other_wallet,
|
||||
wallet_to_password=WALLET_PASS,
|
||||
)
|
||||
deposit_gas(
|
||||
shell=client_shell,
|
||||
main_chain=main_chain,
|
||||
amount=deposit,
|
||||
wallet_from_path=self.other_wallet,
|
||||
wallet_from_password=WALLET_PASS,
|
||||
)
|
||||
|
||||
@allure.title("Test S3: put object with ACL")
|
||||
@pytest.mark.parametrize("bucket_versioning", ["ENABLED", "SUSPENDED"])
|
||||
def test_s3_put_object_acl(
|
||||
self,
|
||||
prepare_two_wallets,
|
||||
bucket_versioning,
|
||||
bucket,
|
||||
complex_object_size,
|
||||
|
|
|
|||
Loading…
Reference in a new issue