From e97ece5532ad6ea7159461a5076a0d1ad50694fc Mon Sep 17 00:00:00 2001
From: Brian Lai <51336873+brianjlai@users.noreply.github.com>
Date: Thu, 17 Nov 2022 16:15:28 -0500
Subject: [PATCH] [low-code cdk] break resolving reference preprocessing into
 its own class so it can be reused (#19517)

* break resolving reference preprocessing into its own class so it can be reused

* move reference resolution into the ManifestDeclarativeSource and deprecate the parser

* formatting

* last formatting i promise

* rename

* bump version

Co-authored-by: Alexandre Girard <alexandre@airbyte.io>
---
 airbyte-cdk/python/CHANGELOG.md               |   3 +
 .../manifest_declarative_source.py            |  10 +-
 .../declarative/parsers/config_parser.py      |  17 ---
 ...rser.py => manifest_reference_resolver.py} |  32 +---
 .../declarative/yaml_declarative_source.py    |  14 +-
 airbyte-cdk/python/setup.py                   |   2 +-
 .../test_manifest_reference_resolver.py       | 143 +++++++++++++++++
 .../declarative/parsers/test_yaml_parser.py   | 144 ------------------
 .../sources/declarative/test_factory.py       |  51 ++++---
 .../test_yaml_declarative_source.py           |   3 +-
 10 files changed, 201 insertions(+), 218 deletions(-)
 delete mode 100644 airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/config_parser.py
 rename airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/{yaml_parser.py => manifest_reference_resolver.py} (80%)
 create mode 100644 airbyte-cdk/python/unit_tests/sources/declarative/parsers/test_manifest_reference_resolver.py
 delete mode 100644 airbyte-cdk/python/unit_tests/sources/declarative/parsers/test_yaml_parser.py

diff --git a/airbyte-cdk/python/CHANGELOG.md b/airbyte-cdk/python/CHANGELOG.md
index d1690cdf748b85..5adba2bc6aec11 100644
--- a/airbyte-cdk/python/CHANGELOG.md
+++ b/airbyte-cdk/python/CHANGELOG.md
@@ -1,5 +1,8 @@
 # Changelog
 
+## 0.9.4
+Low-code: Fix reference resolution for connector builder
+
 ## 0.9.3
 Low-code: Avoid duplicate HTTP query in `simple_retriever`
 
diff --git a/airbyte-cdk/python/airbyte_cdk/sources/declarative/manifest_declarative_source.py b/airbyte-cdk/python/airbyte_cdk/sources/declarative/manifest_declarative_source.py
index f44342d14ffe36..240799fe3ad1d4 100644
--- a/airbyte-cdk/python/airbyte_cdk/sources/declarative/manifest_declarative_source.py
+++ b/airbyte-cdk/python/airbyte_cdk/sources/declarative/manifest_declarative_source.py
@@ -23,6 +23,7 @@
 from airbyte_cdk.sources.declarative.declarative_stream import DeclarativeStream
 from airbyte_cdk.sources.declarative.exceptions import InvalidConnectorDefinitionException
 from airbyte_cdk.sources.declarative.parsers.factory import DeclarativeComponentFactory
+from airbyte_cdk.sources.declarative.parsers.manifest_reference_resolver import ManifestReferenceResolver
 from airbyte_cdk.sources.declarative.types import ConnectionDefinition
 from airbyte_cdk.sources.streams.core import Stream
 from dataclasses_jsonschema import JsonSchemaMixin
