repo-mirrors/dbt-core
1
0
Fork 1
mirror of https://github.com/dbt-labs/dbt-core synced 2025-12-17 19:31:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

make graph_selection tests just checking selection (#5012)

Browse Source 
* make graph_selection tests just checking selection

* use util function
This commit is contained in:
Chenyu Li
2022-04-08 11:04:54 -06:00
committed by GitHub
parent 40e64b238c
commit 6c277b5fe1
4 changed files with 249 additions and 389 deletions
Download Patch File Download Diff File Expand all files Collapse all files

275
tests/functional/graph_selection/test_graph_selection.py
View File

@@ -2,7 +2,7 @@ import os
import json
import pytest
from dbt.tests.util import run_dbt, check_relations_equal
from dbt.tests.util import run_dbt, check_result_nodes_by_name
from tests.functional.graph_selection.fixtures import SelectionFixtures
@@ -32,41 +32,21 @@ def clear_schema(project):
project.run_sql("create schema {schema}")
@pytest.fixture
def run_seed(project):
clear_schema(project)
run_dbt(["seed"])
@pytest.mark.usefixtures("project", "run_seed")
class TestGraphSelection(SelectionFixtures):
# The tests here aiming to test whether the correct node is selected,
# we don't need the run to pass
@pytest.fixture(scope="class")
def selectors(self):
return selectors_yml
def test_specific_model(self, project):
run_dbt(["seed"])
results = run_dbt(["run", "--select", "users"])
assert len(results) == 1
check_relations_equal(project.adapter, ["seed", "users"])
created_tables = project.get_tables_in_schema()
assert "users_rollup" not in created_tables
assert "alternative.users" not in created_tables
assert "base_users" not in created_tables
assert "emails" not in created_tables
results = run_dbt(["run", "--select", "users"], expect_pass=False)
check_result_nodes_by_name(results, ["users"])
assert_correct_schemas(project)
def test_tags(self, project, project_root):
results = run_dbt(["run", "--selector", "bi_selector"])
assert len(results) == 2
created_tables = project.get_tables_in_schema()
assert not ("alternative.users" in created_tables)
assert not ("base_users" in created_tables)
assert not ("emails" in created_tables)
assert "users" in created_tables
assert "users_rollup" in created_tables
results = run_dbt(["run", "--selector", "bi_selector"], expect_pass=False)
check_result_nodes_by_name(results, ["users", "users_rollup"])
assert_correct_schemas(project)
manifest_path = project_root.join("target/manifest.json")
assert os.path.exists(manifest_path)
@@ -75,178 +55,122 @@ class TestGraphSelection(SelectionFixtures):
assert "selectors" in manifest
def test_tags_and_children(self, project):
results = run_dbt(["run", "--select", "tag:base+"])
assert len(results) == 5
created_models = project.get_tables_in_schema()
assert not ("base_users" in created_models)
assert not ("emails" in created_models)
assert "emails_alt" in created_models
assert "users_rollup" in created_models
assert "users" in created_models
assert "alternative.users" in created_models
results = run_dbt(["run", "--select", "tag:base+"], expect_pass=False)
check_result_nodes_by_name(
results,
[
"emails_alt",
"users_rollup",
"users",
"alternative.users",
"users_rollup_dependency",
],
)
assert_correct_schemas(project)
def test_tags_and_children_limited(self, project):
results = run_dbt(["run", "--select", "tag:base+2"])
assert len(results) == 4
created_models = project.get_tables_in_schema()
assert not ("base_users" in created_models)
assert not ("emails" in created_models)
assert "emails_alt" in created_models
assert "users_rollup" in created_models
assert "users" in created_models
assert "alternative.users" in created_models
assert "users_rollup_dependency" not in created_models
results = run_dbt(["run", "--select", "tag:base+2"], expect_pass=False)
check_result_nodes_by_name(
results, ["emails_alt", "users_rollup", "users", "alternative.users"]
)
assert_correct_schemas(project)
def test_specific_model_and_children(self, project):
results = run_dbt(["run", "--select", "users+"])
assert len(results) == 4
check_relations_equal(project.adapter, ["seed", "users"])
check_relations_equal(project.adapter, ["summary_expected", "users_rollup"])
created_models = project.get_tables_in_schema()
assert "emails_alt" in created_models
assert "base_users" not in created_models
assert "alternative.users" not in created_models
assert "emails" not in created_models
results = run_dbt(["run", "--select", "users+"], expect_pass=False)
check_result_nodes_by_name(
results, ["users", "users_rollup", "emails_alt", "users_rollup_dependency"]
)
assert_correct_schemas(project)
def test_specific_model_and_children_limited(self, project):
results = run_dbt(["run", "--select", "users+1"])
assert len(results) == 3
check_relations_equal(project.adapter, ["seed", "users"])
check_relations_equal(project.adapter, ["summary_expected", "users_rollup"])
created_models = project.get_tables_in_schema()
assert "emails_alt" in created_models
assert "base_users" not in created_models
assert "emails" not in created_models
assert "users_rollup_dependency" not in created_models
results = run_dbt(["run", "--select", "users+1"], expect_pass=False)
check_result_nodes_by_name(results, ["users", "users_rollup", "emails_alt"])
assert_correct_schemas(project)
def test_specific_model_and_parents(self, project):
results = run_dbt(["run", "--select", "+users_rollup"])
assert len(results) == 2
check_relations_equal(project.adapter, ["seed", "users"])
check_relations_equal(project.adapter, ["summary_expected", "users_rollup"])
created_models = project.get_tables_in_schema()
assert not ("base_users" in created_models)
assert not ("emails" in created_models)
results = run_dbt(["run", "--select", "+users_rollup"], expect_pass=False)
check_result_nodes_by_name(results, ["users_rollup", "users"])
assert_correct_schemas(project)
def test_specific_model_and_parents_limited(self, project):
results = run_dbt(["run", "--select", "1+users_rollup"])
assert len(results) == 2
check_relations_equal(project.adapter, ["seed", "users"])
check_relations_equal(project.adapter, ["summary_expected", "users_rollup"])
created_models = project.get_tables_in_schema()
assert not ("base_users" in created_models)
assert not ("emails" in created_models)
results = run_dbt(["run", "--select", "1+users_rollup"], expect_pass=False)
check_result_nodes_by_name(results, ["users", "users_rollup"])
assert_correct_schemas(project)
def test_specific_model_with_exclusion(self, project):
results = run_dbt(
["run", "--select", "+users_rollup", "--exclude", "models/users_rollup.sql"]
[
"run",
"--select",
"+users_rollup",
"--exclude",
"models/users_rollup.sql",
],
expect_pass=False,
)
assert len(results) == 1
check_relations_equal(project.adapter, ["seed", "users"])
created_models = project.get_tables_in_schema()
assert not ("base_users" in created_models)
assert not ("users_rollup" in created_models)
assert not ("emails" in created_models)
check_result_nodes_by_name(results, ["users"])
assert_correct_schemas(project)
def test_locally_qualified_name(self, project):
results = run_dbt(["run", "--select", "test.subdir"])
assert len(results) == 2
created_models = project.get_tables_in_schema()
assert "users_rollup" not in created_models
assert "base_users" not in created_models
assert "emails" not in created_models
assert "subdir" in created_models
assert "nested_users" in created_models
check_result_nodes_by_name(results, ["nested_users", "subdir"])
assert_correct_schemas(project)
results = run_dbt(["run", "--select", "models/test/subdir*"])
assert len(results) == 2
created_models = project.get_tables_in_schema()
assert "users_rollup" not in created_models
assert "base_users" not in created_models
assert "emails" not in created_models
assert "subdir" in created_models
assert "nested_users" in created_models
check_result_nodes_by_name(results, ["nested_users", "subdir"])
assert_correct_schemas(project)
def test_locally_qualified_name_model_with_dots(self, project):
results = run_dbt(["run", "--select", "alternative.users"])
assert len(results) == 1
created_models = project.get_tables_in_schema()
assert "alternative.users" in created_models
results = run_dbt(["run", "--select", "alternative.users"], expect_pass=False)
check_result_nodes_by_name(results, ["alternative.users"])
assert_correct_schemas(project)
results = run_dbt(["run", "--select", "models/alternative.*"])
assert len(results) == 1
created_models = project.get_tables_in_schema()
assert "alternative.users" in created_models
results = run_dbt(["run", "--select", "models/alternative.*"], expect_pass=False)
check_result_nodes_by_name(results, ["alternative.users"])
assert_correct_schemas(project)
def test_childrens_parents(self, project):
results = run_dbt(["run", "--select", "@base_users"])
assert len(results) == 5
created_models = project.get_tables_in_schema()
assert "users_rollup" in created_models
assert "users" in created_models
assert "emails_alt" in created_models
assert "alternative.users" in created_models
assert "subdir" not in created_models
assert "nested_users" not in created_models
results = run_dbt(["run", "--select", "@base_users"], expect_pass=False)
check_result_nodes_by_name(
results,
[
"alternative.users",
"users_rollup",
"users",
"emails_alt",
"users_rollup_dependency",
],
)
results = run_dbt(["test", "--select", "test_name:not_null"])
assert len(results) == 1
assert results[0].node.name == "not_null_emails_email"
results = run_dbt(["test", "--select", "test_name:not_null"], expect_pass=False)
check_result_nodes_by_name(results, ["not_null_emails_email"])
def test_more_childrens_parents(self, project):
results = run_dbt(["run", "--select", "@users"])
assert len(results) == 4
created_models = project.get_tables_in_schema()
assert "users_rollup" in created_models
assert "users" in created_models
assert "emails_alt" in created_models
assert "subdir" not in created_models
assert "nested_users" not in created_models
results = run_dbt(["test", "--select", "test_name:unique"])
assert len(results) == 2
assert sorted([r.node.name for r in results]) == [
"unique_users_id",
"unique_users_rollup_gender",
]
results = run_dbt(["run", "--select", "@users"], expect_pass=False)
check_result_nodes_by_name(
results, ["users_rollup", "users", "emails_alt", "users_rollup_dependency"]
)
results = run_dbt(["test", "--select", "test_name:unique"], expect_pass=False)
check_result_nodes_by_name(results, ["unique_users_id", "unique_users_rollup_gender"])
def test_concat(self, project):
results = run_dbt(["run", "--select", "@emails_alt", "users_rollup"])
assert len(results) == 3
created_models = project.get_tables_in_schema()
assert "users_rollup" in created_models
assert "users" in created_models
assert "emails_alt" in created_models
assert "subdir" not in created_models
assert "nested_users" not in created_models
results = run_dbt(["run", "--select", "@emails_alt", "users_rollup"], expect_pass=False)
check_result_nodes_by_name(results, ["users_rollup", "users", "emails_alt"])
def test_concat_exclude(self, project):
results = run_dbt(
["run", "--select", "@emails_alt", "users_rollup", "--exclude", "emails_alt"]
[
"run",
"--select",
"@emails_alt",
"users_rollup",
"--exclude",
"emails_alt",
],
expect_pass=False,
)
assert len(results) == 2
created_models = project.get_tables_in_schema()
assert "users" in created_models
assert "users_rollup" in created_models
assert "emails_alt" not in created_models
assert "subdir" not in created_models
assert "nested_users" not in created_models
check_result_nodes_by_name(results, ["users_rollup", "users"])
def test_concat_exclude_concat(self, project):
results = run_dbt(
@@ -258,15 +182,11 @@ class TestGraphSelection(SelectionFixtures):
"--exclude",
"emails_alt",
"users_rollup",
]
],
expect_pass=False,
)
assert len(results) == 1
created_models = project.get_tables_in_schema()
assert "users" in created_models
assert "emails_alt" not in created_models
assert "users_rollup" not in created_models
assert "subdir" not in created_models
assert "nested_users" not in created_models
check_result_nodes_by_name(results, ["users"])
results = run_dbt(
[
"test",
@@ -276,17 +196,18 @@ class TestGraphSelection(SelectionFixtures):
"--exclude",
"emails_alt",
"users_rollup",
]
],
expect_pass=False,
)
assert len(results) == 1
assert results[0].node.name == "unique_users_id"
check_result_nodes_by_name(results, ["unique_users_id"])
def test_exposure_parents(self, project):
results = run_dbt(["ls", "--select", "+exposure:seed_ml_exposure"])
assert len(results) == 2
assert sorted(results) == ["exposure:test.seed_ml_exposure", "source:test.raw.seed"]
assert sorted(results) == [
"exposure:test.seed_ml_exposure",
"source:test.raw.seed",
]
results = run_dbt(["ls", "--select", "1+exposure:user_exposure"])
assert len(results) == 5
assert sorted(results) == [
"exposure:test.user_exposure",
"test.unique_users_id",
@@ -294,11 +215,11 @@ class TestGraphSelection(SelectionFixtures):
"test.users",
"test.users_rollup",
]
results = run_dbt(["run", "-m", "+exposure:user_exposure"])
assert len(results) == 2
created_models = project.get_tables_in_schema()
assert "users_rollup" in created_models
assert "users" in created_models
assert "emails_alt" not in created_models
assert "subdir" not in created_models
assert "nested_users" not in created_models
results = run_dbt(["run", "-m", "+exposure:user_exposure"], expect_pass=False)
check_result_nodes_by_name(
results,
[
"users_rollup",
"users",
],
)

