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

from frostfs_testlib.steps.cli.container import (
    create_container,
    delete_container,
    get_container,
)
from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
from pytest_tests.helpers.utility import placement_policy_from_container
from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
from frostfs_testlib.utils.file_utils import generate_file
import allure
import pytest
from frostfs_testlib.resources.wellknown_acl import PUBLIC_ACL
from frostfs_testlib.steps.cli.container import create_container, get_container
from frostfs_testlib.steps.cli.object import (
    put_object_to_random_node,
)
from frostfs_testlib.steps.node_management import (
    check_node_in_map,
)
from frostfs_testlib.steps.storage_policy import get_nodes_with_object, get_simple_object_copies
from pytest_tests.helpers.utility import (
    placement_policy_from_container,
)

@pytest.mark.container
@pytest.mark.policy
class TestPolicy(ClusterTestBase):
    
    @pytest.mark.skip(reason="ошибка с фикстурой")
    @allure.title("[NEGATIVE] Placement policy")
    @pytest.mark.policy
    def test_placement_policy_negative(
        self, default_wallet, placement_rule
    ):
        """
        Negative test for placement policy.
        """
        wallet = default_wallet
        endpoint = self.cluster.default_rpc_endpoint
        try:
            cid = create_container(
                wallet, rule=placement_rule, basic_acl=PUBLIC_ACL, shell=self.shell, endpoint=endpoint
            )
        except:
            got_policy = placement_policy_from_container(
                get_container(wallet, cid, json_mode=False, shell=self.shell, endpoint=endpoint)
            )
            assert got_policy == placement_rule.replace(
                "'", ""
            ), f"Can't parse placement policy"

    @pytest.mark.skip(reason="ошибка с фикстурой")
    @allure.title("110569 [NEGATIVE] Placement policy: Not enough nodes to SELECT")
    @pytest.mark.policy
    def test_placement_policy_negative_not_enough_nodes_to_select(
        self, default_wallet, placement_rule
    ):
        """
        Negative test for placement policy: Not enough nodes to SELECT.
        """
        wallet = default_wallet
        endpoint = self.cluster.default_rpc_endpoint
        with pytest.raises(RuntimeError, match=".*not enough nodes to SELECT from.*"):
            cid = create_container(
                wallet, rule=placement_rule, basic_acl=PUBLIC_ACL, shell=self.shell, endpoint=endpoint
            )
    
    @pytest.mark.skip(reason="ошибка с фикстурой")       
    @allure.title("110570 [NEGATIVE] Placement policy: Filter not found")
    @pytest.mark.policy
    def test_placement_policy_negative_not_enough_nodes_to_filter(
        self, default_wallet, placement_rule
    ):
        """
        Negative test for placement policy: Filter not found.
        """
        wallet = default_wallet
        endpoint = self.cluster.default_rpc_endpoint
        with pytest.raises(RuntimeError, match=".*not enough nodes to FILTER from.*"):
            cid = create_container(
                wallet, rule=placement_rule, basic_acl=PUBLIC_ACL, shell=self.shell, endpoint=endpoint
            )
            
    @pytest.mark.skip(reason="ошибка с фикстурой")   
    @allure.title("110572 [NEGATIVE] Placement policy: SELECTOR not found")
    @pytest.mark.policy
    def test_placement_policy_negative_not_enough_nodes_to_selector(
        self, default_wallet, placement_rule
    ):
        """
        Negative test for placement policy: Filter not found.
        """
        wallet = default_wallet
        endpoint = self.cluster.default_rpc_endpoint
        with pytest.raises(RuntimeError, match=".*not enough nodes to SELECTOR from.*"):
            cid = create_container(
                wallet, rule=placement_rule, basic_acl=PUBLIC_ACL, shell=self.shell, endpoint=endpoint
            )

    @pytest.mark.parametrize(
        "placement_rule,expected_copies,expected_nodes_id",
        [
            ("REP 1 REP 1 CBF 1", 2, {2, 2}),
        ]
    )
    @pytest.mark.policy
    @allure.title("110571 Object should have {expected_copies} copies with policy {placement_rule}")
    def test_simple_policy_results_with_one_node(
        self,
        default_wallet,
        placement_rule,
        expected_copies,
        expected_nodes_id: set[int],
        simple_object_size: ObjectSize,
    ):
        """
        This test checks object's copies based on container's placement simple policy results with one node.
        """
        wallet = default_wallet    
        file_path = generate_file(simple_object_size.value)

        cid, oid = self.validate_object_copies(
            wallet, placement_rule, file_path
        )

        self.check_expected_copies(cid, oid, expected_copies, expected_nodes_id)

    @pytest.mark.parametrize(
        "placement_rule,expected_copies,expected_nodes_id",
        [
            ("UNIQUE REP 1 IN AnyNode REP 1 IN AnyNode CBF 1 SELECT 1 FROM * AS AnyNode", 2, {2, 3}),
        ]
    )
    @pytest.mark.policy
    @allure.title("110544 Object should have {expected_copies} copies with policy {placement_rule}")
    def test_policy_with_select_results_with_unique_nodes(
        self,
        default_wallet,
        placement_rule,
        expected_copies,
        expected_nodes_id: set[int],
        simple_object_size: ObjectSize,
    ):
        """
        This test checks object's copies based on container's placement policy with SELECT results with UNIQUE nodes.
        """
        wallet = default_wallet    
        file_path = generate_file(simple_object_size.value)

        cid, oid = self.validate_object_copies(
            wallet, placement_rule, file_path
        )

        self.check_expected_copies(cid, oid, expected_copies, expected_nodes_id)

    @pytest.mark.parametrize(
        "placement_rule,expected_copies,expected_nodes_id",
        [
            ('UNIQUE REP 1 IN RUS REP 1 IN RUS CBF 1 SELECT 1 FROM RU AS RUS FILTER Country NE Sweden AS NotSE FILTER @NotSE AND NOT (CountryCode EQ FI) AND Country EQ "Russia" AS RU', 2, {3, 1}),
        ]
    )
    @pytest.mark.policy
    @allure.title("110545 Object should have {expected_copies} copies with policy {placement_rule}")
    def test_policy_with_select_and_complex_filter_results_with_unique_nodes(
        self,
        default_wallet,
        placement_rule,
        expected_copies,
        expected_nodes_id: set[int],
        simple_object_size: ObjectSize,
    ):
        """
        This test checks object's copies based on container's placement policy with SELECT and Complex FILTER results with UNIQUE nodes.
        """
        wallet = default_wallet    
        file_path = generate_file(simple_object_size.value)

        cid, oid = self.validate_object_copies(
            wallet, placement_rule, file_path
        )

        self.check_expected_copies(cid, oid, expected_copies, expected_nodes_id)

    @pytest.mark.parametrize(
        "placement_rule,expected_copies,expected_nodes_id",
        [
            ("""REP 4""",
             4, {3, 2, 1, 4}),
        ]
    )
    @pytest.mark.policy
    @allure.title("110610 Object should have {expected_copies} copies with policy {placement_rule}")
    def test_simple_policy_results_with_100_of_available_nodes(
        self,
        default_wallet,
        placement_rule,
        expected_copies,
        expected_nodes_id: set[int],
        simple_object_size: ObjectSize,
    ):
        """
        This test checks object's copies based on container's placement simple policy results with 100% of available nodes.
        """
        wallet = default_wallet    
        file_path = generate_file(simple_object_size.value)

        cid, oid = self.validate_object_copies(
            wallet, placement_rule, file_path
        )

        self.check_expected_copies(cid, oid, expected_copies, expected_nodes_id)

    @pytest.mark.parametrize(
        "placement_rule,expected_copies,expected_nodes_id",
        [
            ("UNIQUE REP 1 REP 1 CBF 1", 2, {2, 3}),
        ]
    )
    @pytest.mark.policy
    @allure.title("110537 Object should have {expected_copies} copies with policy {placement_rule}")
    def test_policy_with_select_and_complex_filter_results_with_unique_nodes(
        self,
        default_wallet,
        placement_rule,
        expected_copies,
        expected_nodes_id: set[int],
        simple_object_size: ObjectSize,
    ):
        """
        This test checks object's copies based on container's placement simple policy results with UNIQUE nodes.
        """
        wallet = default_wallet    
        file_path = generate_file(simple_object_size.value)

        cid, oid = self.validate_object_copies(
            wallet, placement_rule, file_path
        )

        self.check_expected_copies(cid, oid, expected_copies, expected_nodes_id)

    @pytest.mark.parametrize(
        "placement_rule,expected_copies,expected_nodes_id",
        [
            ("UNIQUE REP 1 REP 1 CBF 1", 2, {2, 3}),
        ]
    )
    @pytest.mark.policy
    @allure.title("110587 Object should have {expected_copies} copies with policy {placement_rule}")
    def test_policy_with_multi_selects_and_filters_results_with_one_node(
        self,
        default_wallet,
        placement_rule,
        expected_copies,
        expected_nodes_id: set[int],
        simple_object_size: ObjectSize,
    ):
        """
        This test checks object's copies based on container's placement policy with Multi SELECTs and FILTERs results with one nodes.
        """
        wallet = default_wallet    
        file_path = generate_file(simple_object_size.value)

        cid, oid = self.validate_object_copies(
            wallet, placement_rule, file_path
        )

        self.check_expected_copies(cid, oid, expected_copies, expected_nodes_id)

    @pytest.mark.parametrize(
        "placement_rule,expected_copies,expected_nodes_id",
        [
            ("REP 1 CBF 1", 1, {2}),
        ]
    )
    @pytest.mark.policy
    @allure.title("110593 Object should have {expected_copies} copies with policy {placement_rule}")
    def test_simple_policy_results_with_25_of_available_nodes(
        self,
        default_wallet,
        placement_rule,
        expected_copies,
        expected_nodes_id: set[int],
        simple_object_size: ObjectSize,
    ):
        """
        This test checks object's copies based on container's placement policy results with 25% of available nodes.
        """
        wallet = default_wallet    
        file_path = generate_file(simple_object_size.value)

        cid, oid = self.validate_object_copies(
            wallet, placement_rule, file_path
        )

        self.check_expected_copies(cid, oid, expected_copies, expected_nodes_id)

    @pytest.mark.parametrize(
        "placement_rule,expected_copies,expected_nodes_id",
        [
            ("REP 1 IN One CBF 1 SELECT 1 FROM * AS One", 1, {2}),
        ]
    )
    @pytest.mark.policy
    @allure.title("110594 Object should have {expected_copies} copies with policy {placement_rule}")
    def test_policy_with_select_results_with_25_of_available_nodes(
        self,
        default_wallet,
        placement_rule,
        expected_copies,
        expected_nodes_id: set[int],
        simple_object_size: ObjectSize,
    ):
        """
        This test checks object's copies based on container's placement policy with SELECT results with 25% of available nodes. 
        """
        wallet = default_wallet    
        file_path = generate_file(simple_object_size.value)

        cid, oid = self.validate_object_copies(
            wallet, placement_rule, file_path
        )

        self.check_expected_copies(cid, oid, expected_copies, expected_nodes_id)
    
    @pytest.mark.parametrize(
        "placement_rule,expected_copies,expected_nodes_id",
        [
            ("REP 1 IN Nodes25 SELECT 1 FROM LE10 AS Nodes25 FILTER Price LE 10 AS LE10", 1, {2}),
        ]
    )
    @pytest.mark.policy
    @allure.title("110595 Object should have {expected_copies} copies with policy {placement_rule}")
    def test_policy_with_select_and_filter_results_with_25_of_available_nodes(
        self,
        default_wallet,
        placement_rule,
        expected_copies,
        expected_nodes_id: set[int],
        simple_object_size: ObjectSize,
    ):
        """
        This test checks object's copies based on container's placement policy with SELECT and FILTER results with 25% of available nodes.
        """
        wallet = default_wallet    
        file_path = generate_file(simple_object_size.value)

        cid, oid = self.validate_object_copies(
            wallet, placement_rule, file_path
        )

        self.check_expected_copies(cid, oid, expected_copies, expected_nodes_id)
    
    @pytest.mark.parametrize(
        "placement_rule,expected_copies,expected_nodes_id",
        [
            ("""REP 1 IN Nodes25 SELECT 1 FROM BET0AND10 AS Nodes25 FILTER Price LE 10 AS LE10 FILTER Price GT 0 AS GT0 FILTER @LE10 AND @GT0 AS BET0AND10""",
             1, {1}),
        ]
    )
    @pytest.mark.policy
    @allure.title("110596 Object should have {expected_copies} copies with policy {placement_rule}")
    def test_policy_with_select_and_complex_filter_results_with_25_of_available_nodes(
        self,
        default_wallet,
        placement_rule,
        expected_copies,
        expected_nodes_id: set[int],
        complex_object_size: ObjectSize,
    ):
        """
        110596 This test checks object's copies based on container's placement policy with SELECT and Complex FILTER results with 25% of available nodes.
        """
        wallet = default_wallet    
        file_path = generate_file(complex_object_size.value)

       
        cid, oid = self.validate_object_copies(
            wallet, placement_rule, file_path
        )

        self.check_expected_copies(cid, oid, expected_copies, expected_nodes_id)
    
    @pytest.mark.parametrize(
        "placement_rule,expected_copies,expected_nodes_id",
        [
            ("""UNIQUE REP 1 IN MyRussianNodes REP 1 IN MyRussianNodes CBF 1 SELECT 1 FROM RussianNodes AS MyRussianNodes FILTER Country EQ Russia AS RussianNodes""",
             2, {3, 1}),
        ]
    )
    @pytest.mark.policy
    @allure.title("110588 Object should have {expected_copies} copies with policy {placement_rule}")
    def test_policy_with_select_and_filter_results_with_unique_nodes(
        self,
        default_wallet,
        placement_rule,
        expected_copies,
        expected_nodes_id: set[int],
        simple_object_size: ObjectSize,
    ):
        """
        This test checks object's copies based on container's placement policy with SELECT and FILTER results with UNIQUE nodes.
        """
        wallet = default_wallet    
        file_path = generate_file(simple_object_size.value)

        cid, oid = self.validate_object_copies(
            wallet, placement_rule, file_path
        )

        self.check_expected_copies(cid, oid, expected_copies, expected_nodes_id)

    @pytest.mark.parametrize(
        "placement_rule,expected_copies,expected_nodes_id",
        [
            ("""UNIQUE REP 1 IN MyRussianNodes REP 1 IN MyRussianNodes CBF 1 SELECT 1 FROM RussianNodes AS MyRussianNodes FILTER Country EQ Russia AS RussianNodes""",
             2, {3, 1}),
        ]
    )
    @pytest.mark.policy
    @allure.title("110586 Object should have {expected_copies} copies with policy {placement_rule}")
    def test_policy_with_select_and_filter_results_with_unique_nodes(
        self,
        default_wallet,
        placement_rule,
        expected_copies,
        expected_nodes_id: set[int],
        simple_object_size: ObjectSize,
    ):
        """
        This test checks object's copies based on container's placement policy with SELECT and FILTER results with UNIQUE nodes.
        """
        wallet = default_wallet    
        file_path = generate_file(simple_object_size.value)

        cid, oid = self.validate_object_copies(
            wallet, placement_rule, file_path
        )

        self.check_expected_copies(cid, oid, expected_copies, expected_nodes_id)

    @allure.step("Validate policy")
    def validate_object_policy(
        self, wallet: str, placement_rule: str, cid: str, endpoint: str
    ):
        got_policy = placement_policy_from_container(
            get_container(wallet, cid, json_mode=False, shell=self.shell, endpoint=endpoint)
        )
        assert got_policy == placement_rule.replace(
            "'", ""
        ), f"Expected \n{placement_rule} and got policy \n{got_policy} are the same"

    @allure.step("Validate expected copies")
    def check_expected_copies(self, cid: str, oid: str, expected_copies: int, expected_copies_id: set):
            nodes = get_nodes_with_object(cid, oid, shell=self.shell, nodes=self.cluster.storage_nodes)
            assert len(nodes) == expected_copies, f"Expected {expected_copies} copies, got {len(nodes)}"

            nodes_id = {node.id for node in nodes}
            assert nodes_id == expected_copies_id, f"Expected {expected_copies_id} copies, got {nodes_id}"

    @allure.step("Validate object copies")
    def validate_object_copies(
        self, wallet: str, placement_rule: str, file_path: str
    ) -> set[int]:
        endpoint = self.cluster.default_rpc_endpoint

        with allure.step(f"Create container"):
            cid = create_container(
                wallet, rule=placement_rule, basic_acl=PUBLIC_ACL, shell=self.shell, endpoint=endpoint
            )

        self.validate_object_policy(wallet, placement_rule, cid, endpoint)

        with allure.step(f"Put object"):
            oid = put_object_to_random_node(
                wallet, file_path, cid, shell=self.shell, cluster=self.cluster
            )
        return cid, oid