Merge pull request ClickHouse#80682 from azat/tests/rabbitmq-pause-fix

tests: attempt to fix RabbitMQ tests (pause/unpause breaks RabbitMQ)

Author: azat

tests/integration/helpers/cluster.py

@@ -2480,7 +2480,20 @@ def start_rabbitmq_app(self, timeout=120):
         run_rabbitmqctl(
             self.rabbitmq_docker_id, self.rabbitmq_cookie, "start_app", timeout
         )
-        self.wait_rabbitmq_to_start()
+        self.wait_rabbitmq_to_start(timeout)
+
+    @contextmanager
+    def pause_rabbitmq(self, monitor=None, timeout=120):
+        if monitor is not None:
+            monitor.stop()
+        self.stop_rabbitmq_app(timeout)
+
+        try:
+            yield
+        finally:
+            self.start_rabbitmq_app(timeout)
+            if monitor is not None:
+                monitor.start(self)
 
     def reset_rabbitmq(self, timeout=120):
         self.stop_rabbitmq_app()

tests/integration/test_storage_rabbitmq/test.py

@@ -2210,7 +2210,7 @@ def test_rabbitmq_no_connection_at_startup_2(rabbitmq_cluster):
     )
     instance.query("DETACH TABLE test.cs")
 
-    with rabbitmq_cluster.pause_container("rabbitmq1"):
+    with rabbitmq_cluster.pause_rabbitmq():
         instance.query("ATTACH TABLE test.cs")
 
     messages_num = 1000

tests/integration/test_storage_rabbitmq/test_failed_connection.py

@@ -126,17 +126,6 @@ def stop(self):
             self.connection = None
 
 
-def suspend_rabbitmq(rabbitmq_cluster, rabbitmq_monitor):
-    rabbitmq_monitor.stop()
-    rabbitmq_cluster.stop_rabbitmq_app()
-
-
-def resume_rabbitmq(rabbitmq_cluster, rabbitmq_monitor):
-    rabbitmq_cluster.start_rabbitmq_app()
-    rabbitmq_cluster.wait_rabbitmq_to_start()
-    rabbitmq_monitor.start(rabbitmq_cluster)
-
-
 # Fixtures
 
 @pytest.fixture(scope="module")
@@ -234,14 +223,11 @@ def test_rabbitmq_restore_failed_connection_without_losses_1(rabbitmq_cluster, r
     else:
         pytest.fail(f"Time limit of {DEFAULT_TIMEOUT_SEC} seconds reached. The count is still 0.")
 
-    suspend_rabbitmq(rabbitmq_cluster, rabbitmq_monitor)
-
-    number = int(instance.query("SELECT count() FROM test.view"))
-    logging.debug(f"{number}/{messages_num} after suspending RabbitMQ")
-    if number == messages_num:
-        pytest.fail("All RabbitMQ messages have been consumed before resuming the RabbitMQ server")
-
-    resume_rabbitmq(rabbitmq_cluster, rabbitmq_monitor)
+    with rabbitmq_cluster.pause_rabbitmq(rabbitmq_monitor):
+        number = int(instance.query("SELECT count() FROM test.view"))
+        logging.debug(f"{number}/{messages_num} after suspending RabbitMQ")
+        if number == messages_num:
+            pytest.fail("All RabbitMQ messages have been consumed before resuming the RabbitMQ server")
 
     deadline = time.monotonic() + CLICKHOUSE_VIEW_TIMEOUT_SEC
     while time.monotonic() < deadline:
@@ -325,14 +311,11 @@ def test_rabbitmq_restore_failed_connection_without_losses_2(rabbitmq_cluster, r
     else:
         pytest.fail(f"Time limit of {DEFAULT_TIMEOUT_SEC} seconds reached. The count is still 0.")
 
-    suspend_rabbitmq(rabbitmq_cluster, rabbitmq_monitor)
-
-    number = int(instance.query("SELECT count() FROM test.view"))
-    logging.debug(f"{number}/{messages_num} after suspending RabbitMQ")
-    if number == messages_num:
-        pytest.fail("All RabbitMQ messages have been consumed before resuming the RabbitMQ server")
-
-    resume_rabbitmq(rabbitmq_cluster, rabbitmq_monitor)
+    with rabbitmq_cluster.pause_rabbitmq(rabbitmq_monitor):
+        number = int(instance.query("SELECT count() FROM test.view"))
+        logging.debug(f"{number}/{messages_num} after suspending RabbitMQ")
+        if number == messages_num:
+            pytest.fail("All RabbitMQ messages have been consumed before resuming the RabbitMQ server")
 
     # while int(instance.query('SELECT count() FROM test.view')) == 0:
     #    time.sleep(0.1)