@@ -47,7 +48,10 @@ def __init__(self, source_config: ConnectionDefinition, debug: bool = False):
         :param debug(bool): True if debug mode is enabled
         """
         self.logger = logging.getLogger(f"airbyte.{self.name}")
-        self._source_config = source_config
+
+        evaluated_manifest = {}
+        resolved_source_config = ManifestReferenceResolver().preprocess_manifest(source_config, evaluated_manifest, "")
+        self._source_config = resolved_source_config
         self._debug = debug
         self._factory = DeclarativeComponentFactory()
 
@@ -135,8 +139,8 @@ def _stream_configs(self):
 
     @staticmethod
     def generate_schema() -> str:
-        expanded_source_definition = ManifestDeclarativeSource.expand_schema_interfaces(ConcreteDeclarativeSource, {})
-        expanded_schema = expanded_source_definition.json_schema()
+        expanded_source_manifest = ManifestDeclarativeSource.expand_schema_interfaces(ConcreteDeclarativeSource, {})
+        expanded_schema = expanded_source_manifest.json_schema()
         return json.dumps(expanded_schema, cls=SchemaEncoder)
 
     @staticmethod
diff --git a/airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/config_parser.py b/airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/config_parser.py
deleted file mode 100644
index 06c61f12154480..00000000000000
--- a/airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/config_parser.py
+++ /dev/null
@@ -1,17 +0,0 @@
-#
-# Copyright (c) 2022 Airbyte, Inc., all rights reserved.
-#
-
-from abc import ABC, abstractmethod
-
-from airbyte_cdk.sources.declarative.types import ConnectionDefinition
-
-
-class ConnectionDefinitionParser(ABC):
-    """
-    Parses a string to a ConnectionDefinition
-    """
-
-    @abstractmethod
-    def parse(self, config_str: str) -> ConnectionDefinition:
-        """Parses the config_str to a ConnectionDefinition"""
diff --git a/airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/yaml_parser.py b/airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/manifest_reference_resolver.py
similarity index 80%
rename from airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/yaml_parser.py
rename to airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/manifest_reference_resolver.py
index 31518c74849b5d..04af303c87f1d8 100644
--- a/airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/yaml_parser.py
+++ b/airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/manifest_reference_resolver.py
@@ -5,17 +5,12 @@
 from copy import deepcopy
 from typing import Any, Mapping, Tuple, Union
 
-import yaml
-from airbyte_cdk.sources.declarative.parsers.config_parser import ConnectionDefinitionParser
 from airbyte_cdk.sources.declarative.parsers.undefined_reference_exception import UndefinedReferenceException
-from airbyte_cdk.sources.declarative.types import ConnectionDefinition
 
 
-class YamlParser(ConnectionDefinitionParser):
+class ManifestReferenceResolver:
     """
-    Parses a Yaml string to a ConnectionDefinition
-
-    In addition to standard Yaml parsing, the input_string can contain references to values previously defined.
+    An incoming manifest can contain references to values previously defined.
     This parser will dereference these values to produce a complete ConnectionDefinition.
 
     References can be defined using a *ref(<arg>) string.
@@ -101,31 +96,20 @@ class YamlParser(ConnectionDefinitionParser):
 
     ref_tag = "$ref"
 
-    def parse(self, connection_definition_str: str) -> ConnectionDefinition:
-        """
-        Parses a yaml file and dereferences string in the form "*ref({reference)"
-        to {reference}
-        :param connection_definition_str: yaml string to parse
-        :return: The ConnectionDefinition parsed from connection_definition_str
-        """
-        input_mapping = yaml.safe_load(connection_definition_str)
-        evaluated_definition = {}
-        return self._preprocess_dict(input_mapping, evaluated_definition, "")
-
-    def _preprocess_dict(self, input_mapping: Mapping[str, Any], evaluated_mapping: Mapping[str, Any], path: Union[str, Tuple[str]]):
+    def preprocess_manifest(self, manifest: Mapping[str, Any], evaluated_mapping: Mapping[str, Any], path: Union[str, Tuple[str]]):
 
         """
-        :param input_mapping: mapping produced by parsing yaml
+        :param manifest: incoming manifest that could have references to previously defined components
         :param evaluated_mapping: mapping produced by dereferencing the content of input_mapping
         :param path: curent path in configuration traversal
         :return:
         """
         d = {}
-        if self.ref_tag in input_mapping:
-            partial_ref_string = input_mapping[self.ref_tag]
+        if self.ref_tag in manifest:
+            partial_ref_string = manifest[self.ref_tag]
             d = deepcopy(self._preprocess(partial_ref_string, evaluated_mapping, path))
 
-        for key, value in input_mapping.items():
+        for key, value in manifest.items():
             if key == self.ref_tag:
                 continue
             full_path = self._resolve_value(key, path)
@@ -180,7 +164,7 @@ def _preprocess(self, value, evaluated_config: Mapping[str, Any], path):
                         key = *key[:-1], split[0], ".".join(split[1:])
                 raise UndefinedReferenceException(path, ref_key)
         elif isinstance(value, dict):
