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 = "with name" if name else "without name" allure.dynamic.title(f"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, await_mode=True, ) 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() for cid in cids: wait_for_container_deletion(wallet, cid, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint)