src/frostfs_testlib/steps/iptables.py

@@ -1,42 +0,0 @@
-from frostfs_testlib.shell import Shell
-from frostfs_testlib.storage.cluster import ClusterNode
-
-
-class IpTablesHelper:
-    @staticmethod
-    def drop_input_traffic_to_port(node: ClusterNode, ports: list[str]) -> None:
-        shell = node.host.get_shell()
-        for port in ports:
-            shell.exec(f"iptables -A INPUT -p tcp --dport {port} -j DROP")
-
-    @staticmethod
-    def drop_input_traffic_to_node(node: ClusterNode, block_ip: list[str]) -> None:
-        shell = node.host.get_shell()
-        for ip in block_ip:
-            shell.exec(f"iptables -A INPUT -s {ip} -j DROP")
-
-    @staticmethod
-    def restore_input_traffic_to_port(node: ClusterNode) -> None:
-        shell = node.host.get_shell()
-        ports = (
-            shell.exec("iptables -L --numeric | grep DROP | awk '{print $7}'")
-            .stdout.strip()
-            .split("\n")
-        )
-        if ports[0] == "":
-            return
-        for port in ports:
-            shell.exec(f"iptables -D INPUT -p tcp --dport {port.split(':')[-1]} -j DROP")
-
-    @staticmethod
-    def restore_input_traffic_to_node(node: ClusterNode) -> None:
-        shell = node.host.get_shell()
-        unlock_ip = (
-            shell.exec("iptables -L --numeric | grep DROP | awk '{print $4}'")
-            .stdout.strip()
-            .split("\n")
-        )
-        if unlock_ip[0] == "":
-            return
-        for ip in unlock_ip:
-            shell.exec(f"iptables -D INPUT -s {ip} -j DROP")

src/frostfs_testlib/steps/network.py

@@ -0,0 +1,89 @@
+from frostfs_testlib.reporter import get_reporter
+from frostfs_testlib.storage.cluster import ClusterNode
+from frostfs_testlib.testing.test_control import retry
+
+reporter = get_reporter()
+
+
+class IpTablesHelper:
+    @staticmethod
+    def drop_input_traffic_to_port(node: ClusterNode, ports: list[str]) -> None:
+        shell = node.host.get_shell()
+        for port in ports:
+            shell.exec(f"iptables -A INPUT -p tcp --dport {port} -j DROP")
+
+    @staticmethod
+    def drop_input_traffic_to_node(node: ClusterNode, block_ip: list[str]) -> None:
+        shell = node.host.get_shell()
+        for ip in block_ip:
+            shell.exec(f"iptables -A INPUT -s {ip} -j DROP")
+
+    @staticmethod
+    def restore_input_traffic_to_port(node: ClusterNode) -> None:
+        shell = node.host.get_shell()
+        ports = (
+            shell.exec("iptables -L --numeric | grep DROP | awk '{print $7}'")
+            .stdout.strip()
+            .split("\n")
+        )
+        if ports[0] == "":
+            return
+        for port in ports:
+            shell.exec(f"iptables -D INPUT -p tcp --dport {port.split(':')[-1]} -j DROP")
+
+    @staticmethod
+    def restore_input_traffic_to_node(node: ClusterNode) -> None:
+        shell = node.host.get_shell()
+        unlock_ip = (
+            shell.exec("iptables -L --numeric | grep DROP | awk '{print $4}'")
+            .stdout.strip()
+            .split("\n")
+        )
+        if unlock_ip[0] == "":
+            return
+        for ip in unlock_ip:
+            shell.exec(f"iptables -D INPUT -s {ip} -j DROP")
+
+
+# TODO Move class to HOST
+class IfUpDownHelper:
+    @reporter.step_deco("Down {interface} to {node}")
+    def down_interface(self, node: ClusterNode, interface: str) -> None:
+        shell = node.host.get_shell()
+        shell.exec(f"ifdown {interface}")
+
+    @reporter.step_deco("Up {interface} to {node}")
+    def up_interface(self, node: ClusterNode, interface: str) -> None:
+        shell = node.host.get_shell()
+        shell.exec(f"ifup {interface}")
+
+    @reporter.step_deco("Up all interface to {node}")
+    def up_all_interface(self, node: ClusterNode) -> None:
+        shell = node.host.get_shell()
+        interfaces = list(node.host.config.interfaces.keys())
+        shell.exec("ifup -av")
+        for name_interface in interfaces:
+            self.check_state_up(node, name_interface)
+
+    @reporter.step_deco("Down all interface to {node}")
+    def down_all_interface(self, node: ClusterNode) -> None:
+        shell = node.host.get_shell()
+        interfaces = list(node.host.config.interfaces.keys())
+        shell.exec("ifdown -av")
+        for name_interface in interfaces:
+            self.check_state_down(node, name_interface)
+
+    @reporter.step_deco("Check {node} to {interface}")
+    def check_state(self, node: ClusterNode, interface: str) -> str:
+        shell = node.host.get_shell()
+        return shell.exec(
+            f"ip link show {interface} | sed -z 's/.*state \(.*\) mode .*/\\1/'"
+        ).stdout.strip()
+
+    @retry(max_attempts=5, sleep_interval=5, expected_result="UP")
+    def check_state_up(self, node: ClusterNode, interface: str) -> str:
+        return self.check_state(node=node, interface=interface)
+
+    @retry(max_attempts=5, sleep_interval=5, expected_result="DOWN")
+    def check_state_down(self, node: ClusterNode, interface: str) -> str:
+        return self.check_state(node=node, interface=interface)

