Merge pull request ClickHouse#78987 from dmitry-sles-novikov/fix_test_nats_no_connection_at_startup_1

Fixed flaky test test_nats_no_connection_at_startup_1

Author: yakov-olkhovskiy

tests/integration/helpers/cluster.py

@@ -348,14 +348,14 @@ def rabbitmq_debuginfo(rabbitmq_id, cookie):
 
 async def check_nats_is_available(nats_port, ssl_ctx=None):
     nc = await nats_connect_ssl(
-        nats_port, user="click", password="house", ssl_ctx=ssl_ctx
+        nats_port, user="click", password="house", ssl_ctx=ssl_ctx, max_reconnect_attempts=1
     )
     available = nc.is_connected
     await nc.close()
     return available
 
 
-async def nats_connect_ssl(nats_port, user, password, ssl_ctx=None):
+async def nats_connect_ssl(nats_port, user, password, ssl_ctx=None, **connect_options):
     if not ssl_ctx:
         ssl_ctx = ssl.create_default_context()
         ssl_ctx.check_hostname = False
@@ -365,6 +365,7 @@ async def nats_connect_ssl(nats_port, user, password, ssl_ctx=None):
         user=user,
         password=password,
         tls=ssl_ctx,
+        **connect_options
     )
     return nc
 

tests/integration/test_storage_nats/test.py

@@ -2,6 +2,7 @@
 import json
 import logging
 import math
+import nats
 import os.path as p
 import random
 import subprocess
@@ -46,6 +47,24 @@ def wait_nats_to_start(nats_port, ssl_ctx=None, timeout=180):
         except Exception as ex:
             logging.debug("Can't connect to NATS " + str(ex))
             time.sleep(0.5)
+    
+    assert False, "NATS is unavailable"
+
+# function to check if nats is paused, because in some cases we successfully connected to it after calling pause_container
+def wait_nats_paused(nats_port, ssl_ctx=None, timeout=180):
+    start = time.time()
+    while time.time() - start < timeout:
+        try:
+            asyncio.run(check_nats_is_available(nats_port, ssl_ctx=ssl_ctx))
+            time.sleep(0.5)
+        except nats.errors.NoServersError:
+            logging.debug("NATS is paused")
+            return
+        except Exception as ex:
+            logging.warning("Detect NATS status failed with error \"" + str(ex) + "\" - continue waiting for proper status...")
+            time.sleep(0.5)
+    
+    assert False, "NATS is not paused"
 
 def nats_check_query_result(query, time_limit_sec = 60):
     query_result = ""
@@ -1326,6 +1345,7 @@ def test_nats_restore_failed_connection_without_losses_on_write(nats_cluster):
 
 def test_nats_no_connection_at_startup_1(nats_cluster):
     with nats_cluster.pause_container("nats1"):
+        wait_nats_paused(nats_cluster.nats_port, nats_cluster.nats_ssl_context)
         instance.query_and_get_error(
             """
             CREATE TABLE test.cs (key UInt64, value UInt64)
@@ -1369,6 +1389,7 @@ def test_nats_no_connection_at_startup_2(nats_cluster):
         """
     )
     with nats_cluster.pause_container("nats1"):
+        wait_nats_paused(nats_cluster.nats_port, nats_cluster.nats_ssl_context)
         instance.query("ATTACH TABLE test.cs")
 
     wait_for_table_is_ready(instance, "test.cs")