Skip to content

Commit 6d01c5f

Browse files
arbivarbiv
committed
tests
1 parent 77702e8 commit 6d01c5f

File tree

2 files changed

+216
-0
lines changed

2 files changed

+216
-0
lines changed

‎integration_tests/tests/dbt_project.py‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -277,3 +277,17 @@ def write_yaml(self, content: dict, name: Optional[str] = None):
277277
YAML().dump(content, f)
278278
yield path
279279
path.unlink()
280+
281+
def assert_table_exists(self, table_name: str):
282+
table_exists_query = f"""
283+
select 1
284+
from information_schema.tables
285+
where table_name = '{table_name}'
286+
"""
287+
try:
288+
table_exists_result = self.run_query(table_exists_query)
289+
except IndexError:
290+
table_exists_result = []
291+
except Exception as e:
292+
raise AssertionError(f"Failed to check if {table_name} exists: {e}")
293+
assert table_exists_result, f"{table_name} table does not exist. The artifact was not created."

‎integration_tests/tests/test_dbt_artifacts/test_groups.py‎

Lines changed: 202 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,202 @@
1+
from dbt_project import DbtProject
2+
3+
GROUP_NAME = "test_group"
4+
OWNER_NAME = "Data Spock"
5+
OWNER_EMAIL = "test@example.com"
6+
7+
8+
def test_dbt_group_artifact(dbt_project: DbtProject):
9+
# Define a group configuration using constants
10+
group_config = {
11+
"groups": [
12+
{
13+
"name": GROUP_NAME,
14+
"owner": {
15+
"name": OWNER_NAME,
16+
"email": OWNER_EMAIL,
17+
}
18+
}
19+
]
20+
}
21+
22+
# Write the group config to a YAML file in the dbt project
23+
with dbt_project.write_yaml(group_config, name="groups_test.yml"):
24+
# Run dbt to generate artifacts
25+
dbt_project.dbt_runner.vars["disable_dbt_artifacts_autoupload"] = False
26+
dbt_project.dbt_runner.run()
27+
28+
# Query the dbt_groups artifact table
29+
groups = dbt_project.read_table("dbt_groups", raise_if_empty=True)
30+
group = next((g for g in groups if g["name"] == GROUP_NAME), None)
31+
32+
# Assert the expected group and owner details are present
33+
assert group is not None, f"Group {GROUP_NAME} not found in dbt_groups artifact table."
34+
assert group.get("owner_name") == OWNER_NAME, f"Expected owner name {OWNER_NAME}, got {group.get('owner_name')}"
35+
assert group.get("owner_email") == OWNER_EMAIL, f"Expected owner email {OWNER_EMAIL}, got {group.get('owner_email')}"
36+
37+
38+
def test_dbt_groups_artifact_with_two_groups(dbt_project: DbtProject):
39+
# Define two group configurations: one with owner, one without
40+
group1 = {
41+
"name": "test_group_1",
42+
"owner": {
43+
"name": "Owner One",
44+
"email": "owner1@example.com",
45+
}
46+
}
47+
group2 = {
48+
"name": "test_group_2",
49+
"owner": {
50+
"name": "Owner Two",
51+
"email": "owner2@example.com",
52+
}
53+
}
54+
group_config = {"groups": [group1, group2]}
55+
56+
# Write the group config to a YAML file in the dbt project
57+
with dbt_project.write_yaml(group_config, name="groups_test_two_groups.yml"):
58+
# Run dbt to generate artifacts
59+
dbt_project.dbt_runner.vars["disable_dbt_artifacts_autoupload"] = False
60+
dbt_project.dbt_runner.run()
61+
62+
# Query the dbt_groups artifact table
63+
groups = dbt_project.read_table("dbt_groups", raise_if_empty=True)
64+
group1_result = next((g for g in groups if g["name"] == "test_group_1"), None)
65+
group2_result = next((g for g in groups if g["name"] == "test_group_2"), None)
66+
67+
# Assert group 1 (with owner)
68+
assert group1_result is not None, "Group test_group_1 not found in dbt_groups artifact table."
69+
assert group1_result.get("owner_name") == "Owner One", f"Expected owner name Owner One, got {group1_result.get('owner_name')}"
70+
assert group1_result.get("owner_email") == "owner1@example.com", f"Expected owner email owner1@example.com, got {group1_result.get('owner_email')}"
71+
72+
# Assert group 2 (without owner)
73+
assert group2_result is not None, "Group test_group_2 not found in dbt_groups artifact table."
74+
assert group2_result.get("owner_name") == "Owner Two", f"Expected owner name Owner Two, got {group2_result.get('owner_name')}"
75+
assert group2_result.get("owner_email") == "owner2@example.com", f"Expected owner email owner2@example.com, got {group2_result.get('owner_email')}"
76+
77+
78+
def test_model_group_attribute(dbt_project: DbtProject):
79+
model_name = "model_with_group"
80+
group_name = GROUP_NAME
81+
owner_name = OWNER_NAME
82+
owner_email = OWNER_EMAIL
83+
84+
group_config = {
85+
"groups": [
86+
{
87+
"name": group_name,
88+
"owner": {
89+
"name": owner_name,
90+
"email": owner_email,
91+
},
92+
}
93+
]
94+
}
95+
96+
model_sql = """
97+
select 1 as col
98+
"""
99+
100+
schema_yaml = {
101+
"version": 2,
102+
"models": [
103+
{
104+
"name": model_name,
105+
"group": group_name,
106+
"description": "A model assigned to a group for testing",
107+
}
108+
]
109+
}
110+
111+
with dbt_project.write_yaml(group_config, name="groups_test_model_inherits.yml"), \
112+
dbt_project.write_yaml(schema_yaml, name="schema_model_with_group.yml"):
113+
model_path = dbt_project.models_dir_path / "tmp" / f"{model_name}.sql"
114+
model_path.parent.mkdir(parents=True, exist_ok=True)
115+
model_path.write_text(model_sql)
116+
try:
117+
# Run dbt
118+
dbt_project.dbt_runner.vars["disable_dbt_artifacts_autoupload"] = False
119+
dbt_project.dbt_runner.run(select=model_name)
120+
121+
models = dbt_project.read_table("dbt_models", where=f"name = '{model_name}'", raise_if_empty=True)
122+
assert len(models) == 1, f"Expected 1 model, got {len(models)}"
123+
model_row = models[0]
124+
assert model_row["group_name"] == group_name, f"Expected group_name {group_name}, got {model_row['group_name']}"
125+
finally:
126+
if model_path.exists():
127+
model_path.unlink()
128+
129+
130+
def test_test_group_attribute(dbt_project: DbtProject):
131+
model_name = "model_with_group"
132+
group_name = GROUP_NAME
133+
owner_name = OWNER_NAME
134+
owner_email = OWNER_EMAIL
135+
test_args = {"where": "col = 1"}
136+
137+
schema_yaml = {
138+
"version": 2,
139+
"models": [
140+
{
141+
"name": model_name,
142+
"group": group_name,
143+
"description": "A model assigned to a group for testing",
144+
"tests": [
145+
{"generic": test_args}
146+
]
147+
}
148+
]
149+
}
150+
151+
group_config = {
152+
"groups": [
153+
{
154+
"name": group_name,
155+
"owner": {
156+
"name": owner_name,
157+
"email": owner_email,
158+
},
159+
}
160+
]
161+
}
162+
163+
model_sql = """
164+
select 1 as col
165+
"""
166+
167+
schema_yaml = {
168+
"version": 2,
169+
"models": [
170+
{
171+
"name": model_name,
172+
"group": group_name,
173+
"description": "A model assigned to a group for testing",
174+
"columns": [
175+
{
176+
"name": "col",
177+
"tests": ["unique"]
178+
}
179+
]
180+
}
181+
]
182+
}
183+
184+
with dbt_project.write_yaml(group_config, name="groups_test_model_inherits.yml"), \
185+
dbt_project.write_yaml(schema_yaml, name="schema_model_with_group.yml"):
186+
model_path = dbt_project.models_dir_path / "tmp" / f"{model_name}.sql"
187+
model_path.parent.mkdir(parents=True, exist_ok=True)
188+
model_path.write_text(model_sql)
189+
try:
190+
# Run dbt
191+
dbt_project.dbt_runner.vars["disable_dbt_artifacts_autoupload"] = False
192+
dbt_project.dbt_runner.run(select=model_name)
193+
194+
tests = dbt_project.read_table("dbt_tests", where=f"name LIKE 'unique%'", raise_if_empty=True)
195+
assert len(tests) == 1, f"Expected 1 test, got {len(tests)}"
196+
test_row = tests[0]
197+
assert test_row["group_name"] == group_name, f"Expected group_name {group_name}, got {test_row['group_name']}"
198+
finally:
199+
if model_path.exists():
200+
model_path.unlink()
201+
202+

0 commit comments

Comments
 (0)