1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
|
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright 2020 Confluent Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import os
import pytest
from tests.integration.cluster_fixture import TrivupFixture
from tests.integration.cluster_fixture import ByoFixture
work_dir = os.path.dirname(os.path.realpath(__file__))
def create_trivup_cluster():
return TrivupFixture({'with_sr': True,
'debug': True,
'cp_version': 'latest',
'broker_conf': ['transaction.state.log.replication.factor=1',
'transaction.state.log.min.isr=1']})
def create_byo_cluster(conf):
"""
The cluster's bootstrap.servers must be set in dict.
"""
return ByoFixture(conf)
@pytest.fixture(scope="package")
def kafka_cluster():
"""
If BROKERS environment variable is set to a CSV list of bootstrap servers
an existing cluster is used.
Additionally, if SR_URL environment variable is set the Schema-Registry
client will use the given URL.
If BROKERS is not set a TrivUp cluster is created and used.
"""
bootstraps = os.environ.get("BROKERS", "")
if bootstraps != "":
conf = {"bootstrap.servers": bootstraps}
sr_url = os.environ.get("SR_URL", "")
if sr_url != "":
conf["schema.registry.url"] = sr_url
print("Using ByoFixture with config from env variables: ", conf)
cluster = create_byo_cluster(conf)
else:
cluster = create_trivup_cluster()
try:
yield cluster
finally:
cluster.stop()
@pytest.fixture()
def load_file():
def get_handle(name):
with open(os.path.join(work_dir, 'schema_registry', 'data', name)) as fd:
return fd.read()
return get_handle
|
