src/frostfs_testlib/storage/controllers/cluster_state_controller.py

@@ -23,6 +23,7 @@ class ClusterStateController:
         self.stopped_nodes: list[ClusterNode] = []
         self.detached_disks: dict[str, DiskController] = {}
         self.stopped_storage_nodes: list[ClusterNode] = []
+        self.stopped_s3_gate: list[ClusterNode] = []
         self.cluster = cluster
         self.shell = shell
         self.suspended_services: dict[str, list[ClusterNode]] = {}
@@ -131,6 +132,28 @@ class ClusterStateController:
         wait_all_storage_nodes_returned(self.shell, self.cluster)
         self.stopped_storage_nodes = []
 
+    
+    @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
+    @reporter.step_deco("Stop s3 gate on {node}")
+    def stop_s3_gate(self, node: ClusterNode):
+        node.s3_gate.stop_service()
+        self.stopped_s3_gate.append(node)
+
+    @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
+    @reporter.step_deco("Start s3 gate on {node}")
+    def start_s3_gate(self, node: ClusterNode):
+        node.s3_gate.start_service()
+        self.stopped_s3_gate.remove(node)
+
+
+    @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
+    @reporter.step_deco("Start stopped S3 gates")
+    def start_stopped_s3_gate(self):
+        # not sure if we need here to use threads like in start_stopped_storage_services
+        for s3_gate in self.stopped_s3_gate:
+            s3_gate.start_service()
+        self.stopped_s3_gate = []
+
     @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
     @reporter.step_deco("Suspend {process_name} service in {node}")
     def suspend_service(self, process_name: str, node: ClusterNode):