Merge branch 'main' into 50395-2

* main:
  fix: update cryptography package to latest version to address CVE (airbytehq#377)
  fix: (CDK) (HttpRequester) - Make the `HttpRequester.path` optional (airbytehq#370)
  feat: improved custom components handling (airbytehq#350)
  feat: add microseconds timestamp format (airbytehq#373)
  fix: Replace Unidecode with anyascii for permissive license (airbytehq#367)
  feat: add IncrementingCountCursor (airbytehq#346)
  feat: (low-code cdk)  datetime format with milliseconds (airbytehq#369)
  fix: (CDK) (AsyncRetriever) - Improve UX on variable naming and interpolation (airbytehq#368)
  fix: (CDK) (AsyncRetriever) - Add the `request` and `response` to each `async` operations (airbytehq#356)
  fix: (CDK) (ConnectorBuilder) - Add `auxiliary requests` to slice; support `TestRead` for AsyncRetriever (part 1/2) (airbytehq#355)
  feat(concurrent perpartition cursor): Add parent state updates (airbytehq#343)
  fix: update csv parser for builder compatibility (airbytehq#364)
  feat(low-code cdk): add interpolation for limit field in Rate (airbytehq#353)
  feat(low-code cdk): add AbstractStreamFacade processing as concurrent streams in declarative source (airbytehq#347)
  fix: (CDK) (CsvParser) - Fix the `\\` escaping when passing the `delimiter` from Builder's UI (airbytehq#358)
  feat: expose `str_to_datetime` jinja macro (airbytehq#351)
  fix: update CDK migration for 6.34.0 (airbytehq#348)
  feat: Removes `stream_state` interpolation from CDK (airbytehq#320)
  fix(declarative): Pass `extra_fields` in `global_substream_cursor` (airbytehq#195)
  feat(concurrent perpartition cursor): Refactor ConcurrentPerPartitionCursor (airbytehq#331)
  feat(HttpMocker): adding support for PUT requests and bytes responses (airbytehq#342)
  chore: use certified source for manifest-only test (airbytehq#338)
  feat: check for request_option mapping conflicts in individual components (airbytehq#328)
  feat(file-based): sync file acl permissions and identities (airbytehq#260)
  fix: (CDK) (Connector Builder) - refactor the `MessageGrouper` > `TestRead` (airbytehq#332)
  fix(low code): Fix missing cursor for ClientSideIncrementalRecordFilterDecorator (airbytehq#334)
  feat(low-code): Add API Budget (airbytehq#314)
  chore(decoder): clean decoders and make csvdecoder available (airbytehq#326)

Author: rpopov

Diff for: .github/workflows/connector-tests.yml

@@ -88,7 +88,7 @@ jobs:
           #   cdk_extra: n/a
           # TODO: These are manifest connectors and won't work as expected until we
           #       add `--use-local-cdk` support for manifest connectors.
-          - connector: source-the-guardian-api
+          - connector: source-amplitude
             cdk_extra: n/a
           - connector: source-pokeapi
             cdk_extra: n/a

Diff for: airbyte_cdk/connector_builder/connector_builder_handler.py

@@ -2,10 +2,11 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
-import dataclasses
+
+from dataclasses import asdict, dataclass, field
 from typing import Any, List, Mapping
 
-from airbyte_cdk.connector_builder.message_grouper import MessageGrouper
+from airbyte_cdk.connector_builder.test_reader import TestReader
 from airbyte_cdk.models import (
     AirbyteMessage,
     AirbyteRecordMessage,
@@ -32,11 +33,11 @@
 MAX_RECORDS_KEY = "max_records"
 
 
-@dataclasses.dataclass
+@dataclass
 class TestReadLimits:
-    max_records: int = dataclasses.field(default=DEFAULT_MAXIMUM_RECORDS)
-    max_pages_per_slice: int = dataclasses.field(default=DEFAULT_MAXIMUM_NUMBER_OF_PAGES_PER_SLICE)
-    max_slices: int = dataclasses.field(default=DEFAULT_MAXIMUM_NUMBER_OF_SLICES)
+    max_records: int = field(default=DEFAULT_MAXIMUM_RECORDS)
+    max_pages_per_slice: int = field(default=DEFAULT_MAXIMUM_NUMBER_OF_PAGES_PER_SLICE)
+    max_slices: int = field(default=DEFAULT_MAXIMUM_NUMBER_OF_SLICES)
 
 
 def get_limits(config: Mapping[str, Any]) -> TestReadLimits:
@@ -73,17 +74,20 @@ def read_stream(
     limits: TestReadLimits,
 ) -> AirbyteMessage:
     try:
-        handler = MessageGrouper(limits.max_pages_per_slice, limits.max_slices, limits.max_records)
-        stream_name = configured_catalog.streams[
-            0
-        ].stream.name  # The connector builder only supports a single stream
-        stream_read = handler.get_message_groups(
+        test_read_handler = TestReader(
+            limits.max_pages_per_slice, limits.max_slices, limits.max_records
+        )
+        # The connector builder only supports a single stream
+        stream_name = configured_catalog.streams[0].stream.name
+
+        stream_read = test_read_handler.run_test_read(
             source, config, configured_catalog, state, limits.max_records
         )
+
         return AirbyteMessage(
             type=MessageType.RECORD,
             record=AirbyteRecordMessage(
-                data=dataclasses.asdict(stream_read), stream=stream_name, emitted_at=_emitted_at()
+                data=asdict(stream_read), stream=stream_name, emitted_at=_emitted_at()
             ),
         )
     except Exception as exc: