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

import json

import allure
import pytest
from epoch import tick_epoch
from python_keywords.container import (
    create_container,
    delete_container,
    get_container,
    list_containers,
    wait_for_container_creation,
    wait_for_container_deletion,
)
from utility import placement_policy_from_container
from wellknown_acl import PRIVATE_ACL_F

from steps.cluster_test_base import ClusterTestBase


@pytest.mark.container
@pytest.mark.sanity
@pytest.mark.container
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, name):
        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
            )
            tick_epoch(self.shell, self.cluster)
            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):
        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(f"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
                )
            tick_epoch(self.shell, self.cluster)
            wait_for_container_deletion(
                wallet, cid, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint
            )