sqlmesh/tests/dbt/cli/test_selectors.py at 0d6bdde3813cba3d0448533bca1445cc56bd66b6 · SQLMesh/sqlmesh · 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
import typing as t
import pytest
from sqlmesh_dbt import selectors


@pytest.mark.parametrize(
    "dbt_select,expected",
    [
        ([], None),
        (["main.model_a"], "main.model_a"),
        (["main.model_a main.model_b"], "main.model_a | main.model_b"),
        (["main.model_a", "main.model_b"], "main.model_a | main.model_b"),
        (["(main.model_a & ^main.model_b)"], "(main.model_a & ^main.model_b)"),
        (
            ["(+main.model_a & ^main.model_b)", "main.model_c"],
            "(+main.model_a & ^main.model_b) | main.model_c",
        ),
    ],
)
def test_selection(dbt_select: t.List[str], expected: t.Optional[str]):
    assert selectors.to_sqlmesh(dbt_select=dbt_select, dbt_exclude=[]) == expected


@pytest.mark.parametrize(
    "dbt_exclude,expected",
    [
        ([], None),
        (["main.model_a"], "^(main.model_a)"),
        (["(main.model_a & main.model_b)"], "^(main.model_a & main.model_b)"),
        (["main.model_a,main.model_b"], "^(main.model_a & main.model_b)"),
        (["main.model_a +main.model_b"], "^(main.model_a | +main.model_b)"),
        (
            ["(+main.model_a & ^main.model_b)", "main.model_c"],
            "^((+main.model_a & ^main.model_b) | main.model_c)",
        ),
    ],
)
def test_exclusion(dbt_exclude: t.List[str], expected: t.Optional[str]):
    assert selectors.to_sqlmesh(dbt_select=[], dbt_exclude=dbt_exclude) == expected


@pytest.mark.parametrize(
    "dbt_select,dbt_exclude,expected",
    [
        ([], [], None),
        (["+main.model_a"], ["raw.src_data"], "+main.model_a & ^(raw.src_data)"),
        (
            ["+main.model_a", "main.*b+"],
            ["raw.src_data"],
            "(+main.model_a | main.*b+) & ^(raw.src_data)",
        ),
        (
            ["+main.model_a", "main.*b+"],
            ["raw.src_data", "tag:disabled"],
            "(+main.model_a | main.*b+) & ^(raw.src_data | tag:disabled)",
        ),
    ],
)
def test_selection_and_exclusion(
    dbt_select: t.List[str], dbt_exclude: t.List[str], expected: t.Optional[str]
):
    assert selectors.to_sqlmesh(dbt_select=dbt_select, dbt_exclude=dbt_exclude) == expected


@pytest.mark.parametrize(
    "expression,expected",
    [
        ("", ([], [])),
        ("model_a", (["model_a"], [])),
        ("model_a model_b", (["model_a", "model_b"], [])),
        ("model_a,model_b", ([], ["model_a", "model_b"])),
        ("model_a model_b,model_c", (["model_a"], ["model_b", "model_c"])),
        ("model_a,model_b model_c", (["model_c"], ["model_a", "model_b"])),
    ],
)
def test_split_unions_and_intersections(
    expression: str, expected: t.Tuple[t.List[str], t.List[str]]
):
    assert selectors._split_unions_and_intersections(expression) == expected