189
tests/functional/graph_selection/test_intersection_syntax.py
View File

@@ -1,6 +1,6 @@
import pytest
from dbt.tests.util import run_dbt
from dbt.tests.util import run_dbt, check_result_nodes_by_name
from tests.functional.graph_selection.fixtures import SelectionFixtures
@@ -66,87 +66,57 @@ selectors_yml = """
"""
def verify_selected_users(project, results):
# users
assert len(results) == 1
created_models = project.get_tables_in_schema()
assert "users" in created_models
assert "users_rollup" not in created_models
assert "emails_alt" not in created_models
assert "subdir" not in created_models
assert "nested_users" not in created_models
def verify_selected_users_and_rollup(project, results):
# users, users_rollup
assert len(results) == 2
created_models = project.get_tables_in_schema()
assert "users" in created_models
assert "users_rollup" in created_models
assert "emails_alt" not in created_models
assert "subdir" not in created_models
assert "nested_users" not in created_models
@pytest.fixture
def run_seed(project):
run_dbt(["seed"])
# The project and run_seed fixtures will be executed for each test method
@pytest.mark.usefixtures("project", "run_seed")
class TestIntersectionSyncs(SelectionFixtures):
# The tests here aiming to test whether the correct node is selected,
# we don't need the run to pass
@pytest.fixture(scope="class")
def selectors(self):
return selectors_yml
def test_same_model_intersection(self, project):
run_dbt(["seed"])
results = run_dbt(["run", "--models", "users,users"])
verify_selected_users(project, results)
results = run_dbt(["run", "--models", "users,users"], expect_pass=False)
check_result_nodes_by_name(results, ["users"])
def test_same_model_intersection_selectors(self, project):
results = run_dbt(["run", "--selector", "same_intersection"])
verify_selected_users(project, results)
results = run_dbt(["run", "--selector", "same_intersection"], expect_pass=False)
check_result_nodes_by_name(results, ["users"])
def test_tags_intersection(self, project):
results = run_dbt(["run", "--models", "tag:bi,tag:users"])
verify_selected_users(project, results)
results = run_dbt(["run", "--models", "tag:bi,tag:users"], expect_pass=False)
check_result_nodes_by_name(results, ["users"])
def test_tags_intersection_selectors(self, project):
results = run_dbt(["run", "--selector", "tags_intersection"])
verify_selected_users(project, results)
results = run_dbt(["run", "--selector", "tags_intersection"], expect_pass=False)
check_result_nodes_by_name(results, ["users"])
def test_intersection_triple_descending(self, project):
results = run_dbt(["run", "--models", "*,tag:bi,tag:users"])
verify_selected_users(project, results)
results = run_dbt(["run", "--models", "*,tag:bi,tag:users"], expect_pass=False)
check_result_nodes_by_name(results, ["users"])
def test_intersection_triple_descending_schema(self, project):
results = run_dbt(["run", "--models", "*,tag:bi,tag:users"])
verify_selected_users(project, results)
results = run_dbt(["run", "--models", "*,tag:bi,tag:users"], expect_pass=False)
check_result_nodes_by_name(results, ["users"])
def test_intersection_triple_descending_schema_selectors(self, project):
results = run_dbt(["run", "--selector", "triple_descending"])
verify_selected_users(project, results)
results = run_dbt(["run", "--selector", "triple_descending"], expect_pass=False)
check_result_nodes_by_name(results, ["users"])
def test_intersection_triple_ascending(self, project):
results = run_dbt(["run", "--models", "tag:users,tag:bi,*"])
verify_selected_users(project, results)
results = run_dbt(["run", "--models", "tag:users,tag:bi,*"], expect_pass=False)
check_result_nodes_by_name(results, ["users"])
def test_intersection_triple_ascending_schema_selectors(self, project):
results = run_dbt(["run", "--selector", "triple_ascending"])
verify_selected_users(project, results)
results = run_dbt(["run", "--selector", "triple_ascending"], expect_pass=False)
check_result_nodes_by_name(results, ["users"])
def test_intersection_with_exclusion(self, project):
@@ -157,90 +127,86 @@ class TestIntersectionSyncs(SelectionFixtures):
"+users_rollup_dependency,users+",
"--exclude",
"users_rollup_dependency",
]
],
expect_pass=False,
)
verify_selected_users_and_rollup(project, results)
check_result_nodes_by_name(results, ["users", "users_rollup"])
def test_intersection_with_exclusion_selectors(self, project):
results = run_dbt(["run", "--selector", "intersection_with_exclusion"])
verify_selected_users_and_rollup(project, results)
results = run_dbt(["run", "--selector", "intersection_with_exclusion"], expect_pass=False)
check_result_nodes_by_name(results, ["users", "users_rollup"])
def test_intersection_exclude_intersection(self, project):
results = run_dbt(
["run", "--models", "tag:bi,@users", "--exclude", "tag:bi,users_rollup+"]
["run", "--models", "tag:bi,@users", "--exclude", "tag:bi,users_rollup+"],
expect_pass=False,
)
verify_selected_users(project, results)
check_result_nodes_by_name(results, ["users"])
def test_intersection_exclude_intersection_selectors(self, project):
results = run_dbt(["run", "--selector", "intersection_exclude_intersection"])
verify_selected_users(project, results)
results = run_dbt(
["run", "--selector", "intersection_exclude_intersection"],
expect_pass=False,
)
check_result_nodes_by_name(results, ["users"])
def test_intersection_exclude_intersection_lack(self, project):
results = run_dbt(["run", "--models", "tag:bi,@users", "--exclude", "@emails,@emails_alt"])
verify_selected_users_and_rollup(project, results)
results = run_dbt(
["run", "--models", "tag:bi,@users", "--exclude", "@emails,@emails_alt"],
expect_pass=False,
)
check_result_nodes_by_name(results, ["users", "users_rollup"])
def test_intersection_exclude_intersection_lack_selector(self, project):
results = run_dbt(["run", "--selector", "intersection_exclude_intersection_lack"])
verify_selected_users_and_rollup(project, results)
results = run_dbt(
["run", "--selector", "intersection_exclude_intersection_lack"],
expect_pass=False,
)
check_result_nodes_by_name(results, ["users", "users_rollup"])
def test_intersection_exclude_triple_intersection(self, project):
results = run_dbt(
["run", "--models", "tag:bi,@users", "--exclude", "*,tag:bi,users_rollup"]
["run", "--models", "tag:bi,@users", "--exclude", "*,tag:bi,users_rollup"],
expect_pass=False,
)
verify_selected_users(project, results)
check_result_nodes_by_name(results, ["users"])
def test_intersection_concat(self, project):
results = run_dbt(["run", "--models", "tag:bi,@users", "emails_alt"])
# users, users_rollup, emails_alt
assert len(results) == 3
created_models = project.get_tables_in_schema()
assert "users" in created_models
assert "users_rollup" in created_models
assert "emails_alt" in created_models
assert "subdir" not in created_models
assert "nested_users" not in created_models
results = run_dbt(["run", "--models", "tag:bi,@users", "emails_alt"], expect_pass=False)
check_result_nodes_by_name(results, ["users", "users_rollup", "emails_alt"])
def test_intersection_concat_intersection(self, project):
run_dbt(["seed"])
results = run_dbt(["run", "--models", "tag:bi,@users", "@emails_alt,emails_alt"])
# users, users_rollup, emails_alt
assert len(results) == 3
created_models = project.get_tables_in_schema()
assert "users" in created_models
assert "users_rollup" in created_models
assert "emails_alt" in created_models
assert "subdir" not in created_models
assert "nested_users" not in created_models
results = run_dbt(
["run", "--models", "tag:bi,@users", "@emails_alt,emails_alt"],
expect_pass=False,
)
check_result_nodes_by_name(results, ["users", "users_rollup", "emails_alt"])
def test_intersection_concat_exclude(self, project):
run_dbt(["seed"])
results = run_dbt(
["run", "--models", "tag:bi,@users", "emails_alt", "--exclude", "users_rollup"]
[
"run",
"--models",
"tag:bi,@users",
"emails_alt",
"--exclude",
"users_rollup",
],
expect_pass=False,
)
# users, emails_alt
assert len(results) == 2
created_models = project.get_tables_in_schema()
assert "users" in created_models
assert "emails_alt" in created_models
assert "users_rollup" not in created_models
assert "subdir" not in created_models
assert "nested_users" not in created_models
check_result_nodes_by_name(results, ["users", "emails_alt"])
def test_intersection_concat_exclude_concat(self, project):
run_dbt(["seed"])
results = run_dbt(
[
"run",
@@ -250,21 +216,13 @@ class TestIntersectionSyncs(SelectionFixtures):
"--exclude",
"users_rollup_dependency",
"users_rollup",
]
],
expect_pass=False,
)
# users, emails_alt
assert len(results) == 2
created_models = project.get_tables_in_schema()
assert "users" in created_models
assert "emails_alt" in created_models
assert "users_rollup" not in created_models
assert "subdir" not in created_models
assert "nested_users" not in created_models
check_result_nodes_by_name(results, ["users", "emails_alt"])
def test_intersection_concat_exclude_intersection_concat(self, project):
run_dbt(["seed"])
results = run_dbt(
[
"run",
@@ -274,14 +232,7 @@ class TestIntersectionSyncs(SelectionFixtures):
"--exclude",
"@users,users_rollup_dependency",
"@users,users_rollup",
]
],
expect_pass=False,
)
# users, emails_alt
assert len(results) == 2
created_models = project.get_tables_in_schema()
assert "users" in created_models
assert "emails_alt" in created_models
assert "users_rollup" not in created_models
assert "subdir" not in created_models
assert "nested_users" not in created_models
check_result_nodes_by_name(results, ["users", "emails_alt"])

