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

I was looking at [1] failure on CI, all problems pops up there after
unpause of rabbitmq container failed:

    # dockerd.log
    time="2025-05-21T11:07:28.863531720Z" level=debug msg="Calling POST /v1.46/containers/1f6103a506ad393e4ff79565ff27e01e73d760722a0412b66f3e0ac3759b2905/unpause"
    time="2025-05-21T11:07:54.715714117Z" level=warning msg="Health check for container 1f6103a506ad393e4ff79565ff27e01e73d760722a0412b66f3e0ac3759b2905 error: timed out starting health check for container 1f6103a506ad393e4ff79565ff27e01e73d760722a0412b66f3e0ac3759b2905"

    # pytest logs
    cat test_storage_rabbitmq_test_py_0.jsonl | jq -r '[.nodeid, .outcome, .duration] | @TSV'
    test_storage_rabbitmq/test.py::test_rabbitmq_no_connection_at_startup_1 passed  0.06497019900052692
    test_storage_rabbitmq/test.py::test_rabbitmq_no_connection_at_startup_1 passed  4.069896273000268
    test_storage_rabbitmq/test.py::test_rabbitmq_no_connection_at_startup_1 passed  2.9214494500010915
    test_storage_rabbitmq/test.py::test_rabbitmq_no_connection_at_startup_2 passed  0.06504274200051441
    test_storage_rabbitmq/test.py::test_rabbitmq_no_connection_at_startup_2 failed  305.5343935210003
    test_storage_rabbitmq/test.py::test_rabbitmq_no_connection_at_startup_2 failed  120.9480928060002
    test_storage_rabbitmq/test.py::test_rabbitmq_format_factory_settings    passed  0.06524586100022134
    test_storage_rabbitmq/test.py::test_rabbitmq_format_factory_settings    failed  4.070239695000055
    test_storage_rabbitmq/test.py::test_rabbitmq_format_factory_settings    failed  120.14076365499932
    test_storage_rabbitmq/test.py::test_rabbitmq_vhost      passed  0.06514747200162674
    test_storage_rabbitmq/test.py::test_rabbitmq_vhost      failed  4.070600553000986
    test_storage_rabbitmq/test.py::test_rabbitmq_vhost      failed  120.16239366799891

As you can see all tests timed out after
test_rabbitmq_no_connection_at_startup_2 failure.

So let's try to properly pause it using stop_app/start_app.

  [1]: https://s3.amazonaws.com/clickhouse-test-reports/json.html?REF=master&sha=87a5d902c552e9bf1502023611f7341b665754e0&name_0=MasterCI&name_1=Integration%20tests%20%28aarch64%2C%20distributed%20plan%2C%202%2F4%29

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)