diff --git a/pytest_tests/testsuites/object/conftest.py b/pytest_tests/testsuites/object/conftest.py new file mode 100644 index 00000000..9898f505 --- /dev/null +++ b/pytest_tests/testsuites/object/conftest.py @@ -0,0 +1,9 @@ +import pytest +from frostfs_testlib.storage.dataclasses.policy import PlacementPolicy + + +@pytest.fixture() +def expected_object_copies(placement_policy: PlacementPolicy) -> int: + if placement_policy.name == "rep": + return 2 + return 4 diff --git a/pytest_tests/testsuites/object/test_object_api.py b/pytest_tests/testsuites/object/test_object_api.py index eb36df1e..d3f7baa7 100755 --- a/pytest_tests/testsuites/object/test_object_api.py +++ b/pytest_tests/testsuites/object/test_object_api.py @@ -94,11 +94,17 @@ def generate_ranges( scope="module" ) def storage_objects( - default_wallet: WalletInfo, client_shell: Shell, cluster: Cluster, object_size: ObjectSize, placement_policy: PlacementPolicy + default_wallet: WalletInfo, + client_shell: Shell, + cluster: Cluster, + object_size: ObjectSize, + placement_policy: PlacementPolicy, ) -> list[StorageObjectInfo]: wallet = default_wallet # Separate containers for complex/simple objects to avoid side-effects - cid = create_container(wallet, shell=client_shell, rule=placement_policy.value, endpoint=cluster.default_rpc_endpoint) + cid = create_container( + wallet, shell=client_shell, rule=placement_policy.value, endpoint=cluster.default_rpc_endpoint + ) file_path = generate_file(object_size.value) file_hash = get_file_hash(file_path) @@ -140,6 +146,7 @@ class TestObjectApi(ClusterTestBase): self, storage_objects: list[StorageObjectInfo], simple_object_size: ObjectSize, + expected_object_copies: int, ): """ Validate object storage policy @@ -163,7 +170,7 @@ class TestObjectApi(ClusterTestBase): shell=self.shell, nodes=self.cluster.storage_nodes, ) - assert copies == 2, "Expected 2 copies" + assert copies == expected_object_copies, f"Expected {expected_object_copies} copies" @allure.title("Get object by native API (obj_size={object_size}, policy={placement_policy})") def test_get_object_api(self, storage_objects: list[StorageObjectInfo]):