16
tests/functional/graph_selection/test_schema_test_graph_selection.py
View File

@@ -73,7 +73,10 @@ class TestSchemaTestGraphSelection(SelectionFixtures):
def test_schema_tests_specify_model_and_parents(self, project):
run_schema_and_assert(
project, "+users_rollup", None, ["unique_users_id", "unique_users_rollup_gender"]
project,
"+users_rollup",
None,
["unique_users_id", "unique_users_rollup_gender"],
)
def test_schema_tests_specify_model_and_parents_with_exclude(self, project):
@@ -102,7 +105,11 @@ class TestSchemaTestGraphSelection(SelectionFixtures):
project,
"*",
"users",
["not_null_emails_email", "unique_table_model_id", "unique_users_rollup_gender"],
[
"not_null_emails_email",
"unique_table_model_id",
"unique_users_rollup_gender",
],
)
def test_schema_tests_dep_package_only(self, project):
@@ -110,7 +117,10 @@ class TestSchemaTestGraphSelection(SelectionFixtures):
def test_schema_tests_model_in_dep_pkg(self, project):
run_schema_and_assert(
project, "dbt_integration_project.table_model", None, ["unique_table_model_id"]
project,
"dbt_integration_project.table_model",
None,
["unique_table_model_id"],
)
def test_schema_tests_exclude_pkg(self, project):

