dbt-sqlserver/tests/functional/adapter/dbt/test_empty.py at 61f949f480e0eb4e973ef2d55b082a907b391a96 · Benjamin-Knight/dbt-sqlserver · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
import pytest

from dbt.tests.adapter.empty._models import model_input_sql, schema_sources_yml

# switch for 1.9
# from dbt.tests.adapter.empty import _models
from dbt.tests.adapter.empty.test_empty import (  # MetadataWithEmptyFlag
    BaseTestEmpty,
    BaseTestEmptyInlineSourceRef,
)
from dbt.tests.util import run_dbt

model_sql_sqlserver = """
select *
from {{ ref('model_input') }} as model_input_alias
union all
select *
from {{ source('seed_sources', 'raw_source') }} as raw_source_alias
"""

model_inline_sql_sqlserver = """
select * from {{ source('seed_sources', 'raw_source') }} as raw_source_alias
"""

model_sql_user_alias_sqlserver = """
select user_alias.id as id
from {{ ref('model_input') }} as user_alias
inner join {{ source('seed_sources', 'raw_source') }} as source_alias
    on user_alias.id = source_alias.id
"""


class TestEmpty(BaseTestEmpty):
    @pytest.fixture(scope="class")
    def models(self):
        return {
            "model_input.sql": model_input_sql,
            # # no support for ephemeral models in SQLServer
            # "ephemeral_model_input.sql": _models.ephemeral_model_input_sql,
            "model.sql": model_sql_sqlserver,
            "sources.yml": schema_sources_yml,
        }

    def test_run_with_empty(self, project):
        # create source from seed
        run_dbt(["seed"])

        # run without empty - 3 expected rows in output - 1 from each input
        run_dbt(["run"])
        self.assert_row_count(project, "model", 2)

        # run with empty - 0 expected rows in output
        run_dbt(["run", "--empty"])
        self.assert_row_count(project, "model", 0)


class TestEmptyWithUserAlias(BaseTestEmpty):
    @pytest.fixture(scope="class")
    def models(self):
        return {
            "model_input.sql": model_input_sql,
            "model.sql": model_sql_user_alias_sqlserver,
            "sources.yml": schema_sources_yml,
        }

    def test_run_with_empty(self, project):
        run_dbt(["seed"])

        run_dbt(["run", "--empty"])
        self.assert_row_count(project, "model", 0)


@pytest.mark.xfail(
    reason="Upstream dbt empty-mode alias handling needs to be contextual aware.",
)
class TestEmptyWithUserAliasAndNoAliasFlag(BaseTestEmpty):
    @pytest.fixture(scope="class")
    def models(self):
        return {
            "model_input.sql": model_input_sql,
            "model.sql": model_sql_user_alias_sqlserver,
            "sources.yml": schema_sources_yml,
        }

    @pytest.fixture(scope="class")
    def project_config_update(self):
        return {
            "flags": {
                "dbt_sqlserver_disable_empty_relation_aliases": False,
            }
        }

    def test_run_with_empty(self, project):
        run_dbt(["seed"])

        run_dbt(["run", "--empty"])
        self.assert_row_count(project, "model", 0)


class TestemptyInlineSourceRef(BaseTestEmptyInlineSourceRef):
    @pytest.fixture(scope="class")
    def models(self):
        return {
            "model.sql": model_inline_sql_sqlserver,
            "sources.yml": schema_sources_yml,
        }