Skip to content

Commit 8ff68b6

Browse files
fix: use robust query for relationship tests across dbt versions
Co-Authored-By: Yosef Arbiv <[email protected]>
1 parent 9f1685a commit 8ff68b6

File tree

1 file changed

+36
-10
lines changed

1 file changed

+36
-10
lines changed

integration_tests/tests/test_dbt_artifacts/test_test_owners.py

Lines changed: 36 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -175,14 +175,27 @@ def test_relationship_test_uses_primary_model_owner_only(
175175
select=f"{primary_model_name} {referenced_model_name}"
176176
)
177177

178-
tests = dbt_project.read_table(
178+
# Query by parent_model_unique_id and filter by test_original_name in Python
179+
# This is more robust across dbt versions (fusion vs latest_official)
180+
all_tests = dbt_project.read_table(
179181
"dbt_tests",
180-
where=f"name LIKE '%relationships%' AND name LIKE '%{primary_model_name}%'",
181-
raise_if_empty=True,
182+
where=f"parent_model_unique_id LIKE '%{primary_model_name}%'",
183+
raise_if_empty=False,
182184
)
183185

184-
assert len(tests) == 1, f"Expected 1 relationship test, got {len(tests)}"
185-
test_row = tests[0]
186+
# Filter for relationship tests
187+
relationship_tests = [
188+
t
189+
for t in all_tests
190+
if t.get("test_original_name") == "relationships"
191+
or "relationships" in (t.get("short_name") or "").lower()
192+
or "relationships" in (t.get("name") or "").lower()
193+
]
194+
195+
assert (
196+
len(relationship_tests) == 1
197+
), f"Expected 1 relationship test, got {len(relationship_tests)}. All tests found: {[t.get('name') for t in all_tests]}"
198+
test_row = relationship_tests[0]
186199
model_owners = _parse_model_owners(test_row.get("model_owners"))
187200

188201
assert model_owners == [
@@ -261,14 +274,27 @@ def test_relationship_test_no_owner_on_primary_model(dbt_project: DbtProject, tm
261274
select=f"{primary_model_name} {referenced_model_name}"
262275
)
263276

264-
tests = dbt_project.read_table(
277+
# Query by parent_model_unique_id and filter by test_original_name in Python
278+
# This is more robust across dbt versions (fusion vs latest_official)
279+
all_tests = dbt_project.read_table(
265280
"dbt_tests",
266-
where=f"name LIKE '%relationships%' AND name LIKE '%{primary_model_name}%'",
267-
raise_if_empty=True,
281+
where=f"parent_model_unique_id LIKE '%{primary_model_name}%'",
282+
raise_if_empty=False,
268283
)
269284

270-
assert len(tests) == 1, f"Expected 1 relationship test, got {len(tests)}"
271-
test_row = tests[0]
285+
# Filter for relationship tests
286+
relationship_tests = [
287+
t
288+
for t in all_tests
289+
if t.get("test_original_name") == "relationships"
290+
or "relationships" in (t.get("short_name") or "").lower()
291+
or "relationships" in (t.get("name") or "").lower()
292+
]
293+
294+
assert (
295+
len(relationship_tests) == 1
296+
), f"Expected 1 relationship test, got {len(relationship_tests)}. All tests found: {[t.get('name') for t in all_tests]}"
297+
test_row = relationship_tests[0]
272298
model_owners = _parse_model_owners(test_row.get("model_owners"))
273299

274300
assert (

0 commit comments

Comments
 (0)