158
tests/functional/graph_selection/test_tag_selection.py
View File

@@ -1,6 +1,6 @@
import pytest
from dbt.tests.util import run_dbt
from dbt.tests.util import run_dbt, check_result_nodes_by_name
from tests.functional.graph_selection.fixtures import SelectionFixtures
@@ -48,31 +48,9 @@ selectors_yml = """
"""
def _verify_select_tag(results):
assert len(results) == 1
models_run = [r.node.name for r in results]
assert "users" in models_run
def _verify_select_tag_and_children(results):
assert len(results) == 3
models_run = [r.node.name for r in results]
assert "users" in models_run
assert "users_rollup" in models_run
# check that model configs aren't squashed by project configs
def _verify_select_bi(results):
assert len(results) == 2
models_run = [r.node.name for r in results]
assert "users" in models_run
assert "users_rollup" in models_run
class TestTagSelection(SelectionFixtures):
# The tests here aiming to test whether the correct node is selected,
# we don't need the run to pass
@pytest.fixture(scope="class")
def project_config_update(self):
return {
@@ -92,100 +70,100 @@ class TestTagSelection(SelectionFixtures):
return selectors_yml
def test_select_tag(self, project):
run_dbt(["seed"])
results = run_dbt(["run", "--models", "tag:specified_as_string"])
_verify_select_tag(results)
results = run_dbt(["run", "--models", "tag:specified_as_string"], expect_pass=False)
check_result_nodes_by_name(results, ["users"])
def test_select_tag_selector_str(self, project):
results = run_dbt(["run", "--selector", "tag_specified_as_string_str"])
_verify_select_tag(results)
results = run_dbt(["run", "--selector", "tag_specified_as_string_str"], expect_pass=False)
check_result_nodes_by_name(results, ["users"])
def test_select_tag_selector_dict(self, project):
results = run_dbt(["run", "--selector", "tag_specified_as_string_dict"])
_verify_select_tag(results)
results = run_dbt(["run", "--selector", "tag_specified_as_string_dict"], expect_pass=False)
check_result_nodes_by_name(results, ["users"])
def test_select_tag_and_children(self, project): # noqa
results = run_dbt(["run", "--models", "+tag:specified_in_project+"])
_verify_select_tag_and_children(results)
results = run_dbt(["run", "--models", "+tag:specified_in_project+"], expect_pass=False)
check_result_nodes_by_name(results, ["users", "users_rollup", "users_rollup_dependency"])
def test_select_tag_and_children_selector_str(self, project): # noqa
results = run_dbt(["run", "--selector", "tag_specified_in_project_children_str"])
_verify_select_tag_and_children(results)
results = run_dbt(
["run", "--selector", "tag_specified_in_project_children_str"],
expect_pass=False,
)
check_result_nodes_by_name(results, ["users", "users_rollup", "users_rollup_dependency"])
def test_select_tag_and_children_selector_dict(self, project): # noqa
results = run_dbt(["run", "--selector", "tag_specified_in_project_children_dict"])
_verify_select_tag_and_children(results)
results = run_dbt(
["run", "--selector", "tag_specified_in_project_children_dict"],
expect_pass=False,
)
check_result_nodes_by_name(results, ["users", "users_rollup", "users_rollup_dependency"])
def test_select_tag_in_model_with_project_config(self, project): # noqa
results = run_dbt(["run", "--models", "tag:bi"])
_verify_select_bi(results)
results = run_dbt(["run", "--models", "tag:bi"], expect_pass=False)
check_result_nodes_by_name(results, ["users", "users_rollup"])
def test_select_tag_in_model_with_project_config_selector(self, project): # noqa
results = run_dbt(["run", "--selector", "tagged-bi"])
_verify_select_bi(results)
results = run_dbt(["run", "--selector", "tagged-bi"], expect_pass=False)
check_result_nodes_by_name(results, ["users", "users_rollup"])
# check that model configs aren't squashed by project configs
def test_select_tag_in_model_with_project_config_parents_children(self, project): # noqa
results = run_dbt(["run", "--models", "@tag:users"])
assert len(results) == 4
models_run = set(r.node.name for r in results)
assert {"users", "users_rollup", "emails_alt", "users_rollup_dependency"} == models_run
results = run_dbt(["run", "--models", "@tag:users"], expect_pass=False)
check_result_nodes_by_name(
results, ["users", "users_rollup", "emails_alt", "users_rollup_dependency"]
)
# just the users/users_rollup tests
results = run_dbt(["test", "--models", "@tag:users"])
assert len(results) == 2
assert sorted(r.node.name for r in results) == [
"unique_users_id",
"unique_users_rollup_gender",
]
results = run_dbt(["test", "--models", "@tag:users"], expect_pass=False)
check_result_nodes_by_name(results, ["unique_users_rollup_gender", "unique_users_id"])
# just the email test
results = run_dbt(["test", "--models", "tag:base,config.materialized:ephemeral"])
assert len(results) == 1
assert results[0].node.name == "not_null_emails_email"
results = run_dbt(
["test", "--models", "tag:base,config.materialized:ephemeral"],
expect_pass=False,
)
check_result_nodes_by_name(results, ["not_null_emails_email"])
# also just the email test
results = run_dbt(["test", "--models", "config.severity:warn"])
assert len(results) == 1
assert results[0].node.name == "not_null_emails_email"
results = run_dbt(["test", "--models", "config.severity:warn"], expect_pass=False)
check_result_nodes_by_name(results, ["not_null_emails_email"])
# all 3 tests
results = run_dbt(
["test", "--models", "@tag:users tag:base,config.materialized:ephemeral"]
["test", "--models", "@tag:users tag:base,config.materialized:ephemeral"],
expect_pass=False,
)
check_result_nodes_by_name(
results,
["not_null_emails_email", "unique_users_id", "unique_users_rollup_gender"],
)
assert len(results) == 3
assert sorted(r.node.name for r in results) == [
"not_null_emails_email",
"unique_users_id",
"unique_users_rollup_gender",
]
def test_select_tag_in_model_with_project_config_parents_children_selectors(self, project):
results = run_dbt(["run", "--selector", "user_tagged_childrens_parents"])
assert len(results) == 4
models_run = set(r.node.name for r in results)
assert {"users", "users_rollup", "emails_alt", "users_rollup_dependency"} == models_run
results = run_dbt(
["run", "--selector", "user_tagged_childrens_parents"], expect_pass=False
)
check_result_nodes_by_name(
results, ["users", "users_rollup", "emails_alt", "users_rollup_dependency"]
)
# just the users/users_rollup tests
results = run_dbt(["test", "--selector", "user_tagged_childrens_parents"])
assert len(results) == 2
assert sorted(r.node.name for r in results) == [
"unique_users_id",
"unique_users_rollup_gender",
]
results = run_dbt(
["test", "--selector", "user_tagged_childrens_parents"], expect_pass=False
)
check_result_nodes_by_name(results, ["unique_users_id", "unique_users_rollup_gender"])
# just the email test
results = run_dbt(["test", "--selector", "base_ephemerals"])
assert len(results) == 1
assert results[0].node.name == "not_null_emails_email"
results = run_dbt(["test", "--selector", "base_ephemerals"], expect_pass=False)
check_result_nodes_by_name(results, ["not_null_emails_email"])
# also just the email test
results = run_dbt(["test", "--selector", "warn-severity"])
assert len(results) == 1
assert results[0].node.name == "not_null_emails_email"
results = run_dbt(["test", "--selector", "warn-severity"], expect_pass=False)
check_result_nodes_by_name(results, ["not_null_emails_email"])
# all 3 tests
results = run_dbt(["test", "--selector", "roundabout-everything"])
assert len(results) == 3
assert sorted(r.node.name for r in results) == [
"not_null_emails_email",
"unique_users_id",
"unique_users_rollup_gender",
]
results = run_dbt(["test", "--selector", "roundabout-everything"], expect_pass=False)
check_result_nodes_by_name(
results,
["unique_users_rollup_gender", "unique_users_id", "not_null_emails_email"],
)