src/frostfs_testlib/storage/controllers/cluster_state_controller.py

@@ -4,7 +4,7 @@ import time
 import frostfs_testlib.resources.optionals as optionals
 from frostfs_testlib.reporter import get_reporter
 from frostfs_testlib.shell import CommandOptions, Shell
-from frostfs_testlib.steps.iptables import IpTablesHelper
+from frostfs_testlib.steps.network import IfUpDownHelper, IpTablesHelper
 from frostfs_testlib.storage.cluster import Cluster, ClusterNode, StorageNode
 from frostfs_testlib.storage.controllers.disk_controller import DiskController
 from frostfs_testlib.storage.dataclasses.node_base import NodeBase, ServiceClass
@@ -18,6 +18,7 @@ from frostfs_testlib.utils.failover_utils import (
 )
 
 reporter = get_reporter()
+if_up_down_helper = IfUpDownHelper()
 
 
 class ClusterStateController:
@@ -31,6 +32,7 @@ class ClusterStateController:
         self.cluster = cluster
         self.shell = shell
         self.suspended_services: dict[str, list[ClusterNode]] = {}
+        self.nodes_with_modified_interface: list[ClusterNode] = []
 
     @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
     @reporter.step_deco("Stop host of node {node}")
@@ -312,6 +314,26 @@ class ClusterStateController:
             wait_for_host_online(self.shell, node.storage_node)
             wait_for_node_online(node.storage_node)
 
+    @reporter.step_deco("Down {interface} to {nodes}")
+    def down_interface(self, nodes: list[ClusterNode], interface: str):
+        for node in nodes:
+            if_up_down_helper.down_interface(node=node, interface=interface)
+            assert if_up_down_helper.check_state(node=node, interface=interface) == "DOWN"
+            self.nodes_with_modified_interface.append(node)
+
+    @reporter.step_deco("Up {interface} to {nodes}")
+    def up_interface(self, nodes: list[ClusterNode], interface: str):
+        for node in nodes:
+            if_up_down_helper.up_interface(node=node, interface=interface)
+            assert if_up_down_helper.check_state(node=node, interface=interface) == "UP"
+            if node in self.nodes_with_modified_interface:
+                self.nodes_with_modified_interface.remove(node)
+
+    @reporter.step_deco("Restore interface")
+    def restore_interfaces(self):
+        for node in self.nodes_with_modified_interface:
+            if_up_down_helper.up_all_interface(node)
+
     def _get_disk_controller(
         self, node: StorageNode, device: str, mountpoint: str
     ) -> DiskController:

src/frostfs_testlib/testing/cluster_test_base.py

@@ -1,10 +1,13 @@
+import time
 from typing import Optional
 
 from frostfs_testlib.reporter import get_reporter
+from frostfs_testlib.resources.common import MORPH_BLOCK_TIME
 from frostfs_testlib.shell import Shell
 from frostfs_testlib.steps import epoch
 from frostfs_testlib.storage.cluster import Cluster
 from frostfs_testlib.storage.dataclasses.frostfs_services import StorageNode
+from frostfs_testlib.utils import datetime_utils
 
 reporter = get_reporter()
 
@@ -14,13 +17,24 @@ class ClusterTestBase:
     shell: Shell
     cluster: Cluster
 
-    @reporter.step_deco("Tick {epochs_to_tick} epochs")
-    def tick_epochs(self, epochs_to_tick: int, alive_node: Optional[StorageNode] = None):
+    @reporter.step_deco("Tick {epochs_to_tick} epochs, wait {wait_block} block")
+    def tick_epochs(
+        self,
+        epochs_to_tick: int,
+        alive_node: Optional[StorageNode] = None,
+        wait_block: int = None,
+    ):
         for _ in range(epochs_to_tick):
-            self.tick_epoch(alive_node)
+            self.tick_epoch(alive_node, wait_block)
 
-    def tick_epoch(self, alive_node: Optional[StorageNode] = None):
+    def tick_epoch(
+        self,
+        alive_node: Optional[StorageNode] = None,
+        wait_block: int = None,
+    ):
         epoch.tick_epoch(self.shell, self.cluster, alive_node=alive_node)
+        if wait_block:
+            time.sleep(datetime_utils.parse_time(MORPH_BLOCK_TIME) * wait_block)
 
     def wait_for_epochs_align(self):
         epoch.wait_for_epochs_align(self.shell, self.cluster)