-            return self._preprocess_dict(value, evaluated_config, path)
+            return self.preprocess_manifest(value, evaluated_config, path)
         elif type(value) == list:
             evaluated_list = [
                 # pass in elem's path instead of the list's path
diff --git a/airbyte-cdk/python/airbyte_cdk/sources/declarative/yaml_declarative_source.py b/airbyte-cdk/python/airbyte_cdk/sources/declarative/yaml_declarative_source.py
index 1907b8284b76cf..88d449192fe6f9 100644
--- a/airbyte-cdk/python/airbyte_cdk/sources/declarative/yaml_declarative_source.py
+++ b/airbyte-cdk/python/airbyte_cdk/sources/declarative/yaml_declarative_source.py
@@ -4,8 +4,8 @@
 
 import pkgutil
 
+import yaml
 from airbyte_cdk.sources.declarative.manifest_declarative_source import ManifestDeclarativeSource
-from airbyte_cdk.sources.declarative.parsers.yaml_parser import YamlParser
 from airbyte_cdk.sources.declarative.types import ConnectionDefinition
 
 
@@ -25,8 +25,18 @@ def _read_and_parse_yaml_file(self, path_to_yaml_file) -> ConnectionDefinition:
 
         yaml_config = pkgutil.get_data(package, path_to_yaml_file)
         decoded_yaml = yaml_config.decode()
-        return YamlParser().parse(decoded_yaml)
+        return self._parse(decoded_yaml)
 
     def _emit_manifest_debug_message(self, extra_args: dict):
         extra_args["path_to_yaml"] = self._path_to_yaml
         self.logger.debug("declarative source created from parsed YAML manifest", extra=extra_args)
+
+    @staticmethod
+    def _parse(connection_definition_str: str) -> ConnectionDefinition:
+        """
+        Parses a yaml file into a manifest. Component references still exist in the manifest which will be
+        resolved during the creating of the DeclarativeSource.
+        :param connection_definition_str: yaml string to parse
+        :return: The ConnectionDefinition parsed from connection_definition_str
+        """
+        return yaml.safe_load(connection_definition_str)
diff --git a/airbyte-cdk/python/setup.py b/airbyte-cdk/python/setup.py
index eb85d487532874..92bab1beefcfe4 100644
--- a/airbyte-cdk/python/setup.py
+++ b/airbyte-cdk/python/setup.py
@@ -15,7 +15,7 @@
 
 setup(
     name="airbyte-cdk",
-    version="0.9.3",
+    version="0.9.4",
     description="A framework for writing Airbyte Connectors.",
     long_description=README,
     long_description_content_type="text/markdown",
diff --git a/airbyte-cdk/python/unit_tests/sources/declarative/parsers/test_manifest_reference_resolver.py b/airbyte-cdk/python/unit_tests/sources/declarative/parsers/test_manifest_reference_resolver.py
new file mode 100644
index 00000000000000..59f7412ef1a1c9
--- /dev/null
+++ b/airbyte-cdk/python/unit_tests/sources/declarative/parsers/test_manifest_reference_resolver.py
@@ -0,0 +1,143 @@
+#
+# Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+#
+
+import pytest
+from airbyte_cdk.sources.declarative.parsers.manifest_reference_resolver import ManifestReferenceResolver
+from airbyte_cdk.sources.declarative.parsers.undefined_reference_exception import UndefinedReferenceException
+
+resolver = ManifestReferenceResolver()
+
+
+def test_get_ref():
+    s = "*ref(limit)"
+    ref_key = resolver._get_ref_key(s)
+    assert ref_key == "limit"
+
+
+def test_get_ref_no_ref():
+    s = "limit: 50"
+
+    ref_key = resolver._get_ref_key(s)
+    assert ref_key is None
+
+
+def test_refer():
+    content = {
+        "limit": 50,
+        "limit_ref": "*ref(limit)"
+    }
+    config = resolver.preprocess_manifest(content, {}, "")
+    assert config["limit_ref"] == 50
+
+
+def test_refer_to_inner():
+    content = {
+        "dict": {
+            "limit": 50
+        },
+        "limit_ref": "*ref(dict.limit)"
+    }
+    config = resolver.preprocess_manifest(content, {}, "")
+    assert config["limit_ref"] == 50
+
+
+def test_refer_to_non_existant_struct():
+    content = {
+        "dict": {
+            "limit": 50
+        },
+        "limit_ref": "*ref(not_dict)"
+    }
+    with pytest.raises(UndefinedReferenceException):
+        resolver.preprocess_manifest(content, {}, "")
+
+
+def test_refer_in_dict():
+    content = {
+        "limit": 50,
+        "offset_request_parameters": {
+            "offset": "{{ next_page_token['offset'] }}",
+            "limit": "*ref(limit)"
+        }
+    }
+    config = resolver.preprocess_manifest(content, {}, "")
+    assert config["offset_request_parameters"]["offset"] == "{{ next_page_token['offset'] }}"
+    assert config["offset_request_parameters"]["limit"] == 50
+
+
+def test_refer_to_dict():
+    content = {
+        "limit": 50,
+        "offset_request_parameters": {
+            "offset": "{{ next_page_token['offset'] }}",
+            "limit": "*ref(limit)"
+        },
+        "offset_pagination_request_parameters": {
+            "class": "InterpolatedRequestParameterProvider",
+            "request_parameters": "*ref(offset_request_parameters)"
+        }
+    }
+    config = resolver.preprocess_manifest(content, {}, "")
+    assert config["limit"] == 50
+    assert config["offset_request_parameters"]["limit"] == 50
+    assert len(config["offset_pagination_request_parameters"]) == 2
+    assert config["offset_pagination_request_parameters"]["request_parameters"]["limit"] == 50
+    assert config["offset_pagination_request_parameters"]["request_parameters"]["offset"] == "{{ next_page_token['offset'] }}"
+
+
+def test_refer_and_overwrite():
+    content = {
+        "limit": 50,
+        "custom_limit": 25,
+        "offset_request_parameters": {
+            "offset": "{{ next_page_token['offset'] }}",
+            "limit": "*ref(limit)"
+        },
+        "custom_request_parameters": {
+            "$ref": "*ref(offset_request_parameters)",
+            "limit": "*ref(custom_limit)"
+        }
+    }
+    config = resolver.preprocess_manifest(content, {}, "")
+    assert config["offset_request_parameters"]["limit"] == 50
+    assert config["custom_request_parameters"]["limit"] == 25
+
+    assert config["offset_request_parameters"]["offset"] == "{{ next_page_token['offset'] }}"
+    assert config["custom_request_parameters"]["offset"] == "{{ next_page_token['offset'] }}"
+
+
+def test_collision():
+    content = {
+        "example": {
+            "nested":{
+                "path": "first one",
+                "more_nested": {
+                    "value": "found it!"
+                }
+            },
+            "nested.path": "uh oh",
+        },
+        "reference_to_nested_path": {
+            "$ref": "*ref(example.nested.path)"
+        },
+        "reference_to_nested_nested_value": {
+            "$ref": "*ref(example.nested.more_nested.value)"
+        }
+    }
+    config = resolver.preprocess_manifest(content, {}, "")
+    assert config["example"]["nested"]["path"] == "first one"
+    assert config["example"]["nested.path"] == "uh oh"
+    assert config["reference_to_nested_path"] == "uh oh"
+    assert config["example"]["nested"]["more_nested"]["value"] == "found it!"
+    assert config["reference_to_nested_nested_value"] == "found it!"
+
+
+def test_list():
+    content = {
+        "list": ["A", "B"],
+        "elem_ref": "*ref(list[0])"
+    }
+    config = resolver.preprocess_manifest(content, {}, "")
+    elem_ref = config["elem_ref"]
+    assert elem_ref == "A"
diff --git a/airbyte-cdk/python/unit_tests/sources/declarative/parsers/test_yaml_parser.py b/airbyte-cdk/python/unit_tests/sources/declarative/parsers/test_yaml_parser.py
deleted file mode 100644
index bbc9104ab1b7f4..00000000000000
--- a/airbyte-cdk/python/unit_tests/sources/declarative/parsers/test_yaml_parser.py
+++ /dev/null
@@ -1,144 +0,0 @@
-#
-# Copyright (c) 2022 Airbyte, Inc., all rights reserved.
-#
-
-import pytest
-from airbyte_cdk.sources.declarative.parsers.undefined_reference_exception import UndefinedReferenceException
-from airbyte_cdk.sources.declarative.parsers.yaml_parser import YamlParser
-
-parser = YamlParser()
-
-
-def test():
-    content = """
-    limit: 50
-    """
-    config = parser.parse(content)
-    assert config["limit"] == 50
-
-
-def test_get_ref():
-    s = """
-    limit_ref: "*ref(limit)"
-    """
-    ref_key = parser._get_ref_key(s)
-    assert ref_key == "limit"
-
-
-def test_get_ref_no_ref():
-    s = """
-    limit: 50
-    """
-    ref_key = parser._get_ref_key(s)
-    assert ref_key is None
-
-
-def test_refer():
-    content = """
-    limit: 50
-    limit_ref: "*ref(limit)"
-    """
-    config = parser.parse(content)
-    assert config["limit_ref"] == 50
-
-
-def test_refer_to_inner():
-    content = """
-    dict:
-      limit: 50
-    limit_ref: "*ref(dict.limit)"
-    """
-    config = parser.parse(content)
-    assert config["limit_ref"] == 50
-
-
-def test_refer_to_non_existant_struct():
-    content = """
-    dict:
-      limit: 50
-    limit_ref: "*ref(not_dict)"
-    """
-    with pytest.raises(UndefinedReferenceException):
-        parser.parse(content)
-
-
-def test_refer_in_dict():
-    content = """
-    limit: 50
-    offset_request_parameters:
-      offset: "{{ next_page_token['offset'] }}"
-      limit: "*ref(limit)"
-    """
-    config = parser.parse(content)
-    assert config["offset_request_parameters"]["offset"] == "{{ next_page_token['offset'] }}"
-    assert config["offset_request_parameters"]["limit"] == 50
-
-
-def test_refer_to_dict():
-    content = """
-    limit: 50
-    offset_request_parameters:
-      offset: "{{ next_page_token['offset'] }}"
-      limit: "*ref(limit)"
-    offset_pagination_request_parameters:
-      class: InterpolatedRequestParameterProvider
-      request_parameters: "*ref(offset_request_parameters)"
-    """
-    config = parser.parse(content)
-    assert config["limit"] == 50
-    assert config["offset_request_parameters"]["limit"] == 50
-    assert len(config["offset_pagination_request_parameters"]) == 2
-    assert config["offset_pagination_request_parameters"]["request_parameters"]["limit"] == 50
-    assert config["offset_pagination_request_parameters"]["request_parameters"]["offset"] == "{{ next_page_token['offset'] }}"
-
-
-def test_refer_and_overwrite():
-    content = """
-    limit: 50
-    custom_limit: 25
-    offset_request_parameters:
-      offset: "{{ next_page_token['offset'] }}"
-      limit: "*ref(limit)"
-    custom_request_parameters:
-      $ref: "*ref(offset_request_parameters)"
-      limit: "*ref(custom_limit)"
-    """
-    config = parser.parse(content)
-    assert config["offset_request_parameters"]["limit"] == 50
-    assert config["custom_request_parameters"]["limit"] == 25
-
-    assert config["offset_request_parameters"]["offset"] == "{{ next_page_token['offset'] }}"
-    assert config["custom_request_parameters"]["offset"] == "{{ next_page_token['offset'] }}"
-
-
-def test_collision():
-    content = """
-example:
-  nested:
-    path: "first one"
-    more_nested:
-        value: "found it!"
-  nested.path: "uh oh"
-reference_to_nested_path:
-  $ref: "*ref(example.nested.path)"
-reference_to_nested_nested_value:
-  $ref: "*ref(example.nested.more_nested.value)"
-    """
-    config = parser.parse(content)
-    assert config["example"]["nested"]["path"] == "first one"
-    assert config["example"]["nested.path"] == "uh oh"
-    assert config["reference_to_nested_path"] == "uh oh"
-    assert config["example"]["nested"]["more_nested"]["value"] == "found it!"
-    assert config["reference_to_nested_nested_value"] == "found it!"
-
-
-def test_list():
-    content = """
-    list:
-      - "A"
-      - "B"
-    elem_ref: "*ref(list[0])"
-    """
-    config = parser.parse(content)
-    elem_ref = config["elem_ref"]
-    assert elem_ref == "A"
diff --git a/airbyte-cdk/python/unit_tests/sources/declarative/test_factory.py b/airbyte-cdk/python/unit_tests/sources/declarative/test_factory.py
index 88ad065ac69099..bb47c97097fc5a 100644
--- a/airbyte-cdk/python/unit_tests/sources/declarative/test_factory.py
+++ b/airbyte-cdk/python/unit_tests/sources/declarative/test_factory.py
@@ -15,7 +15,7 @@
 from airbyte_cdk.sources.declarative.extractors.record_selector import RecordSelector
 from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
 from airbyte_cdk.sources.declarative.parsers.factory import DeclarativeComponentFactory
-from airbyte_cdk.sources.declarative.parsers.yaml_parser import YamlParser
+from airbyte_cdk.sources.declarative.parsers.manifest_reference_resolver import ManifestReferenceResolver
 from airbyte_cdk.sources.declarative.requesters.error_handlers import BackoffStrategy
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies import (
     ConstantBackoffStrategy,
@@ -40,12 +40,13 @@
 from airbyte_cdk.sources.declarative.stream_slicers.list_stream_slicer import ListStreamSlicer
 from airbyte_cdk.sources.declarative.transformations import AddFields, RemoveFields
 from airbyte_cdk.sources.declarative.transformations.add_fields import AddedFieldDefinition
+from airbyte_cdk.sources.declarative.yaml_declarative_source import YamlDeclarativeSource
 from dateutil.relativedelta import relativedelta
 from jsonschema import ValidationError
 
 factory = DeclarativeComponentFactory()
 
-parser = YamlParser()
+resolver = ManifestReferenceResolver()
 
 input_config = {"apikey": "verysecrettoken", "repos": ["airbyte", "airbyte-cloud"]}
 
@@ -64,7 +65,7 @@ def test_factory():
       request_body_json:
         body_offset: "{{ next_page_token['offset'] }}"
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
     factory.create_component(config["request_options"], input_config, False)
 
@@ -89,7 +90,7 @@ def test_interpolate_config():
         body_field: "yoyoyo"
         interpolated_body_field: "{{ config['apikey'] }}"
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
     factory.create_component(config["authenticator"], input_config, False)
 
@@ -111,7 +112,7 @@ def test_list_based_stream_slicer_with_values_refd():
       slice_values: "*ref(repositories)"
       cursor_field: repository
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
     factory.create_component(config["stream_slicer"], input_config, False)
 
@@ -129,7 +130,7 @@ def test_list_based_stream_slicer_with_values_defined_in_config():
         inject_into: header
         field_name: repository
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
     factory.create_component(config["stream_slicer"], input_config, False)
 
@@ -181,7 +182,7 @@ def test_create_substream_slicer():
           parent_key: someid
           stream_slice_field: word_id
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
     stream_slicer = factory.create_component(config["stream_slicer"], input_config)()
     parent_stream_configs = stream_slicer.parent_stream_configs
@@ -216,7 +217,7 @@ def test_create_cartesian_stream_slicer():
         - "*ref(stream_slicer_A)"
         - "*ref(stream_slicer_B)"
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
     factory.create_component(config["stream_slicer"], input_config, False)
 
@@ -248,7 +249,7 @@ def test_datetime_stream_slicer():
           field_name: created[gte]
     """
 
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
     factory.create_component(config["stream_slicer"], input_config, False)
 
@@ -362,7 +363,7 @@ def test_full_config():
         description: Test API Key
         order: 0
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
     factory.create_component(config["list_stream"], input_config, False)
 
@@ -429,7 +430,7 @@ def test_create_record_selector(test_name, record_selector, expected_runtime_sel
         $ref: "*ref(extractor)"
         field_pointer: ["{record_selector}"]
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
     factory.create_component(config["selector"], input_config, False)
 
@@ -476,7 +477,7 @@ def test_create_record_selector(test_name, record_selector, expected_runtime_sel
     ],
 )
 def test_options_propagation(test_name, content, expected_field_pointer_value):
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
     selector = factory.create_component(config["selector"], input_config, True)()
     assert selector.extractor.field_pointer[0].eval(input_config) == expected_field_pointer_value
@@ -546,7 +547,7 @@ def test_create_requester(test_name, error_handler):
         header: header_value
     {error_handler}
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
     factory.create_component(config["requester"], input_config, False)
 
@@ -576,7 +577,7 @@ def test_create_composite_error_handler():
                 - http_codes: [ 403 ]
                   action: RETRY
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
     factory.create_component(config["error_handler"], input_config, False)
 
@@ -625,7 +626,7 @@ def test_config_with_defaults():
     streams:
       - "*ref(lists_stream)"
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
     factory.create_component(config["lists_stream"], input_config, False)
 
@@ -662,7 +663,7 @@ def test_create_default_paginator():
           page_size: 50
           cursor_value: "{{ response._metadata.next }}"
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
     factory.create_component(config["paginator"], input_config, False)
 
@@ -701,7 +702,7 @@ def test_no_transformations(self):
             $options:
                 {self.base_options}
         """
-        config = parser.parse(content)
+        config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
         factory.create_component(config["the_stream"], input_config, False)
 
@@ -721,7 +722,7 @@ def test_remove_fields(self):
                         - ["path", "to", "field1"]
                         - ["path2"]
         """
-        config = parser.parse(content)
+        config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
         factory.create_component(config["the_stream"], input_config, False)
 
@@ -742,7 +743,7 @@ def test_add_fields(self):
                         - path: ["field1"]
                           value: "static_value"
         """
-        config = parser.parse(content)
+        config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
 
         factory.create_component(config["the_stream"], input_config, False)
 
@@ -774,7 +775,7 @@ def test_validation_wrong_input_type():
         $ref: "*ref(extractor)"
         field_pointer: 408
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
     with pytest.raises(ValidationError):
         factory.create_component(config["selector"], input_config, False)
 
@@ -797,7 +798,7 @@ def test_validation_type_missing_required_fields():
         field_name: created[gte]
     """
 
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
     with pytest.raises(ValidationError):
         factory.create_component(config["stream_slicer"], input_config, False)
 
@@ -817,7 +818,7 @@ def test_validation_wrong_interface_type():
         type: "MinMaxDatetime"
         datetime: "{{ response._metadata.next }}"
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
     with pytest.raises(ValidationError):
         factory.create_component(config["paginator"], input_config, False)
 
@@ -833,7 +834,7 @@ def test_validation_create_composite_error_handler():
             - response_filters:
                 - http_codes: [ 403 ]
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
     with pytest.raises(ValidationError):
         factory.create_component(config["error_handler"], input_config, False)
 
@@ -858,7 +859,7 @@ def test_validation_wrong_object_type():
           type: "MinMaxDatetime"
           datetime: "{{ response._metadata.next }}"
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
     factory.create_component(config["paginator"], input_config, False)
 
 
@@ -872,7 +873,7 @@ def test_validate_types_nested_in_list():
         - type: DpathExtractor
           field_pointer: ["result"]
     """
-    config = parser.parse(content)
+    config = resolver.preprocess_manifest(YamlDeclarativeSource._parse(content), {}, "")
     factory.create_component(config["error_handler"], input_config, False)
 
 
diff --git a/airbyte-cdk/python/unit_tests/sources/declarative/test_yaml_declarative_source.py b/airbyte-cdk/python/unit_tests/sources/declarative/test_yaml_declarative_source.py
index 0b3f0c2e45ee30..ba6af18094f4d8 100644
--- a/airbyte-cdk/python/unit_tests/sources/declarative/test_yaml_declarative_source.py
+++ b/airbyte-cdk/python/unit_tests/sources/declarative/test_yaml_declarative_source.py
@@ -8,7 +8,6 @@
 
 import pytest
 from airbyte_cdk.sources.declarative.parsers.undefined_reference_exception import UndefinedReferenceException
-from airbyte_cdk.sources.declarative.parsers.yaml_parser import YamlParser
 from airbyte_cdk.sources.declarative.yaml_declarative_source import YamlDeclarativeSource
 from yaml.parser import ParserError
 
@@ -39,7 +38,7 @@ def _read_and_parse_yaml_file(self, path_to_yaml_file):
         """
         with open(path_to_yaml_file, "r") as f:
             config_content = f.read()
-            parsed_config = YamlParser().parse(config_content)
+            parsed_config = YamlDeclarativeSource._parse(config_content)
             return parsed_config