diff --git a/airbyte_cdk/sources/declarative/interpolation/macros.py b/airbyte_cdk/sources/declarative/interpolation/macros.py index 9b8aca336..dcbaa306a 100644 --- a/airbyte_cdk/sources/declarative/interpolation/macros.py +++ b/airbyte_cdk/sources/declarative/interpolation/macros.py @@ -16,6 +16,7 @@ from isodate import parse_duration from airbyte_cdk.sources.declarative.datetime.datetime_parser import DatetimeParser +from airbyte_cdk.sources.declarative.interpolation.filters import regex_replace, regex_search """ This file contains macros that can be evaluated by a `JinjaInterpolation` object @@ -232,5 +233,7 @@ def generate_uuid() -> str: sanitize_url, camel_case_to_snake_case, generate_uuid, + regex_replace, + regex_search, ] macros = {f.__name__: f for f in _macros_list} diff --git a/unit_tests/sources/declarative/interpolation/test_filters.py b/unit_tests/sources/declarative/interpolation/test_filters.py index 5ffd2aff3..29e52e2f5 100644 --- a/unit_tests/sources/declarative/interpolation/test_filters.py +++ b/unit_tests/sources/declarative/interpolation/test_filters.py @@ -138,6 +138,56 @@ def test_regex_replace(expression: str, expected: str) -> None: assert val == expected +@pytest.mark.parametrize( + "expression, expected", + [ + pytest.param( + "{{ regex_replace('hello world', 'world', 'there') }}", + "hello there", + id="basic_replacement", + ), + pytest.param( + "{{ regex_replace('abc123def456', '[0-9]+', '') }}", + "abcdef", + id="regex_pattern_strip_digits", + ), + pytest.param( + "{{ regex_replace('hello world', 'xyz', 'replaced') }}", + "hello world", + id="no_match_returns_original", + ), + pytest.param( + "{{ regex_replace('aaa bbb aaa', 'aaa', 'ccc') }}", + "ccc bbb ccc", + id="multiple_occurrences", + ), + ], +) +def test_regex_replace_as_macro(expression: str, expected: str) -> None: + val = interpolation.eval(expression, {}) + assert val == expected + + +@pytest.mark.parametrize( + "expression, expected", + [ + pytest.param( + "{{ regex_search('; rel=\"next\"', '<(.*)>; rel=.*') }}", + "https://example.com/?page=2", + id="valid_match", + ), + pytest.param( + "{{ regex_search('no match here', 'WATWATWAT') }}", + None, + id="no_match", + ), + ], +) +def test_regex_search_as_macro(expression: str, expected: str) -> None: + val = interpolation.eval(expression, {}) + assert val == expected + + def test_hmac_sha256_default() -> None: message = "test_message" secret_key = "test_secret_key"