forked from dbt-labs/dbt-core
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_macro_calls.py
52 lines (42 loc) · 2.26 KB
/
test_macro_calls.py
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
import os
import unittest
from unittest.mock import MagicMock, patch
from dataclasses import dataclass, field
from typing import Dict, Any
from dbt.clients.jinja_static import statically_extract_macro_calls
from dbt.context.base import generate_base_context
class MacroCalls(unittest.TestCase):
def setUp(self):
self.macro_strings = [
"{% macro parent_macro() %} {% do return(nested_macro()) %} {% endmacro %}",
"{% macro lr_macro() %} {{ return(load_result('relations').table) }} {% endmacro %}",
"{% macro get_snapshot_unique_id() -%} {{ return(adapter.dispatch('get_snapshot_unique_id')()) }} {%- endmacro %}",
"{% macro get_columns_in_query(select_sql) -%} {{ return(adapter.dispatch('get_columns_in_query')(select_sql)) }} {% endmacro %}",
"""{% macro test_mutually_exclusive_ranges(model) %}
with base as (
select {{ get_snapshot_unique_id() }} as dbt_unique_id,
*
from {{ model }} )
{% endmacro %}""",
"{% macro test_my_test(model) %} select {{ dbt_utils.current_timestamp() }} {% endmacro %}",
"{% macro some_test(model) -%} {{ return(adapter.dispatch('test_some_kind4', 'foo_utils4')) }} {%- endmacro %}",
"{% macro some_test(model) -%} {{ return(adapter.dispatch('test_some_kind5', macro_namespace = 'foo_utils5')) }} {%- endmacro %}",
]
self.possible_macro_calls = [
['nested_macro'],
['load_result'],
['get_snapshot_unique_id'],
['get_columns_in_query'],
['get_snapshot_unique_id'],
['dbt_utils.current_timestamp'],
['test_some_kind4', 'foo_utils4.test_some_kind4'],
['test_some_kind5', 'foo_utils5.test_some_kind5'],
]
def test_macro_calls(self):
cli_vars = {'local_utils_dispatch_list': ['foo_utils4']}
ctx = generate_base_context(cli_vars)
index = 0
for macro_string in self.macro_strings:
possible_macro_calls = statically_extract_macro_calls(macro_string, ctx)
self.assertEqual(self.possible_macro_calls[index], possible_macro_calls)
index = index + 1