split_kafka_tests: works

Author: ilejn

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

@@ -1,31 +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):
     errors = []
@@ -490,3 +464,8 @@ 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

@@ -1,44 +1,16 @@
 """Quick tests, faster than 30 seconds"""
 
-import json
-import logging
-import math
-import random
-import threading
-import time
-
-import avro.datafile
-import avro.io
-import avro.schema
-import kafka.errors
-import pytest
-from confluent_kafka.avro.cached_schema_registry_client import (
-    CachedSchemaRegistryClient,
-)
-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.cluster import ClickHouseCluster, is_arm
-from helpers.network import PartitionManager
-from helpers.test_tools import TSV, assert_eq_with_retry
+from helpers.kafka.common_direct import *
+from helpers.kafka.common_direct import _VarintBytes
+import helpers.kafka.common as k
 
-from . import common as k
-from . import message_with_repeated_pb2
 
 # protoc --version
 # libprotoc 3.0.0
 # # to create kafka_pb2.py
 # protoc --python_out=. kafka.proto
 
 
-if is_arm():
-    pytestmark = pytest.mark.skip
-
 # TODO: add test for run-time offset update in CH, if we manually update it on Kafka side.
 # TODO: add test for SELECT LIMIT is working.
 
@@ -2825,17 +2797,17 @@ def test_issue26643(kafka_cluster, create_query_generator):
     thread_per_consumer = k.must_use_thread_per_consumer(create_query_generator)
 
     with k.kafka_topic(k.get_admin_client(kafka_cluster), topic_name):
-        msg = message_with_repeated_pb2.Message(
+        msg = k.message_with_repeated_pb2.Message(
             tnow=1629000000,
             server="server1",
             clien="host1",
             sPort=443,
             cPort=50000,
             r=[
-                message_with_repeated_pb2.dd(
+                k.message_with_repeated_pb2.dd(
                     name="1", type=444, ttl=123123, data=b"adsfasd"
                 ),
-                message_with_repeated_pb2.dd(name="2"),
+                k.message_with_repeated_pb2.dd(name="2"),
             ],
             method="GET",
         )
@@ -2844,7 +2816,7 @@ def test_issue26643(kafka_cluster, create_query_generator):
         serialized_msg = msg.SerializeToString()
         data = data + _VarintBytes(len(serialized_msg)) + serialized_msg
 
-        msg = message_with_repeated_pb2.Message(tnow=1629000002)
+        msg = k.message_with_repeated_pb2.Message(tnow=1629000002)
 
         serialized_msg = msg.SerializeToString()
         data = data + _VarintBytes(len(serialized_msg)) + serialized_msg