Skip to content

Commit 95b9a53

Browse files
timkpainetimkpaine
committed
Setup CI for adapter with service dependencies
Signed-off-by: Tim Paine <[email protected]>
1 parent 90739ab commit 95b9a53

File tree

7 files changed

+131
-106
lines changed

7 files changed

+131
-106
lines changed

‎.github/workflows/build.yml‎

Lines changed: 73 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -554,7 +554,6 @@ jobs:
554554
env:
555555
CSP_TEST_SKIP_EXAMPLES: "1"
556556

557-
558557
#################################
559558
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~#
560559
#~~~~~~~~~|##########|~~~~~~~~~~#
@@ -620,6 +619,79 @@ jobs:
620619
run: make test TEST_ARGS="-k TestDBReader"
621620
if: ${{ contains( 'sqlalchemy', matrix.package )}}
622621

622+
###########################
623+
#~~~~~~~~~~~~~~~~~~~~~~~~~#
624+
#~~~~~~|#############|~~~~#
625+
#~~~~~~|#|~~~~~~~/##/~~~~~#
626+
#~~~~~~|#|~~~~~/##/~~~~~~~#
627+
#~~~~~~~~~~~~/##/~~~~~~~~~#
628+
#~~~~~~~~~~/##/~~~~~~~~~~~#
629+
#~~~~~~~~/##/~~~~~~~~~~~~~#
630+
#~~~~~~/##/~~~~~~~~~~~~~~~#
631+
#~~~~~~~~~~~~~~~~~~~~~~~~~#
632+
# Test Service Adapters #
633+
#~~~~~~~~~~~~~~~~~~~~~~~~~#
634+
test_adapters:
635+
needs:
636+
- initialize
637+
- build
638+
639+
strategy:
640+
matrix:
641+
os:
642+
- ubuntu-20.04
643+
python-version:
644+
- 3.9
645+
adapter:
646+
- kafka
647+
648+
runs-on: ${{ matrix.os }}
649+
650+
steps:
651+
- name: Checkout
652+
uses: actions/checkout@v4
653+
with:
654+
submodules: recursive
655+
656+
- name: Set up Python ${{ matrix.python-version }}
657+
uses: ./.github/actions/setup-python
658+
with:
659+
version: '${{ matrix.python-version }}'
660+
661+
- name: Install python dependencies
662+
run: make requirements
663+
664+
- name: Install test dependencies
665+
shell: bash
666+
run: sudo apt-get install graphviz
667+
668+
# Download artifact
669+
- name: Download wheel
670+
uses: actions/download-artifact@v4
671+
with:
672+
name: csp-dist-${{ runner.os }}-${{ runner.arch }}-${{ matrix.python-version }}
673+
674+
- name: Install wheel
675+
run: python -m pip install -U *manylinux2014*.whl --target .
676+
677+
- name: Spin up adapter service
678+
run: make dockerup ADAPTER=${{ matrix.adapter }} DOCKERARGS="--wait --wait-timeout 30"
679+
680+
- name: Wait a few seconds after images have been spun up
681+
run: sleep 30
682+
683+
# Run tests
684+
- name: Setup test flags
685+
shell: bash
686+
run: echo "CSP_TEST_$( echo ${{ matrix.adapter }} | awk '{print toupper($0)}' )=1" >> $GITHUB_ENV
687+
688+
- name: Python Test Steps
689+
run: make test TEST_ARGS="-k ${{ matrix.adapter }}"
690+
691+
- name: Spin down adapter service
692+
run: make dockerdown ADAPTER=${{ matrix.adapter }}
693+
if: ${{ always() }}
694+
623695
#############################
624696
#~~~~~~~~~~~~~~~~~~~~~~~~~~~#
625697
#~~~~~~|#############|~~~~~~#

‎Makefile‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -86,21 +86,22 @@ tests: test
8686

8787
.PHONY: dockerup dockerps dockerdown initpodmanmac
8888
ADAPTER := kafka
89-
DOCKER := podman
89+
DOCKER_COMPOSE := docker compose # or podman-compose
90+
DOCKERARGS :=
9091

9192
initpodmanmac:
9293
podman machine stop
9394
podman machine set --cpus 4 --memory 8096
9495
podman machine start
9596

9697
dockerup: ## spin up docker compose services for adapter testing
97-
$(DOCKER) compose -f ci/$(ADAPTER)/docker-compose.yml up -d
98+
$(DOCKER_COMPOSE) -f ci/$(ADAPTER)/docker-compose.yml up -d $(DOCKERARGS)
9899

99100
dockerps: ## spin up docker compose services for adapter testing
100-
$(DOCKER) compose -f ci/$(ADAPTER)/docker-compose.yml ps
101+
$(DOCKER_COMPOSE) -f ci/$(ADAPTER)/docker-compose.yml ps
101102

