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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
#!/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.common import TestUtils
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(conf={}):
trivup_fixture_conf = {'with_sr': True,
'debug': True,
'cp_version': '7.6.0',
'kraft': TestUtils.use_kraft(),
'version': TestUtils.broker_version(),
'broker_conf': TestUtils.broker_conf()}
trivup_fixture_conf.update(conf)
return TrivupFixture(trivup_fixture_conf)
def create_sasl_cluster(conf={}):
trivup_fixture_conf = {'with_sr': False,
'version': TestUtils.broker_version(),
'sasl_mechanism': "PLAIN",
'kraft': TestUtils.use_kraft(),
'sasl_users': 'sasl_user=sasl_user',
'debug': True,
'cp_version': 'latest',
'broker_conf': TestUtils.broker_conf()}
trivup_fixture_conf.update(conf)
return TrivupFixture(trivup_fixture_conf)
def create_byo_cluster(conf):
"""
The cluster's bootstrap.servers must be set in dict.
"""
return ByoFixture(conf)
def kafka_cluster_fixture(
brokers_env="BROKERS",
sr_url_env="SR_URL",
trivup_cluster_conf={}
):
"""
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_env, "")
if bootstraps != "":
conf = {"bootstrap.servers": bootstraps}
sr_url = os.environ.get(sr_url_env, "")
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(trivup_cluster_conf)
try:
yield cluster
finally:
cluster.stop()
def sasl_cluster_fixture(
trivup_cluster_conf={}
):
"""
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.
"""
cluster = create_sasl_cluster(trivup_cluster_conf)
try:
yield cluster
finally:
cluster.stop()
@pytest.fixture(scope="session")
def kafka_cluster():
for fixture in kafka_cluster_fixture():
yield fixture
@pytest.fixture(scope="session")
def sasl_cluster(request):
for fixture in sasl_cluster_fixture(request.param):
yield fixture
@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
|
