sqlmesh/tests/core/linter/test_helpers.py at 88473258d584dc2b635d68ea605b0e4944e68f59 · 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
from sqlmesh import Context
from sqlmesh.core.linter.helpers import (
    read_range_from_file,
    get_range_of_model_block,
    get_range_of_a_key_in_model_block,
)
from sqlmesh.core.model import SqlModel


def test_get_position_of_model_block():
    context = Context(paths=["examples/sushi"])

    sql_models = [
        model
        for model in context.models.values()
        if isinstance(model, SqlModel)
        and model._path is not None
        and str(model._path).endswith(".sql")
    ]
    assert len(sql_models) > 0

    for model in sql_models:
        dialect = model.dialect
        assert dialect is not None

        path = model._path
        assert path is not None

        with open(path, "r", encoding="utf-8") as file:
            content = file.read()

        as_lines = content.splitlines()

        range = get_range_of_model_block(content, dialect)
        assert range is not None

        #  Check that the range starts with MODEL and ends with ;
        read_range = read_range_from_file(path, range)
        assert read_range.startswith("MODEL")
        assert read_range.endswith(";")


def test_get_range_of_a_key_in_model_block_testing_on_sushi():
    context = Context(paths=["examples/sushi"])

    sql_models = [
        model
        for model in context.models.values()
        if isinstance(model, SqlModel)
        and model._path is not None
        and str(model._path).endswith(".sql")
    ]
    assert len(sql_models) > 0

    for model in sql_models:
        possible_keys = ["name", "tags", "description", "columns", "owner", "cron", "dialect"]

        dialect = model.dialect
        assert dialect is not None

        path = model._path
        assert path is not None

        with open(path, "r", encoding="utf-8") as file:
            content = file.read()

        count_properties_checked = 0

        for key in possible_keys:
            range = get_range_of_a_key_in_model_block(content, dialect, key)

            # Check that the range starts with the key and ends with ;
            if range:
                read_range = read_range_from_file(path, range)
                assert read_range.lower() == key.lower()
                count_properties_checked += 1

        assert count_properties_checked > 0