102103
dockerdown: ## spin up docker compose services for adapter testing
103-
$(DOCKER) compose -f ci/$(ADAPTER)/docker-compose.yml down
104+
$(DOCKER_COMPOSE) -f ci/$(ADAPTER)/docker-compose.yml down
104105

105106
###########
106107
# VERSION #

‎ci/kafka/docker-compose.yml‎

Lines changed: 25 additions & 79 deletions
Original file line numberDiff line numberDiff line change
@@ -87,84 +87,6 @@ services:
8787
CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"
8888
CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=ERROR,org.I0Itec.zkclient=ERROR,org.reflections=ERROR
8989

90-
control-center:
91-
image: confluentinc/cp-enterprise-control-center:7.5.3
92-
hostname: control-center
93-
container_name: control-center
94-
depends_on:
95-
- broker
96-
- schema-registry
97-
- connect
98-
- ksqldb-server
99-
ports:
100-
- "9021:9021"
101-
environment:
102-
CONTROL_CENTER_BOOTSTRAP_SERVERS: 'broker:29092'
103-
CONTROL_CENTER_CONNECT_CONNECT-DEFAULT_CLUSTER: 'connect:8083'
104-
CONTROL_CENTER_KSQL_KSQLDB1_URL: "http://ksqldb-server:8088"
105-
CONTROL_CENTER_KSQL_KSQLDB1_ADVERTISED_URL: "http://localhost:8088"
106-
CONTROL_CENTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
107-
CONTROL_CENTER_REPLICATION_FACTOR: 1
108-
CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS: 1
109-
CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1
110-
CONFLUENT_METRICS_TOPIC_REPLICATION: 1
111-
PORT: 9021
112-
113-
ksqldb-server:
114-
image: confluentinc/cp-ksqldb-server:7.5.3
115-
hostname: ksqldb-server
116-
container_name: ksqldb-server
117-
depends_on:
118-
- broker
119-
- connect
120-
ports:
121-
- "8088:8088"
122-
environment:
123-
KSQL_CONFIG_DIR: "/etc/ksql"
124-
KSQL_BOOTSTRAP_SERVERS: "broker:29092"
125-
KSQL_HOST_NAME: ksqldb-server
126-
KSQL_LISTENERS: "http://0.0.0.0:8088"
127-
KSQL_CACHE_MAX_BYTES_BUFFERING: 0
128-
KSQL_KSQL_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
129-
KSQL_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"
130-
KSQL_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor"
131-
KSQL_KSQL_CONNECT_URL: "http://connect:8083"
132-
KSQL_KSQL_LOGGING_PROCESSING_TOPIC_REPLICATION_FACTOR: 1
133-
KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: 'true'
134-
KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: 'true'
135-
136-
# ksqldb-cli:
137-
# image: confluentinc/cp-ksqldb-cli:7.5.3
138-
# container_name: ksqldb-cli
139-
# depends_on:
140-
# - broker
141-
# - connect
142-
# - ksqldb-server
143-
# entrypoint: /bin/sh
144-
# tty: true
145-
146-
# ksql-datagen:
147-
# image: confluentinc/ksqldb-examples:7.5.3
148-
# hostname: ksql-datagen
149-
# container_name: ksql-datagen
150-
# depends_on:
151-
# - ksqldb-server
152-
# - broker
153-
# - schema-registry
154-
# - connect
155-
# command: "bash -c 'echo Waiting for Kafka to be ready... && \
156-
# cub kafka-ready -b broker:29092 1 40 && \
157-
# echo Waiting for Confluent Schema Registry to be ready... && \
158-
# cub sr-ready schema-registry 8081 40 && \
159-
# echo Waiting a few seconds for topic creation to finish... && \
160-
# sleep 11 && \
161-
# tail -f /dev/null'"
162-
# environment:
163-
# KSQL_CONFIG_DIR: "/etc/ksql"
164-
# STREAMS_BOOTSTRAP_SERVERS: broker:29092
165-
# STREAMS_SCHEMA_REGISTRY_HOST: schema-registry
166-
# STREAMS_SCHEMA_REGISTRY_PORT: 8081
167-
16890
rest-proxy:
16991
image: confluentinc/cp-kafka-rest:7.5.3
17092
depends_on:
@@ -178,4 +100,28 @@ services:
178100
KAFKA_REST_HOST_NAME: rest-proxy
179101
KAFKA_REST_BOOTSTRAP_SERVERS: 'broker:29092'
180102
KAFKA_REST_LISTENERS: "http://0.0.0.0:8082"
181-
KAFKA_REST_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'
103+
KAFKA_REST_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'
104+
105+
# Uncomment for a helpful UI
106+
# control-center:
107+
# image: confluentinc/cp-enterprise-control-center:7.5.3
108+
# hostname: control-center
109+
# container_name: control-center
110+
# depends_on:
111+
# - broker
112+
# - schema-registry
113+
# - connect
114+
# ports:
115+
# - "9021:9021"
116+
# environment:
117+
# CONTROL_CENTER_BOOTSTRAP_SERVERS: 'broker:29092'
118+
# CONTROL_CENTER_CONNECT_CONNECT-DEFAULT_CLUSTER: 'connect:8083'
119+
# CONTROL_CENTER_KSQL_KSQLDB1_URL: "http://ksqldb-server:8088"
120+
# CONTROL_CENTER_KSQL_KSQLDB1_ADVERTISED_URL: "http://localhost:8088"
121+
# CONTROL_CENTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
122+
# CONTROL_CENTER_REPLICATION_FACTOR: 1
123+
# CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS: 1
124+
# CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1
125+
# CONFLUENT_METRICS_TOPIC_REPLICATION: 1
126+
# PORT: 9021
127+

