pytest_tests/testsuites/network/test_node_management.py

@@ -30,9 +30,11 @@ from pytest_tests.helpers.node_management import (
     include_node_to_network_map,
     node_shard_list,
     node_shard_set_mode,
+    start_storage_nodes,
+    stop_random_storage_nodes,
     storage_node_healthcheck,
     storage_node_set_status,
-    wait_for_node_to_be_ready
+    wait_for_node_to_be_ready,
 )
 from pytest_tests.helpers.storage_policy import get_nodes_with_object, get_simple_object_copies
 from pytest_tests.helpers.utility import (
@@ -412,6 +414,38 @@ class TestNodeManagement(ClusterTestBase):
             oid = put_object_to_random_node(wallet, file_path, cid, self.shell, self.cluster)
             delete_object(wallet, cid, oid, self.shell, self.cluster.default_rpc_endpoint)
 
+    @pytest.mark.node_mgmt
+    @allure.title("Put object with stopped node")
+    def test_stop_node(self, default_wallet, return_nodes_after_test_run, simple_object_size):
+        wallet = default_wallet
+        placement_rule = "REP 3 SELECT 4 FROM * AS X"
+        source_file_path = generate_file(simple_object_size)
+        storage_nodes = self.cluster.storage_nodes
+        random_node = random.choice(storage_nodes[1:])
+        alive_node = random.choice(
+            [storage_node for storage_node in storage_nodes if storage_node.id != random_node.id]
+        )
+
+        cid = create_container(
+            wallet,
+            rule=placement_rule,
+            basic_acl=PUBLIC_ACL,
+            shell=self.shell,
+            endpoint=random_node.get_rpc_endpoint(),
+        )
+        with allure.step(f"Stop the random node"):
+            check_nodes.append(random_node)
+            random_node.stop_service()
+        with allure.step("Try to put an object and expect success"):
+            put_object(
+                wallet,
+                source_file_path,
+                cid,
+                shell=self.shell,
+                endpoint=alive_node.get_rpc_endpoint(),
+            )
+        self.return_nodes(alive_node)
+
     @allure.step("Validate object has {expected_copies} copies")
     def validate_object_copies(
         self, wallet: str, placement_rule: str, file_path: str, expected_copies: int