selective import from 'origin/devin/1752808596-manual-tests-static-schema-discover'

Author: aaronsteers

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -617,7 +617,8 @@ def _uses_dynamic_schema_loader(self) -> bool:
         Returns:
             bool: True if any stream uses a DynamicSchemaLoader, False otherwise.
         """
-        for stream_config in self._stream_configs(self._source_config):
+        empty_config: Dict[str, Any] = {}
+        for stream_config in self._stream_configs(self._source_config, empty_config):
             schema_loader = stream_config.get("schema_loader", {})
             if (
                 isinstance(schema_loader, dict)

airbyte_cdk/test/standard_tests/connector_base.py

@@ -136,6 +136,7 @@ def test_check(
             and not result.errors
         ):
             raise AssertionError(f"Expected error in `check` but got success.")
+
     @classmethod
     def get_connector_root_dir(cls) -> Path:
         """Get the root directory of the connector."""

airbyte_cdk/test/standard_tests/source_base.py

@@ -159,7 +159,7 @@ def test_fail_read_with_bad_catalog(
         # Recreate the scenario with the same config but set the status to "failed".
         scenario = ConnectorTestScenario(
             config_dict=scenario.get_config_dict(
-                connector_root=scenario.connector_root,
+                connector_root=self.get_connector_root_dir(),
                 empty_if_missing=False,
             ),
             status="failed",