fix: Add comprehensive integration tests and fix schema inference issues

- Fix stream_name propagation in factory to use stream context name
- Add type conversion for Mapping-like wrapper objects to plain dicts
- Add 5 comprehensive integration tests using HttpMocker and JSONPlaceholder API
- Update test assertions to handle genson's nullable type format
- Add proper error handling for retriever failures
- Update all unit test mocks to include stream_slices() method

Co-Authored-By: AJ Steers <[email protected]>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2534,7 +2534,7 @@ def create_inferred_schema_loader(
             retriever=retriever,
             config=config,
             record_sample_size=model.record_sample_size or 100,
-            stream_name=model.stream_name or "",
+            stream_name=model.stream_name or name,
             parameters=model.parameters or {},
         )
 

airbyte_cdk/sources/declarative/schema/inferred_schema_loader.py

@@ -2,6 +2,7 @@
 # Copyright (c) 2025 Airbyte, Inc., all rights reserved.
 #
 
+from collections.abc import Mapping as ABCMapping
 from dataclasses import InitVar, dataclass
 from typing import Any, Mapping, Optional
 
@@ -55,18 +56,27 @@ def get_json_schema(self) -> Mapping[str, Any]:
 
         record_count = 0
         try:
-            for record in self.retriever.read_records({}):  # type: ignore[call-overload]
-                if record_count >= self.record_sample_size:
-                    break
+            for stream_slice in self.retriever.stream_slices():
+                for record in self.retriever.read_records(
+                    records_schema={}, stream_slice=stream_slice
+                ):
+                    if record_count >= self.record_sample_size:
+                        break
+
+                    if isinstance(record, ABCMapping) and not isinstance(record, dict):
+                        record = dict(record)
 
-                airbyte_record = AirbyteRecordMessage(
-                    stream=self.stream_name,
-                    data=record,  # type: ignore[arg-type]
-                    emitted_at=0,  # Not used for schema inference
-                )
+                    airbyte_record = AirbyteRecordMessage(
+                        stream=self.stream_name,
+                        data=record,  # type: ignore[arg-type]
+                        emitted_at=0,
+                    )
 
-                schema_inferrer.accumulate(airbyte_record)
-                record_count += 1
+                    schema_inferrer.accumulate(airbyte_record)
+                    record_count += 1
+
+                if record_count >= self.record_sample_size:
+                    break
         except Exception:
             return {}
 

unit_tests/sources/declarative/schema/test_inferred_schema_loader.py

@@ -80,6 +80,7 @@
 def mock_retriever():
     """Create a mock retriever that returns sample records."""
     retriever = MagicMock()
+    retriever.stream_slices.return_value = iter([None])
     retriever.read_records.return_value = iter(
         [
             {"id": 1, "name": "Alice", "age": 30, "active": True},
@@ -126,6 +127,7 @@ def test_inferred_schema_loader_basic(inferred_schema_loader):
 def test_inferred_schema_loader_empty_records():
     """Test that InferredSchemaLoader returns empty schema when no records are available."""
     retriever = MagicMock()
+    retriever.stream_slices.return_value = iter([None])
     retriever.read_records.return_value = iter([])
 
     config = MagicMock()
@@ -147,6 +149,7 @@ def test_inferred_schema_loader_respects_sample_size():
     """Test that InferredSchemaLoader respects the record_sample_size parameter."""
     retriever = MagicMock()
     records = [{"id": i, "name": f"User{i}"} for i in range(10)]
+    retriever.stream_slices.return_value = iter([None])
     retriever.read_records.return_value = iter(records)
 
     config = MagicMock()
@@ -169,6 +172,7 @@ def test_inferred_schema_loader_respects_sample_size():
 def test_inferred_schema_loader_handles_errors():
     """Test that InferredSchemaLoader handles errors gracefully."""
     retriever = MagicMock()
+    retriever.stream_slices.return_value = iter([None])
     retriever.read_records.side_effect = Exception("API Error")
 
     config = MagicMock()
@@ -189,6 +193,7 @@ def test_inferred_schema_loader_handles_errors():
 def test_inferred_schema_loader_with_nested_objects():
     """Test that InferredSchemaLoader handles nested objects correctly."""
     retriever = MagicMock()
+    retriever.stream_slices.return_value = iter([None])
     retriever.read_records.return_value = iter(
         [
             {
@@ -224,6 +229,7 @@ def test_inferred_schema_loader_with_nested_objects():
 def test_inferred_schema_loader_with_arrays():
     """Test that InferredSchemaLoader handles arrays correctly."""
     retriever = MagicMock()
+    retriever.stream_slices.return_value = iter([None])
     retriever.read_records.return_value = iter(
         [
             {"id": 1, "name": "Alice", "tags": ["admin", "user"]},