Skip to content

Commit 116d76d

Browse files
amotlamotl
committed
Tests: Add integration tests for CrateDB destination
1 parent 0ea82af commit 116d76d

18 files changed

Lines changed: 211 additions & 13 deletions

‎pyproject.toml‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,7 @@ dependencies = [
8585
"databricks-sqlalchemy<3",
8686
"dataclasses-json<0.7",
8787
"dlt<1.29",
88-
"dlt-cratedb<0.2",
88+
"dlt-cratedb @ git+https://github.com/zerotired/dlt-cratedb.git@fix-dynamic-install",
8989
"duckdb<1.6",
9090
"duckdb-engine<0.18",
9191
"elasticsearch<10",
@@ -126,6 +126,7 @@ dependencies = [
126126
"sqlalchemy-bigquery<2",
127127
"sqlalchemy-cratedb<1",
128128
"sqlalchemy-hana<5",
129+
"sqlalchemy-postgresql-relaxed<0.2",
129130
"sqlalchemy-redshift<2",
130131
"sqlalchemy-spanner<2",
131132
"stripe<16",

‎tests/stream/test_kafka.py‎

Lines changed: 29 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -52,18 +52,30 @@ def get_output_table():
5252

5353
res = get_output_table()
5454
assert len(res) == 3
55-
assert res[0] == ("message1",)
56-
assert res[1] == ("message2",)
57-
assert res[2] == ("message3",)
55+
if dest_uri.startswith("cratedb://"):
56+
messages_db = [res[0][0], res[1][0], res[2][0]]
57+
assert "message1" in messages_db
58+
assert "message2" in messages_db
59+
assert "message3" in messages_db
60+
else:
61+
assert res[0] == ("message1",)
62+
assert res[1] == ("message2",)
63+
assert res[2] == ("message3",)
5864

5965
# run again, nothing should be inserted into the output table
6066
run()
6167

6268
res = get_output_table()
6369
assert len(res) == 3
64-
assert res[0] == ("message1",)
65-
assert res[1] == ("message2",)
66-
assert res[2] == ("message3",)
70+
if dest_uri.startswith("cratedb://"):
71+
messages_db = [res[0][0], res[1][0], res[2][0]]
72+
assert "message1" in messages_db
73+
assert "message2" in messages_db
74+
assert "message3" in messages_db
75+
else:
76+
assert res[0] == ("message1",)
77+
assert res[1] == ("message2",)
78+
assert res[2] == ("message3",)
6779

6880
# add a new message
6981
producer.produce(topic, "message4".encode("utf-8"))
@@ -73,10 +85,17 @@ def get_output_table():
7385
run()
7486
res = get_output_table()
7587
assert len(res) == 4
76-
assert res[0] == ("message1",)
77-
assert res[1] == ("message2",)
78-
assert res[2] == ("message3",)
79-
assert res[3] == ("message4",)
88+
if dest_uri.startswith("cratedb://"):
89+
messages_db = [res[0][0], res[1][0], res[2][0], res[3][0]]
90+
assert "message1" in messages_db
91+
assert "message2" in messages_db
92+
assert "message3" in messages_db
93+
assert "message4" in messages_db
94+
else:
95+
assert res[0] == ("message1",)
96+
assert res[1] == ("message2",)
97+
assert res[2] == ("message3",)
98+
assert res[3] == ("message4",)
8099

81100

82101
@pytest.mark.parametrize(

‎tests/stream/test_mqbridge_kafka.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,12 @@ def test_mqbridge_kafka_to_db(kafka, dest, topic):
5555
with ThreadPoolExecutor() as executor:
5656
dest_uri = executor.submit(dest.start).result()
5757

58+
if dest_uri.startswith("cratedb://"):
59+
pytest.skip(
60+
"Fails on CrateDB with `DestinationSchemaTampered`, see "
61+
"https://github.com/crate/dlt-cratedb/issues/14"
62+
)
63+
5864
_produce(kafka, topic, ROWS)
5965

6066
def run():

‎tests/util/__init__.py‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
from typer.testing import CliRunner
77

88
from omniload.main import app
9+
from tests.util.db import dbquery
910

1011
# Sentinel for a required-but-positionally-late argument. `dest_uri` must stay
1112
# effectively required (omitting it is a caller bug), but `source_table` sits
@@ -189,6 +190,15 @@ def invoke_ingest_command(
189190
app,
190191
args,
191192
)
193+
194+
if result.exit_code == 0:
195+
post_ingest(
196+
source_uri=source_uri,
197+
source_table=source_table,
198+
dest_uri=dest_uri,
199+
dest_table=dest_table,
200+
)
201+
192202
if result.exit_code != 0 and print_output:
193203
if result.exc_info is not None:
194204
traceback.print_exception(*result.exc_info)
@@ -218,9 +228,28 @@ def __init__(self, exit_code, stdout, stderr, exc_info=None):
218228

219229
result = Result(process.returncode, process.stdout, process.stderr)
220230

231+
if result.exit_code == 0:
232+
post_ingest(
233+
source_uri=source_uri,
234+
source_table=source_table,
235+
dest_uri=dest_uri,
236+
dest_table=dest_table,
237+
)
238+
221239
if result.exit_code != 0 and print_output:
222240
print(result.stdout)
223241
print(result.stderr)
224242
# traceback.print_exception(result.exc_info)
225243

226244
return result
245+
246+
247+
def post_ingest(
248+
source_uri=None,
249+
source_table=None,
250+
dest_uri=None,
251+
dest_table=None,
252+
):
253+
# CrateDB needs an explicit flush to make data available for reads immediately.
254+
if dest_uri and dest_uri.startswith("cratedb://"):
255+
dbquery(dest_uri, f"REFRESH TABLE {dest_table}")

‎tests/util/container/model.py‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
from typing import Callable, Optional
88

99
import pytest
10+
from testcontainers.community.cratedb import CrateDBContainer
1011
from testcontainers.community.generic import SqlContainer
1112
from testcontainers.community.kafka import KafkaContainer
1213
from testcontainers.core.container import DockerContainer
@@ -213,7 +214,10 @@ def get_connection_url(self):
213214
return self.container.get_bootstrap_server()
214215
elif isinstance(self.container, CouchbaseContainer):
215216
return self.container.get_connection_url() + self.connection_suffix
216-
elif isinstance(self.container, DbContainer):
217+
elif isinstance(self.container, CrateDBContainer):
218+
uri = f"cratedb://{self.container.username}:{self.container.password}@{self.container.get_container_host_ip()}:{self.container.get_exposed_port(5432)}"
219+
return uri + self.connection_suffix
220+
elif isinstance(self.container, (DbContainer, SqlContainer)):
217221
return self.container.get_connection_url() + self.connection_suffix
218222
raise ValueError("Unable to get connection url")
219223

‎tests/util/db.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,12 @@ def dbquery(
88
) -> Union[Sequence[Union[sa.Row[Any], sa.RowMapping]], None]:
99
"""Query database using SQLAlchemy and optionally return results."""
1010

11+
# CrateDB needs a relaxed SQLAlchemy dialect for querying.
12+
# It will not support advanced features of CrateDB,
13+
# but that's okay in this case.
14+
if uri.startswith("cratedb://"):
15+
uri = uri.replace("cratedb://", "postgresql+psycopg_relaxed://")
16+
1117
engine = sa.create_engine(uri, poolclass=sa.NullPool)
1218
response = None
1319
with engine.connect() as conn:

‎tests/warehouse/db/test_arrow.py‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -126,6 +126,12 @@ def run_command(df: pd.DataFrame, incremental_key: Optional[str] = None):
126126

127127
dest_uri = dest.start()
128128

129+
if dest_uri.startswith("cratedb://"):
130+
pytest.skip(
131+
"Fails on CrateDB with `DestinationSchemaTampered`, see "
132+
"https://github.com/crate/dlt-cratedb/issues/14"
133+
)
134+
129135
# let's start with a basic dataframe
130136
row_count = 1000
131137
df = pd.DataFrame(
@@ -265,6 +271,12 @@ def run_command(df: pd.DataFrame):
265271

266272
dest_uri = dest.start()
267273

274+
if dest_uri.startswith("cratedb://"):
275+
pytest.skip(
276+
"Fails on CrateDB with `DestinationSchemaTampered`, see "
277+
"https://github.com/crate/dlt-cratedb/issues/14"
278+
)
279+
268280
# let's start with a basic dataframe
269281
row_count = 1000
270282
df = pd.DataFrame({"id": range(row_count), "value": ["a"] * row_count})

‎tests/warehouse/db/test_cratedb.py‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
import unittest
2+
3+
import dlt
4+
from sqlalchemy.util import classproperty
5+
6+
from omniload.target.cratedb import CrateDBDestination
7+
from tests.main.test_targets import GenericSqlDestinationFixture
8+
9+
10+
class CrateDBDestinationTest(unittest.TestCase, GenericSqlDestinationFixture):
11+
destination = CrateDBDestination()
12+
13+
@classproperty
14+
def expected_class(cls):
15+
return dlt.destinations.cratedb # ty: ignore[unresolved-attribute]

‎tests/warehouse/db/test_dynamodb.py‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -121,6 +121,12 @@ def smoke_test(dest_uri, dynamodb, schema: str):
121121
assert result[i][1] == pendulum.parse(dynamodb.data[i]["updated_at"])
122122

123123
def append_test(dest_uri, dynamodb, schema: str):
124+
125+
if dest_uri.startswith("cratedb://"):
126+
pytest.skip(
127+
"Fails on CrateDB with twice the amount of expected results: AssertionError: assert 6 == 3"
128+
)
129+
124130
dest_table = f"{schema}.dynamodb_{get_random_string(5)}"
125131

126132
# we run it twice to assert that the data in destination doesn't change
@@ -145,6 +151,13 @@ def append_test(dest_uri, dynamodb, schema: str):
145151

146152
def incremental_test_factory(strategy):
147153
def incremental_test(dest_uri, dynamodb, schema: str):
154+
155+
if dest_uri.startswith("cratedb://"):
156+
pytest.skip(
157+
"Fails on CrateDB with `DestinationSchemaTampered`, see "
158+
"https://github.com/crate/dlt-cratedb/issues/14"
159+
)
160+
148161
dest_table = f"{schema}.dynamodb_{get_random_string(5)}"
149162

150163
result = invoke_ingest_command(

‎tests/warehouse/db/test_mongodb.py‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -362,6 +362,12 @@ def test_mongodb_source(mongodb_function, dest):
362362

363363
dest_uri = dest.start()
364364

365+
if dest_uri.startswith("cratedb://"):
366+
pytest.skip(
367+
'Fails on CrateDB with `"_id" conflicts with system column`, '
368+
"see https://github.com/crate/dlt-cratedb/issues/19"
369+
)
370+
365371
try:
366372
invoke_ingest_command(
367373
mongo.get_connection_url(),
@@ -775,5 +781,11 @@ def incremental_multiple_days(mongo, dest_uri: str):
775781
)
776782
def test_mongodb_custom_query(testcase, mongodb_function, dest):
777783
"""Test MongoDB custom aggregation queries"""
778-
testcase(mongodb_function, dest.start())
784+
dest_uri = dest.start()
785+
if dest_uri.startswith("cratedb://"):
786+
pytest.skip(
787+
'Fails on CrateDB with `"_id" conflicts with system column`, '
788+
"see https://github.com/crate/dlt-cratedb/issues/19"
789+
)
790+
testcase(mongodb_function, dest_uri)
779791
dest.stop()

0 commit comments

Comments
 (0)