add functional test and add catalogs to source resolution

core/dbt/contracts/graph/manifest.py

@@ -20,6 +20,7 @@
     Union,
 )
 
+from dbt_config.catalog_config import ExternalCatalog
 from typing_extensions import Protocol
 
 import dbt_common.exceptions
@@ -1333,6 +1334,26 @@ def resolve_source(
         current_project: str,
         node_package: str,
     ) -> MaybeParsedSource:
+        if target_source_name in self.catalogs:
+            catalog = ExternalCatalog.model_validate_json(self.catalogs[target_source_name])
+            identifier = f"{target_source_name}.{target_table_name}"
+            catalog_database = catalog.configuration.internal_namespace.database
+            catalog_schema = catalog.configuration.internal_namespace.schema_
+            return SourceDefinition(
+                database=catalog_database,
+                schema=catalog_schema,
+                fqn=[catalog_database, catalog_schema, catalog.name, target_table_name],
+                name=target_table_name,
+                source_description=f"External Catalog source for {target_source_name}.{target_table_name}",
+                source_name=target_source_name,
+                unique_id=identifier,
+                identifier=identifier,
+                package_name="dbt",
+                path="/root/catalogs.yml",
+                loader=catalog.type.value,
+                resource_type=NodeType.Source,
+                original_file_path="/root/catalogs.yml",
+            )
         search_name = f"{target_source_name}.{target_table_name}"
         candidates = _packages_to_search(current_project, node_package)
 

core/dbt/parser/manifest.py

@@ -445,11 +445,13 @@ def load(self) -> Manifest:
             patcher.construct_sources()
             self.manifest.sources = patcher.sources
             self._perf_info.patch_sources_elapsed = time.perf_counter() - start_patch
+
+            # Get catalog.yml and update the manifest
             raw_catalog = load_external_catalog_config(self.root_project)
             if raw_catalog:
                 catalog_config = ExternalCatalogConfig.model_validate(raw_catalog)
                 self.manifest.catalogs = {
-                    c.name: c.model_dump_json() for c in catalog_config.catalogs
+                    c.name: c.model_dump_json(by_alias=True) for c in catalog_config.catalogs
                 }
             # We need to rebuild disabled in order to include disabled sources
             self.manifest.rebuild_disabled_lookup()
@@ -1151,7 +1153,7 @@ def process_metrics(self, config: RuntimeConfig):
     def process_catalog(self, config: RuntimeConfig):
         if config.catalogs:
             for catalog in config.catalogs.catalogs:
-                self.manifest.catalogs[catalog.name] = catalog.model_dump_json()
+                self.manifest.catalogs[catalog.name] = catalog.model_dump_json(by_alias=True)
 
     def process_saved_queries(self, config: RuntimeConfig):
         """Processes SavedQuery nodes to populate their `depends_on`."""

tests/functional/test_external_catalog.py

@@ -3,7 +3,6 @@
 from dbt_config.catalog_config import ExternalCatalog
 
 from dbt.tests.util import run_dbt, write_file
-from tests.fixtures.jaffle_shop import JaffleShopProject
 
 
 @pytest.fixture(scope="class", autouse=True)
@@ -14,26 +13,36 @@ def dbt_catalog_config(project_root):
                 "name": "my_external_catalog",
                 "type": "iceberg",
                 "configuration": {
-                    "table_format": "parquet",
-                    "namespace": "dbt",
+                    "table_format": "iceberg",
+                    "catalog_namespace": "dbt",
+                    "internal_namespace": {
+                        "database": "my_db",
+                        "schema": "my_schema",
+                    },
                     "external_location": "s3://my-bucket/my-path",
                 },
                 "management": {
                     "enabled": True,
                     "create_if_not_exists": False,
                     "alter_if_different": False,
                     "read_only": True,
-                    "refresh": "on_change",
+                    "refresh": "on-start",
                 },
             }
         ],
     }
     write_file(yaml.safe_dump(config), project_root, "catalog.yml")
 
 
-class TestCatalogConfig(JaffleShopProject):
+class TestCatalogConfig:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "model.sql": "select 1 as id from {{ source('my_external_catalog', 'my_table') }}",
+        }
 
     def test_supplying_external_catalog(self, project):
         manifest = run_dbt(["parse"])
         assert manifest.catalogs != {}
+        assert manifest.nodes["model.test.model"].sources == [["my_external_catalog", "my_table"]]
         ExternalCatalog.model_validate_json(manifest.catalogs["my_external_catalog"])