Merge pull request ClickHouse#78624 from ilejn/split_kafka_tests

Split kafka tests (1)

Author: antaljanosbenjamin

tests/integration/test_storage_kafka/common.py renamed to tests/integration/helpers/kafka/common.py

@@ -1,30 +1,5 @@
-import io
-import logging
-import os.path as p
-import random
-import socket
-import string
-import time
-from contextlib import contextmanager
-
-import avro.datafile
-import avro.io
-from confluent_kafka.avro.serializer.message_serializer import MessageSerializer
-from google.protobuf.internal.encoder import _VarintBytes
-from kafka import BrokerConnection, KafkaAdminClient, KafkaConsumer, KafkaProducer
-from kafka.admin import NewTopic
-from kafka.protocol.admin import DescribeGroupsRequest_v1
-from kafka.protocol.group import MemberAssignment
-
-from helpers.client import QueryRuntimeException
-from helpers.test_tools import TSV
-
-from . import kafka_pb2, social_pb2
-
-KAFKA_TOPIC_OLD = "old_t"
-KAFKA_CONSUMER_GROUP_OLD = "old_cg"
-KAFKA_TOPIC_NEW = "new_t"
-KAFKA_CONSUMER_GROUP_NEW = "new_cg"
+from helpers.kafka.common_direct import *
+from helpers.kafka.common_direct import _VarintBytes
 
 
 def get_kafka_producer(port, serializer, retries):
@@ -490,3 +465,9 @@ def describe_consumer_group(kafka_cluster, name):
         member_info["assignment"] = member_topics_assignment
         res.append(member_info)
     return res
+
+
+KAFKA_TOPIC_OLD = "old_t"
+KAFKA_CONSUMER_GROUP_OLD = "old_cg"
+KAFKA_TOPIC_NEW = "new_t"
+KAFKA_CONSUMER_GROUP_NEW = "new_cg"

tests/integration/helpers/kafka/common_direct.py

@@ -0,0 +1,41 @@
+import json
+import io
+import logging
+import math
+import os.path as p
+import random
+import socket
+import string
+import threading
+
+import time
+from contextlib import contextmanager
+
+import pytest
+
+import avro.datafile
+import avro.io
+import avro.schema
+from confluent_kafka.avro.cached_schema_registry_client import (
+    CachedSchemaRegistryClient,
+)
+
+import kafka.errors
+from confluent_kafka.avro.serializer.message_serializer import MessageSerializer
+from kafka import BrokerConnection, KafkaAdminClient, KafkaConsumer, KafkaProducer
+from kafka.admin import NewTopic
+from kafka.protocol.admin import DescribeGroupsRequest_v1
+from kafka.protocol.group import MemberAssignment
+
+from helpers.client import QueryRuntimeException
+from helpers.cluster import ClickHouseCluster, is_arm
+from helpers.network import PartitionManager
+from helpers.test_tools import TSV, assert_eq_with_retry
+
+from helpers.kafka import kafka_pb2, social_pb2, message_with_repeated_pb2
+
+from google.protobuf.internal.encoder import _VarintBytes
+
+
+if is_arm():
+    pytestmark = pytest.mark.skip

tests/integration/test_storage_kafka/kafka_pb2.py renamed to tests/integration/helpers/kafka/kafka_pb2.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_storage_kafka/message_with_repeated_pb2.py renamed to tests/integration/helpers/kafka/message_with_repeated_pb2.py

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