fix: interpolation should not generate complex types (#579)

Co-authored-by: octavia-squidington-iii <[email protected]>

Author: maxi297

airbyte_cdk/sources/declarative/interpolation/jinja.py

@@ -127,7 +127,9 @@ def _literal_eval(self, result: Optional[str], valid_types: Optional[Tuple[Type[
             evaluated = ast.literal_eval(result)  # type: ignore # literal_eval is able to handle None
         except (ValueError, SyntaxError):
             return result
-        if not valid_types or (valid_types and isinstance(evaluated, valid_types)):
+        if (not valid_types and not isinstance(evaluated, complex)) or (
+            valid_types and isinstance(evaluated, valid_types)
+        ):
             return evaluated
         return result
 

debug_manifest/debug_manifest.py

@@ -5,7 +5,7 @@
 import sys
 from typing import Any, Mapping
 
-from airbyte_cdk.entrypoint import launch
+from airbyte_cdk.entrypoint import AirbyteEntrypoint, launch
 from airbyte_cdk.sources.declarative.yaml_declarative_source import (
     YamlDeclarativeSource,
 )
@@ -23,7 +23,17 @@ def debug_manifest(source: YamlDeclarativeSource, args: list[str]) -> None:
 
 
 if __name__ == "__main__":
+    args = sys.argv[1:]
+    catalog_path = AirbyteEntrypoint.extract_catalog(args)
+    config_path = AirbyteEntrypoint.extract_config(args)
+    state_path = AirbyteEntrypoint.extract_state(args)
+
     debug_manifest(
-        YamlDeclarativeSource(**configuration),
-        sys.argv[1:],
+        YamlDeclarativeSource(
+            path_to_yaml="resources/manifest.yaml",
+            catalog=YamlDeclarativeSource.read_catalog(catalog_path) if catalog_path else None,
+            config=YamlDeclarativeSource.read_config(config_path) if config_path else None,
+            state=YamlDeclarativeSource.read_state(state_path) if state_path else None,
+        ),
+        args,
     )

unit_tests/sources/declarative/interpolation/test_jinja.py

@@ -29,20 +29,6 @@ def test_get_missing_value_from_config():
     assert val is None
 
 
-@pytest.mark.parametrize(
-    "valid_types, expected_value",
-    [
-        pytest.param((str,), "1234J", id="test_value_is_a_string_if_valid_types_is_str"),
-        pytest.param(None, 1234j, id="test_value_is_interpreted_as_complex_number_by_default"),
-    ],
-)
-def test_get_value_with_complex_number(valid_types, expected_value):
-    s = "{{ config['value'] }}"
-    config = {"value": "1234J"}
-    val = interpolation.eval(s, config, valid_types=valid_types)
-    assert val == expected_value
-
-
 def test_get_value_from_stream_slice():
     s = "{{ stream_slice['date'] }}"
     config = {"date": "2022-01-01"}
@@ -357,3 +343,21 @@ def test_interpolation_private_partition_attribute():
     actual_output = JinjaInterpolation().eval(template, {}, **{"stream_slice": stream_slice})
 
     assert actual_output == expected_output
+
+
+def test_given_complex_when_eval_then_return_string():
+    s = "9173710294242221J"
+    config = {}
+
+    val = interpolation.eval(s, config)
+
+    assert isinstance(val, str)
+
+
+def test_given_valid_type_complex_when_eval_then_return_string():
+    s = "9173710294242221J"
+    config = {}
+
+    val = interpolation.eval(s, config, valid_types=(complex,))
+
+    assert isinstance(val, complex)