‎csp/adapters/kafka.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ def __init__(
7373

7474
consumer_properties = {
7575
"group.id": group_id,
76-
# To get end of parition notification for live / not live flag
76+
# To get end of partition notification for live / not live flag
7777
"enable.partition.eof": "true",
7878
}
7979

‎csp/tests/adapters/conftest.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,6 @@ def kafkabroker():
1212
def kafkaadapter(kafkabroker):
1313
group_id = "group.id123"
1414
_kafkaadapter = KafkaAdapterManager(
15-
broker=kafkabroker, group_id=group_id, rd_kafka_conf_options={"allow.auto.create.topics": "true"}
15+
broker=kafkabroker, group_id=group_id
1616
)
1717
return _kafkaadapter

‎csp/tests/adapters/test_kafka.py‎

Lines changed: 25 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -79,11 +79,10 @@ def graph(count: int):
7979
}
8080

8181
topic = f"test.metadata.{os.getpid()}"
82-
_precreate_topic(topic)
8382
subKey = "foo"
8483
pubKey = ["mapped_a", "mapped_b", "mapped_c"]
8584

86-
c = csp.count(csp.timer(timedelta(seconds=0.1)))
85+
c = csp.count(csp.timer(timedelta(seconds=1)))
8786
t = csp.sample(c, csp.const("foo"))
8887

8988
pubStruct = MetaPubData.collectts(
@@ -104,22 +103,27 @@ def graph(count: int):
104103
)
105104

106105
csp.add_graph_output("sub_data", sub_data)
107-
# csp.print('sub', sub_data)
106+
csp.print('sub', sub_data)
108107
# Wait for at least count ticks and until we get a live tick
109-
done_flag = csp.count(sub_data) >= count
110-
done_flag = csp.and_(done_flag, sub_data.mapped_live is True)
108+
done_flag = csp.and_(csp.count(sub_data) >= count, sub_data.mapped_live == True) # noqa: E712
111109
stop = csp.filter(done_flag, done_flag)
112110
csp.stop_engine(stop)
113111

114-
count = 5
115-
results = csp.run(graph, count, starttime=datetime.utcnow(), endtime=timedelta(seconds=30), realtime=True)
112+
# warm up the topic
113+
results = csp.run(graph, 1, starttime=datetime.utcnow(), endtime=timedelta(seconds=3), realtime=True)
114+
115+
# now send some live in
116+
results = csp.run(graph, 5, starttime=datetime.utcnow(), endtime=timedelta(seconds=20), realtime=True)
116117
assert len(results["sub_data"]) >= 5
117118
print(results)
118119
for result in results["sub_data"]:
119120
assert result[1].mapped_partition >= 0
120121
assert result[1].mapped_offset >= 0
121122
assert result[1].mapped_live is not None
122123
assert result[1].mapped_timestamp < datetime.utcnow()
124+
# first record should be non live
125+
assert results["sub_data"][0][1].mapped_live is False
126+
# last record should be live
123127
assert results["sub_data"][-1][1].mapped_live
124128

125129
@pytest.mark.skipif(not os.environ.get("CSP_TEST_KAFKA"), reason="Skipping kafka adapter tests")
@@ -145,8 +149,7 @@ def graph(symbols: list, count: int):
145149
struct_field_map = {"b": "b2", "i": "i2", "d": "d2", "s": "s2", "dt": "dt2"}
146150

