Add test_panic_storage_node

pytest_tests/helpers/sbercloud_helper.py

@@ -92,10 +92,10 @@ class SberCloud:
                                  headers={'Content-Type': 'application/json', 'X-Auth-Token': self.token})
         assert response.status_code < 300, f'Status:{response.status_code}. Server not started: {response.json()}'
 
-    def stop_node(self, node_id: str = None, node_ip: str = None):
+    def stop_node(self, node_id: str = None, node_ip: str = None, stop_type: str = 'HARD'):
         data = {
             'os-stop': {
-                'type': 'HARD',
+                'type': stop_type,
                 'servers': [
                     {
                         'id': node_id or self.find_esc_by_ip(node_ip)

pytest_tests/testsuites/failovers/test_failover_storage.py

@@ -4,10 +4,12 @@ from time import sleep
 
 import allure
 import pytest
+from common import STORAGE_NODE_PWD, STORAGE_NODE_USER
 from python_keywords.container import create_container
 from python_keywords.neofs_verbs import get_object, put_object
 from python_keywords.utility_keywords import get_file_hash
 from sbercloud_helper import SberCloud
+from ssh_helper import HostClient, HostIsNotAvailable
 from storage_policy import get_nodes_with_object
 from wellknown_acl import PUBLIC_ACL
 
@@ -26,6 +28,15 @@ def return_all_storage_nodes_fixture(sbercloud_client):
     return_all_storage_nodes(sbercloud_client)
 
 
+@allure.title('Hard reboot host via magic SysRq option')
+def panic_reboot_host(ip: str = None):
+    ssh = HostClient(ip=ip, login=STORAGE_NODE_USER, password=STORAGE_NODE_PWD)
+    ssh.create_connection(attempts=1)
+    ssh.exec('echo 1 > /proc/sys/kernel/sysrq')
+    with pytest.raises(HostIsNotAvailable):
+        ssh.exec('echo b > /proc/sysrq-trigger', timeout=1)
+
+
 def return_all_storage_nodes(sbercloud_client: SberCloud):
     for host in stopped_hosts:
         sbercloud_client.start_node(node_ip=host.split(':')[-2])
@@ -68,3 +79,25 @@ def test_lost_storage_node(prepare_wallet_and_deposit, generate_file, sbercloud_
 
     got_file_path = get_object(wallet, cid, oid)
     assert get_file_hash(generate_file) == get_file_hash(got_file_path)
+
+
+@allure.title('Panic storage node(s)')
+@pytest.mark.parametrize('sequence', [True, False])
+def test_panic_storage_node(prepare_wallet_and_deposit, generate_file, sequence: bool):
+    wallet, _ = prepare_wallet_and_deposit
+    placement_rule = 'REP 2 IN X CBF 2 SELECT 2 FROM * AS X'
+    cid = create_container(wallet, rule=placement_rule, basic_acl=PUBLIC_ACL)
+    oid = put_object(wallet, generate_file, cid)
+
+    with allure.step(f'Return storage nodes'):
+        nodes = wait_object_replication(wallet, cid, oid, 2)
+        for node in nodes:
+            panic_reboot_host(ip=node.split(':')[-2])
+            if sequence:
+                wait_object_replication(wallet, cid, oid, 2)
+
+        if not sequence:
+            wait_object_replication(wallet, cid, oid, 2)
+
+        got_file_path = get_object(wallet, cid, oid)
+        assert get_file_hash(generate_file) == get_file_hash(got_file_path)