frostfs-testcases/pytest_tests/testsuites/container/test_container.py

import allure
import pytest
from frostfs_testlib import reporter
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.storage.dataclasses.wallet import WalletInfo
from frostfs_testlib.testing.cluster_test_base import ClusterTestBase

from ...helpers.utility import placement_policy_from_container


@pytest.mark.nightly
@pytest.mark.sanity
@pytest.mark.container
class TestContainer(ClusterTestBase):
    PLACEMENT_RULE = "REP 2 IN X CBF 1 SELECT 2 FROM * AS X"

    @allure.title("Create container (name={name})")
    @pytest.mark.parametrize("name", ["", "test-container"], ids=["No name", "Set particular name"])
    @pytest.mark.smoke
    def test_container_creation(self, default_wallet: WalletInfo, name: str, rpc_endpoint: str):
        wallet = default_wallet

        cid = create_container(wallet, self.shell, rpc_endpoint, self.PLACEMENT_RULE, name=name)

        containers = list_containers(wallet, self.shell, rpc_endpoint)
        assert cid in containers, f"Expected container {cid} in containers: {containers}"

        container_info: str = get_container(wallet, cid, self.shell, rpc_endpoint, False)
        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: {wallet.get_address_from_json(0)}",
            f"CID: {cid}",
        }
        if name:
            info_to_check.add(f"Name={name}")

        with reporter.step("Check container has correct information"):
            expected_policy = self.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 reporter.step("Delete container and check it was deleted"):
            # Force to skip frostfs-cli verifictions before delete.
            # Because no APE rules assigned to container, those verifications will fail due to APE requests denial.
            delete_container(wallet, cid, self.shell, rpc_endpoint, force=True, await_mode=True)
            self.tick_epoch()
            wait_for_container_deletion(wallet, cid, self.shell, rpc_endpoint)

    @allure.title("Delete container without force (name={name})")
    @pytest.mark.smoke
    def test_container_deletion_no_force(self, container: str, default_wallet: WalletInfo, rpc_endpoint: str):
        with reporter.step("Delete container and check it was deleted"):
            delete_container(default_wallet, container, self.shell, rpc_endpoint, await_mode=True)
            self.tick_epoch()
            wait_for_container_deletion(default_wallet, container, self.shell, rpc_endpoint)

    @allure.title("Parallel container creation and deletion")
    def test_container_creation_deletion_parallel(self, default_wallet: WalletInfo, rpc_endpoint: str):
        containers_count = 3
        wallet = default_wallet
        iteration_count = 10

        for _ in range(iteration_count):
            cids: list[str] = []
            with reporter.step(f"Create {containers_count} containers"):
                for _ in range(containers_count):
                    cids.append(
                        create_container(wallet, self.shell, rpc_endpoint, self.PLACEMENT_RULE, await_mode=False, wait_for_creation=False)
                    )

            with reporter.step("Wait for containers occur in container list"):
                for cid in cids:
                    wait_for_container_creation(wallet, cid, self.shell, rpc_endpoint, sleep_interval=containers_count)

            with reporter.step("Delete containers and check they were deleted"):
                for cid in cids:
                    # Force to skip frostfs-cli verifictions before delete.
                    # Because no APE rules assigned to container, those verifications will fail due to APE requests denial.
                    delete_container(wallet, cid, self.shell, rpc_endpoint, force=True, await_mode=True)
                    containers_list = list_containers(wallet, self.shell, rpc_endpoint)
                    assert cid not in containers_list, "Container not deleted"