mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-09-20 00:30:49 +00:00
81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
import io
|
|
import logging
|
|
|
|
import avro.schema
|
|
import pytest
|
|
from confluent_kafka.avro.cached_schema_registry_client import CachedSchemaRegistryClient
|
|
from confluent_kafka.avro.serializer.message_serializer import MessageSerializer
|
|
from helpers.cluster import ClickHouseCluster, ClickHouseInstance
|
|
|
|
logging.getLogger().setLevel(logging.INFO)
|
|
logging.getLogger().addHandler(logging.StreamHandler())
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def started_cluster():
|
|
try:
|
|
cluster = ClickHouseCluster(__file__)
|
|
cluster.add_instance("dummy", with_kafka=True)
|
|
logging.info("Starting cluster...")
|
|
cluster.start()
|
|
logging.info("Cluster started")
|
|
|
|
yield cluster
|
|
finally:
|
|
cluster.shutdown()
|
|
|
|
|
|
def run_query(instance, query, data=None, settings=None):
|
|
# type: (ClickHouseInstance, str, object, dict) -> str
|
|
|
|
logging.info("Running query '{}'...".format(query))
|
|
# use http to force parsing on server
|
|
if not data:
|
|
data = " " # make POST request
|
|
result = instance.http_query(query, data=data, params=settings)
|
|
logging.info("Query finished")
|
|
|
|
return result
|
|
|
|
|
|
def test_select(started_cluster):
|
|
# type: (ClickHouseCluster) -> None
|
|
|
|
schema_registry_client = CachedSchemaRegistryClient('http://localhost:{}'.format(started_cluster.schema_registry_port))
|
|
serializer = MessageSerializer(schema_registry_client)
|
|
|
|
schema = avro.schema.make_avsc_object({
|
|
'name': 'test_record',
|
|
'type': 'record',
|
|
'fields': [
|
|
{
|
|
'name': 'value',
|
|
'type': 'long'
|
|
}
|
|
]
|
|
})
|
|
|
|
buf = io.BytesIO()
|
|
for x in range(0, 3):
|
|
message = serializer.encode_record_with_schema(
|
|
'test_subject', schema, {'value': x}
|
|
)
|
|
buf.write(message)
|
|
data = buf.getvalue()
|
|
|
|
instance = started_cluster.instances["dummy"] # type: ClickHouseInstance
|
|
schema_registry_url = "http://{}:{}".format(
|
|
started_cluster.schema_registry_host,
|
|
8081
|
|
)
|
|
|
|
run_query(instance, "create table avro_data(value Int64) engine = Memory()")
|
|
settings = {'format_avro_schema_registry_url': schema_registry_url}
|
|
run_query(instance, "insert into avro_data format AvroConfluent", data, settings)
|
|
stdout = run_query(instance, "select * from avro_data")
|
|
assert list(map(str.split, stdout.splitlines())) == [
|
|
["0"],
|
|
["1"],
|
|
["2"],
|
|
]
|