Improve e2e Kafka testing

tests/e2e/streams/common.py

@@ -299,7 +299,7 @@ def test_start_checked_stream_after_timeout(connection, stream_creator):
     stream_name = "test_start_checked_stream_after_timeout"
     execute_and_fetch_all(cursor, stream_creator(stream_name))
 
-    timeout_in_ms = 2000
+    timeout_in_ms = 5000
 
     def call_check():
         execute_and_fetch_all(connect().cursor(), f"CHECK STREAM {stream_name} TIMEOUT {timeout_in_ms}")
@@ -447,7 +447,6 @@ def test_check_stream_different_number_of_queries_than_messages(connection, stre
 
 def test_start_stream_with_batch_limit(connection, stream_name, stream_creator, messages_sender):
     BATCH_LIMIT = 5
-    TIMEOUT = 10000
 
     cursor = connection.cursor()
     execute_and_fetch_all(cursor, stream_creator())
@@ -457,7 +456,7 @@ def test_start_stream_with_batch_limit(connection, stream_name, stream_creator,
     def start_new_stream_with_limit():
         connection = connect()
         cursor = connection.cursor()
-        start_stream_with_limit(cursor, stream_name, BATCH_LIMIT, TIMEOUT)
+        start_stream_with_limit(cursor, stream_name, BATCH_LIMIT)
 
     def is_running():
         return get_is_running(cursor, stream_name)
@@ -465,7 +464,6 @@ def test_start_stream_with_batch_limit(connection, stream_name, stream_creator,
     thread_stream_running = Process(target=start_new_stream_with_limit)
     thread_stream_running.start()
 
-    execute_and_fetch_all(connection.cursor(), "SHOW STREAMS")
     assert mg_sleep_and_assert(True, is_running)
     messages_sender(BATCH_LIMIT - 1)
 

tests/e2e/streams/kafka_streams_tests.py

@@ -23,6 +23,7 @@ from mg_utils import mg_sleep_and_assert
 TRANSFORMATIONS_TO_CHECK_C = ["c_transformations.empty_transformation"]
 TRANSFORMATIONS_TO_CHECK_PY = ["kafka_transform.simple", "kafka_transform.with_parameters"]
 KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT = 60
+KAFKA_PRODUCER_SENDING_MSG_LARGE_TIMEOUT = 6000
 
 
 @pytest.mark.parametrize("transformation", TRANSFORMATIONS_TO_CHECK_PY)
@@ -448,7 +449,7 @@ def test_start_stream_with_batch_limit_while_check_running(kafka_producer, kafka
         )
 
     def message_sender(message):
-        kafka_producer.send(kafka_topics[0], message).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
+        kafka_producer.send(kafka_topics[0], message).get(timeout=KAFKA_PRODUCER_SENDING_MSG_LARGE_TIMEOUT)
 
     def setup_function(start_check_stream, cursor, stream_name, batch_limit, timeout):
         thread_stream_check = Process(target=start_check_stream, daemon=True, args=(stream_name, batch_limit, timeout))
@@ -475,7 +476,7 @@ def test_check_while_stream_with_batch_limit_running(kafka_producer, kafka_topic
         )
 
     def message_sender(message):
-        kafka_producer.send(kafka_topics[0], message).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
+        kafka_producer.send(kafka_topics[0], message).get(timeout=KAFKA_PRODUCER_SENDING_MSG_LARGE_TIMEOUT)
 
     common.test_check_while_stream_with_batch_limit_running(connection, stream_creator, message_sender)