frostfs-testcases/pytest_tests/testsuites/container/test_container.py
Andrey Berezin c77123f301 Move shared code to testlib
Signed-off-by: Andrey Berezin <a.berezin@yadro.com>
2023-05-15 12:59:33 +03:00

125 lines
4.6 KiB
Python

import json
import allure
import pytest
from frostfs_testlib.resources.wellknown_acl import PRIVATE_ACL_F
from frostfs_testlib.steps.cli.container import (
create_container,
delete_container,
get_container,
list_containers,
wait_for_container_creation,
wait_for_container_deletion,
)
from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
from pytest_tests.helpers.utility import placement_policy_from_container
@pytest.mark.container
@pytest.mark.sanity
class TestContainer(ClusterTestBase):
@pytest.mark.parametrize("name", ["", "test-container"], ids=["No name", "Set particular name"])
@pytest.mark.smoke
def test_container_creation(self, default_wallet: str, name: str):
scenario_title = f"with name {name}" if name else "without name"
allure.dynamic.title(f"User can create container {scenario_title}")
wallet = default_wallet
with open(wallet) as file:
json_wallet = json.load(file)
placement_rule = "REP 2 IN X CBF 1 SELECT 2 FROM * AS X"
cid = create_container(
wallet,
rule=placement_rule,
name=name,
shell=self.shell,
endpoint=self.cluster.default_rpc_endpoint,
)
containers = list_containers(
wallet, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint
)
assert cid in containers, f"Expected container {cid} in containers: {containers}"
container_info: str = get_container(
wallet,
cid,
json_mode=False,
shell=self.shell,
endpoint=self.cluster.default_rpc_endpoint,
)
container_info = (
container_info.casefold()
) # To ignore case when comparing with expected values
info_to_check = {
f"basic ACL: {PRIVATE_ACL_F} (private)",
f"owner ID: {json_wallet.get('accounts')[0].get('address')}",
f"container ID: {cid}",
}
if name:
info_to_check.add(f"Name={name}")
with allure.step("Check container has correct information"):
expected_policy = placement_rule.casefold()
actual_policy = placement_policy_from_container(container_info)
assert (
actual_policy == expected_policy
), f"Expected policy\n{expected_policy} but got policy\n{actual_policy}"
for info in info_to_check:
expected_info = info.casefold()
assert (
expected_info in container_info
), f"Expected {expected_info} in container info:\n{container_info}"
with allure.step("Delete container and check it was deleted"):
delete_container(
wallet, cid, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint
)
self.tick_epoch()
wait_for_container_deletion(
wallet, cid, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint
)
@allure.title("Parallel container creation and deletion")
def test_container_creation_deletion_parallel(self, default_wallet: str):
containers_count = 3
wallet = default_wallet
placement_rule = "REP 2 IN X CBF 1 SELECT 2 FROM * AS X"
cids: list[str] = []
with allure.step(f"Create {containers_count} containers"):
for _ in range(containers_count):
cids.append(
create_container(
wallet,
rule=placement_rule,
await_mode=False,
shell=self.shell,
endpoint=self.cluster.default_rpc_endpoint,
wait_for_creation=False,
)
)
with allure.step("Wait for containers occur in container list"):
for cid in cids:
wait_for_container_creation(
wallet,
cid,
sleep_interval=containers_count,
shell=self.shell,
endpoint=self.cluster.default_rpc_endpoint,
)
with allure.step("Delete containers and check they were deleted"):
for cid in cids:
delete_container(
wallet, cid, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint
)
self.tick_epoch()
wait_for_container_deletion(
wallet, cid, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint
)