fix typing, tests

Author: pedroslopez

airbyte_cdk/manifest_server/command_processor/utils.py

@@ -71,22 +71,26 @@ def build_source(
 ) -> ConcurrentDeclarativeSource[Optional[List[AirbyteStateMessage]]]:
     # We enforce a concurrency level of 1 so that the stream is processed on a single thread
     # to retain ordering for the grouping of the builder message responses.
-    if "concurrency_level" in manifest:
-        manifest["concurrency_level"]["default_concurrency"] = 1
+    manifest_no_concurrency = dict(manifest)
+    if "concurrency_level" in manifest_no_concurrency:
+        manifest_no_concurrency["concurrency_level"]["default_concurrency"] = 1
     else:
-        manifest["concurrency_level"] = {"type": "ConcurrencyLevel", "default_concurrency": 1}
+        manifest_no_concurrency["concurrency_level"] = {
+            "type": "ConcurrencyLevel",
+            "default_concurrency": 1,
+        }
 
     return ConcurrentDeclarativeSource(
         catalog=catalog,
         state=state,
-        source_config=manifest,
+        source_config=manifest_no_concurrency,
         config=config,
         normalize_manifest=should_normalize_manifest(manifest),
         migrate_manifest=should_migrate_manifest(manifest),
         emit_connector_builder_messages=True,
         limits=TestLimits(
-            max_pages_per_slice=page_limit,
-            max_slices=slice_limit,
-            max_records=record_limit,
+            max_pages_per_slice=page_limit or TestLimits.DEFAULT_MAX_PAGES_PER_SLICE,
+            max_slices=slice_limit or TestLimits.DEFAULT_MAX_SLICES,
+            max_records=record_limit or TestLimits.DEFAULT_MAX_RECORDS,
         ),
     )

unit_tests/manifest_server/command_processor/test_utils.py

@@ -32,7 +32,7 @@ def test_build_catalog_creates_correct_structure(self):
         assert configured_stream.destination_sync_mode == DestinationSyncMode.overwrite
 
     @patch("airbyte_cdk.manifest_server.command_processor.utils.ConcurrentDeclarativeSource")
-    def test_build_source_creates_manifest_declarative_source(self, mock_source_class):
+    def test_build_source_creates_source(self, mock_source_class):
         """Test that build_source creates a ConcurrentDeclarativeSource with correct parameters."""
         # Setup mocks
         mock_source = Mock()
@@ -69,10 +69,14 @@ def test_build_source_creates_manifest_declarative_source(self, mock_source_clas
         result = build_source(manifest, catalog, config, state)
 
         # Verify ConcurrentDeclarativeSource was created with correct parameters
+        expected_source_config = {
+            **manifest,
+            "concurrency_level": {"type": "ConcurrencyLevel", "default_concurrency": 1},
+        }
         mock_source_class.assert_called_once_with(
             catalog=catalog,
             state=state,
-            source_config=manifest,
+            source_config=expected_source_config,
             config=config,
             normalize_manifest=False,  # Default when flag not set
             migrate_manifest=False,  # Default when flag not set