airbyte-python-cdk/unit_tests/sources/declarative/interpolation/test_interpolated_nested_mapping.py at ac74d0a013298586e43d678ea7e823ebe59b64a1 · airbytehq/airbyte-python-cdk · 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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
#
# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
#

import dpath
import pytest

from airbyte_cdk.sources.declarative.interpolation.interpolated_nested_mapping import (
    InterpolatedNestedMapping,
)


@pytest.mark.parametrize(
    "test_name, path, expected_value",
    [
        ("test_field_value", "nested/field", "value"),
        ("test_number", "nested/number", 100),
        ("test_interpolated_number", "nested/nested_array/1/value", 5),
        ("test_interpolated_boolean", "nested/nested_array/2/value", True),
        ("test_field_to_interpolate_from_config", "nested/config_value", "VALUE_FROM_CONFIG"),
        ("test_field_to_interpolate_from_kwargs", "nested/kwargs_value", "VALUE_FROM_KWARGS"),
        (
            "test_field_to_interpolate_from_parameters",
            "nested/parameters_value",
            "VALUE_FROM_PARAMETERS",
        ),
        ("test_key_is_interpolated", "nested/nested_array/0/key", "VALUE"),
    ],
)
def test(test_name, path, expected_value):
    d = {
        "nested": {
            "field": "value",
            "number": 100,
            "nested_array": [
                {"{{ parameters.k }}": "VALUE"},
                {"value": "{{ config['num_value'] | int + 2 }}"},
                {"value": "{{ True }}"},
            ],
            "config_value": "{{ config['c'] }}",
            "parameters_value": "{{ parameters['b'] }}",
            "kwargs_value": "{{ kwargs['a'] }}",
        }
    }

    config = {"c": "VALUE_FROM_CONFIG", "num_value": 3}
    kwargs = {"a": "VALUE_FROM_KWARGS"}
    mapping = InterpolatedNestedMapping(
        mapping=d, parameters={"b": "VALUE_FROM_PARAMETERS", "k": "key"}
    )

    interpolated = mapping.eval(config, **{"kwargs": kwargs})

    assert dpath.get(interpolated, path) == expected_value


@pytest.mark.parametrize(
    "mapping, path, expected_value",
    [
        pytest.param(
            {"clientId": "478"},
            "clientId",
            "478",
            id="static_digit_only_string",
        ),
        pytest.param(
            {"value": "0012"},
            "value",
            "0012",
            id="static_leading_zero_string",
        ),
        pytest.param(
            {"value": "None"},
            "value",
            "None",
            id="static_none_string",
        ),
        pytest.param(
            {"value": '{"field": "updated_at", "order": "ascending"}'},
            "value",
            {"field": "updated_at", "order": "ascending"},
            id="static_structured_dict",
        ),
        pytest.param(
            {"value": "[1, 2, 3]"},
            "value",
            [1, 2, 3],
            id="static_structured_list",
        ),
        pytest.param(
            {"sort": '{"field": "updated_at", "order": "ascending"}'},
            "sort",
            {"field": "updated_at", "order": "ascending"},
            id="static_intercom_sort_json",
        ),
        pytest.param(
            {"478": "value"},
            "478",
            "value",
            id="static_digit_only_key",
        ),
        pytest.param(
            {"nested": {"clientId": "478"}},
            "nested/clientId",
            "478",
            id="static_string_nested_in_dict",
        ),
        pytest.param(
            {"nested": {"values": ["478", "0012", "None"]}},
            "nested/values",
            ["478", "0012", "None"],
            id="static_strings_nested_in_list",
        ),
        pytest.param(
            {"value": "{{ 1 + 1 }}"},
            "value",
            2,
            id="jinja_value_still_coerces",
        ),
    ],
)
def test_static_strings_are_preserved_and_jinja_values_are_interpolated(
    mapping, path, expected_value
):
    interpolated = InterpolatedNestedMapping(mapping=mapping, parameters={}).eval({})

    assert dpath.get(interpolated, path) == expected_value