147151
done_flags = []
148-
topic = f"mktdata.{os.getpid()}"
149-
_precreate_topic(topic)
152+
150153
for symbol in symbols:
151154
kafkaadapter.publish(msg_mapper, topic, symbol, b, field_map="b")
152155
kafkaadapter.publish(msg_mapper, topic, symbol, i, field_map="i")
@@ -183,10 +186,12 @@ def graph(symbols: list, count: int):
183186
stop = csp.filter(stop, stop)
184187
csp.stop_engine(stop)
185188

189+
topic = f"mktdata.{os.getpid()}"
190+
_precreate_topic(topic)
186191
symbols = ["AAPL", "MSFT"]
187192
count = 100
188193
results = csp.run(
189-
graph, symbols, count, starttime=datetime.utcnow(), endtime=timedelta(seconds=30), realtime=True
194+
graph, symbols, count, starttime=datetime.utcnow(), endtime=timedelta(seconds=10), realtime=True
190195
)
191196
for symbol in symbols:
192197
pub = results[f"pall_{symbol}"]
@@ -212,7 +217,7 @@ def pub_graph():
212217
csp.stop_engine(stop)
213218
# csp.print('pub', struct)
214219

215-
csp.run(pub_graph, starttime=datetime.utcnow(), endtime=timedelta(seconds=30), realtime=True)
220+
csp.run(pub_graph, starttime=datetime.utcnow(), endtime=timedelta(seconds=10), realtime=True)
216221

217222
# grab start/end times
218223
def get_times_graph():
@@ -232,7 +237,7 @@ def get_times_graph():
232237
# csp.print('sub', data)
233238
# csp.print('status', kafkaadapter.status())
234239

235-
all_data = csp.run(get_times_graph, starttime=datetime.utcnow(), endtime=timedelta(seconds=30), realtime=True)[
240+
all_data = csp.run(get_times_graph, starttime=datetime.utcnow(), endtime=timedelta(seconds=10), realtime=True)[
236241
"data"
237242
]
238243
min_time = all_data[0][1].dt
@@ -258,7 +263,7 @@ def get_data(start_offset, expected_count):
258263
KafkaStartOffset.EARLIEST,
259264
10,
260265
starttime=datetime.utcnow(),
261-
endtime=timedelta(seconds=30),
266+
endtime=timedelta(seconds=10),
262267
realtime=True,
263268
)["data"]
264269
# print(res)
@@ -276,7 +281,7 @@ def get_data(start_offset, expected_count):
276281
assert len(res) == 0
277282

278283
res = csp.run(
279-
get_data, KafkaStartOffset.START_TIME, 10, starttime=min_time, endtime=timedelta(seconds=30), realtime=True
284+
get_data, KafkaStartOffset.START_TIME, 10, starttime=min_time, endtime=timedelta(seconds=10), realtime=True
280285
)["data"]
281286
assert len(res) == 10
282287

@@ -287,12 +292,12 @@ def get_data(start_offset, expected_count):
287292
stime = all_data[2][1].dt + timedelta(milliseconds=1)
288293
expected = [x for x in all_data if x[1].dt >= stime]
289294
res = csp.run(
290-
get_data, stime, len(expected), starttime=datetime.utcnow(), endtime=timedelta(seconds=30), realtime=True
295+
get_data, stime, len(expected), starttime=datetime.utcnow(), endtime=timedelta(seconds=10), realtime=True
291296
)["data"]
292297
assert len(res) == len(expected)
293298

294299
res = csp.run(
295-
get_data, timedelta(seconds=0), len(expected), starttime=stime, endtime=timedelta(seconds=30), realtime=True
300+
get_data, timedelta(seconds=0), len(expected), starttime=stime, endtime=timedelta(seconds=10), realtime=True
296301
)["data"]
297302
assert len(res) == len(expected)
298303

@@ -314,8 +319,6 @@ def graph(symbols: list, count: int):
314319
msg_mapper = RawBytesMessageMapper()
315320

316321
done_flags = []
317-
topic = f"test_str.{os.getpid()}"
318-
_precreate_topic(topic)
319322
for symbol in symbols:
320323
topic = f"test_str.{os.getpid()}"
321324
kafkaadapter.publish(msg_mapper, topic, symbol, d)
@@ -356,10 +359,13 @@ def graph(symbols: list, count: int):
356359
stop = csp.filter(stop, stop)
357360
csp.stop_engine(stop)
358361

362+
topic = f"test_str.{os.getpid()}"
363+
_precreate_topic(topic)
364+
359365
symbols = ["AAPL", "MSFT"]
360366
count = 10
361367
results = csp.run(
362-
graph, symbols, count, starttime=datetime.utcnow(), endtime=timedelta(seconds=30), realtime=True
368+
graph, symbols, count, starttime=datetime.utcnow(), endtime=timedelta(seconds=10), realtime=True
363369
)
364370
# print(results)
365371
for symbol in symbols:

0 commit comments

Comments
 (0)