split_kafka_tests: black formatter

Author: ilejn

tests/integration/helpers/kafka/common.py

@@ -1,6 +1,7 @@
 from helpers.kafka.common_direct import *
 from helpers.kafka.common_direct import _VarintBytes
 
+
 def get_kafka_producer(port, serializer, retries):
     errors = []
     for _ in range(retries):
@@ -465,6 +466,7 @@ def describe_consumer_group(kafka_cluster, name):
         res.append(member_info)
     return res
 
+
 KAFKA_TOPIC_OLD = "old_t"
 KAFKA_CONSUMER_GROUP_OLD = "old_cg"
 KAFKA_TOPIC_NEW = "new_t"

tests/integration/test_kafka_bad_messages/test.py

@@ -6,7 +6,7 @@
 from kafka.admin import NewTopic
 
 from helpers.cluster import ClickHouseCluster, is_arm
-from helpers.kafka_common import (
+from helpers.kafka.common import (
     kafka_create_topic,
     kafka_delete_topic,
     get_kafka_producer,

tests/integration/test_kafka_bad_messages/test_1.py

@@ -1,4 +1,4 @@
-from helpers.kafka_common import (
+from helpers.kafka.common import (
     kafka_create_topic,
     kafka_delete_topic,
     get_kafka_producer,

tests/integration/test_storage_kafka/test_bacth_slow_3.py

@@ -22,6 +22,7 @@
     clickhouse_path_dir="clickhouse_path",
 )
 
+
 # Fixtures
 @pytest.fixture(scope="module")
 def kafka_cluster():
@@ -33,6 +34,7 @@ def kafka_cluster():
     finally:
         cluster.shutdown()
 
+
 @pytest.fixture(autouse=True)
 def kafka_setup_teardown():
     instance.query("DROP DATABASE IF EXISTS test SYNC; CREATE DATABASE test;")
@@ -59,7 +61,8 @@ def get_topics_to_delete():
         retries += 1
         time.sleep(0.5)
     yield  # run test
-        
+
+
 # Tests
 
 # TODO: add test for run-time offset update in CH, if we manually update it on Kafka side.
@@ -398,6 +401,7 @@ def test_kafka_formats_with_broken_message(kafka_cluster, create_query_generator
         ), "Proper error for format: {}".format(format_name)
         k.kafka_delete_topic(admin_client, topic_name)
 
+
 @pytest.mark.parametrize(
     "create_query_generator",
     [k.generate_old_create_table_query, k.generate_new_create_table_query],
@@ -813,6 +817,7 @@ def test_kafka_formats(kafka_cluster, create_query_generator):
         )
         k.kafka_delete_topic(k.get_admin_client(kafka_cluster), topic_name)
 
+
 if __name__ == "__main__":
     cluster.start()
     input("Cluster created, press any key to destroy...")

tests/integration/test_storage_kafka/test_batch_slow_0.py

@@ -22,6 +22,7 @@
     clickhouse_path_dir="clickhouse_path",
 )
 
+
 # Fixtures
 @pytest.fixture(scope="module")
 def kafka_cluster():
@@ -33,6 +34,7 @@ def kafka_cluster():
     finally:
         cluster.shutdown()
 
+
 @pytest.fixture(autouse=True)
 def kafka_setup_teardown():
     instance.query("DROP DATABASE IF EXISTS test SYNC; CREATE DATABASE test;")
@@ -59,7 +61,8 @@ def get_topics_to_delete():
         retries += 1
         time.sleep(0.5)
     yield  # run test
-        
+
+
 # Tests
 
 

tests/integration/test_storage_kafka/test_batch_slow_1.py

@@ -22,6 +22,7 @@
     clickhouse_path_dir="clickhouse_path",
 )
 
+
 # Fixtures
 @pytest.fixture(scope="module")
 def kafka_cluster():
@@ -33,6 +34,7 @@ def kafka_cluster():
     finally:
         cluster.shutdown()
 
+
 @pytest.fixture(autouse=True)
 def kafka_setup_teardown():
     instance.query("DROP DATABASE IF EXISTS test SYNC; CREATE DATABASE test;")
@@ -59,9 +61,11 @@ def get_topics_to_delete():
         retries += 1
         time.sleep(0.5)
     yield  # run test
-        
+
+
 # Tests
 
+
 @pytest.mark.parametrize(
     "create_query_generator",
     [k.generate_old_create_table_query, k.generate_new_create_table_query],

tests/integration/test_storage_kafka/test_batch_slow_2.py

@@ -22,6 +22,7 @@
     clickhouse_path_dir="clickhouse_path",
 )
 
+
 # Fixtures
 @pytest.fixture(scope="module")
 def kafka_cluster():
@@ -33,6 +34,7 @@ def kafka_cluster():
     finally:
         cluster.shutdown()
 
+
 @pytest.fixture(autouse=True)
 def kafka_setup_teardown():
     instance.query("DROP DATABASE IF EXISTS test SYNC; CREATE DATABASE test;")
@@ -60,8 +62,10 @@ def get_topics_to_delete():
         time.sleep(0.5)
     yield  # run test
 
+
 # Tests
 
+
 @pytest.mark.parametrize(
     "create_query_generator",
     [
@@ -259,5 +263,3 @@ def test_kafka_consumer_failover(kafka_cluster, create_query_generator):
         prev_count = instance.query_with_retry(
             count_query, check_callback=lambda res: int(res) > prev_count
         )
-
-

tests/integration/test_storage_kafka/test_batch_slow_4.py

@@ -22,6 +22,7 @@
     clickhouse_path_dir="clickhouse_path",
 )
 
+
 # Fixtures
 @pytest.fixture(scope="module")
 def kafka_cluster():
@@ -33,6 +34,7 @@ def kafka_cluster():
     finally:
         cluster.shutdown()
 
+
 @pytest.fixture(autouse=True)
 def kafka_setup_teardown():
     instance.query("DROP DATABASE IF EXISTS test SYNC; CREATE DATABASE test;")
@@ -60,8 +62,10 @@ def get_topics_to_delete():
         time.sleep(0.5)
     yield  # run test
 
+
 # Tests
 
+
 def test_formats_errors(kafka_cluster):
     admin_client = KafkaAdminClient(
         bootstrap_servers="localhost:{}".format(kafka_cluster.kafka_port)
@@ -151,5 +155,3 @@ def test_formats_errors(kafka_cluster):
 
             instance.query(f"DROP TABLE test.{table_name}")
             instance.query("DROP TABLE test.view")
-
-

tests/integration/test_storage_kafka/test_batch_slow_5.py

@@ -22,6 +22,7 @@
     clickhouse_path_dir="clickhouse_path",
 )
 
+
 # Fixtures
 @pytest.fixture(scope="module")
 def kafka_cluster():
@@ -33,6 +34,7 @@ def kafka_cluster():
     finally:
         cluster.shutdown()
 
+
 @pytest.fixture(autouse=True)
 def kafka_setup_teardown():
     instance.query("DROP DATABASE IF EXISTS test SYNC; CREATE DATABASE test;")
@@ -60,8 +62,10 @@ def get_topics_to_delete():
         time.sleep(0.5)
     yield  # run test
 
+
 # Tests
 
+
 def test_kafka_handling_commit_failure(kafka_cluster):
     messages = [json.dumps({"key": j + 1, "value": "x" * 300}) for j in range(22)]
     k.kafka_produce(kafka_cluster, "handling_commit_failure", messages)
@@ -100,9 +104,7 @@ def test_kafka_handling_commit_failure(kafka_cluster):
     # while materialized view is working to inject zookeeper failure
 
     with kafka_cluster.pause_container("kafka1"):
-        instance.wait_for_log_line(
-            "timeout", timeout=60, look_behind_lines=100
-        )
+        instance.wait_for_log_line("timeout", timeout=60, look_behind_lines=100)
 
     # kafka_cluster.open_bash_shell('instance')
     instance.wait_for_log_line("Committed offset 22")
@@ -363,5 +365,3 @@ def produce():
         kafka_thread.join()
 
         assert result == 1, "Messages from kafka get duplicated!"
-
-