add tests for discover catalog output and creating DefaultStream in model to component factory

Author: brianjlai

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -5030,6 +5030,190 @@ def test_create_stream_with_multiple_schema_loaders():
     assert isinstance(schema_loader.schema_loaders[1], InlineSchemaLoader)
 
 
+@pytest.mark.parametrize(
+    "allow_catalog_defined_cursor_field,catalog_cursor_field,expected_cursor_field",
+    [
+        pytest.param(
+            True,
+            "custom_cursor_field",
+            "custom_cursor_field",
+            id="test_catalog_defined_cursor_field",
+        ),
+        pytest.param(
+            True,
+            None,
+            "updated_at",
+            id="test_no_catalog_cursor_field_defaults_to_stream_defined_cursor_field",
+        ),
+        pytest.param(
+            False,
+            "custom_cursor_field",
+            "updated_at",
+            id="test_allow_catalog_defined_cursor_field_false_defaults_to_stream_defined_cursor_field",
+        ),
+    ],
+)
+def test_catalog_defined_cursor_field(
+    allow_catalog_defined_cursor_field, catalog_cursor_field, expected_cursor_field
+):
+    content = """
+selector:
+  type: RecordSelector
+  extractor:
+      type: DpathExtractor
+      field_path: ["extractor_path"]
+requester:
+  type: HttpRequester
+  name: "{{ parameters['name'] }}"
+  url_base: "https://api.sendgrid.com/v3/"
+  http_method: "GET"
+list_stream:
+  type: DeclarativeStream
+  incremental_sync:
+    type: DatetimeBasedCursor
+    $parameters:
+      datetime_format: "%Y-%m-%dT%H:%M:%S.%f%z"
+    start_datetime:
+      type: MinMaxDatetime
+      datetime: "{{ config.get('start_date', '1970-01-01T00:00:00.0Z') }}"
+      datetime_format: "%Y-%m-%dT%H:%M:%S.%fZ"
+    cursor_field: "updated_at"
+    allow_catalog_defined_cursor_field: true
+  retriever:
+    type: SimpleRetriever
+    name: "{{ parameters['name'] }}"
+    paginator:
+      type: DefaultPaginator
+      pagination_strategy:
+        type: "CursorPagination"
+        cursor_value: "{{ response._metadata.next }}"
+        page_size: 10
+    requester:
+      $ref: "#/requester"
+      path: "/"
+    record_selector:
+      $ref: "#/selector"
+  $parameters:
+    name: "lists"
+    """
+
+    configured_catalog = ConfiguredAirbyteCatalog(
+        streams=[
+            ConfiguredAirbyteStream(
+                stream=AirbyteStream(
+                    name="lists",
+                    json_schema={"type": "object", "properties": {"id": {"type": "string"}}},
+                    supported_sync_modes=[SyncMode.incremental, SyncMode.full_refresh],
+                ),
+                sync_mode=SyncMode.incremental,
+                destination_sync_mode=DestinationSyncMode.overwrite,
+                cursor_field=[catalog_cursor_field] if catalog_cursor_field else None,
+            )
+        ]
+    )
+
+    model_to_component_factory = ModelToComponentFactory(
+        configured_catalog=configured_catalog,
+    )
+
+    parsed_manifest = YamlDeclarativeSource._parse(content)
+    resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
+    stream_manifest = transformer.propagate_types_and_parameters(
+        "", resolved_manifest["list_stream"], {}
+    )
+
+    stream_manifest["incremental_sync"]["allow_catalog_defined_cursor_field"] = (
+        allow_catalog_defined_cursor_field
+    )
+
+    stream: DefaultStream = model_to_component_factory.create_component(
+        model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=input_config
+    )
+
+    assert stream.cursor_field == expected_cursor_field
+    assert stream._cursor_field.cursor_field_key == expected_cursor_field
+    assert (
+        stream._cursor_field.supports_catalog_defined_cursor_field
+        == allow_catalog_defined_cursor_field
+    )
+
+
+def test_catalog_defined_cursor_field_stream_missing():
+    content = """
+selector:
+  type: RecordSelector
+  extractor:
+      type: DpathExtractor
+      field_path: ["extractor_path"]
+requester:
+  type: HttpRequester
+  name: "{{ parameters['name'] }}"
+  url_base: "https://api.sendgrid.com/v3/"
+  http_method: "GET"
+list_stream:
+  type: DeclarativeStream
+  incremental_sync:
+    type: DatetimeBasedCursor
+    $parameters:
+      datetime_format: "%Y-%m-%dT%H:%M:%S.%f%z"
+    start_datetime:
+      type: MinMaxDatetime
+      datetime: "{{ config.get('start_date', '1970-01-01T00:00:00.0Z') }}"
+      datetime_format: "%Y-%m-%dT%H:%M:%S.%fZ"
+    cursor_field: "updated_at"
+    allow_catalog_defined_cursor_field: true
+  retriever:
+    type: SimpleRetriever
+    name: "{{ parameters['name'] }}"
+    paginator:
+      type: DefaultPaginator
+      pagination_strategy:
+        type: "CursorPagination"
+        cursor_value: "{{ response._metadata.next }}"
+        page_size: 10
+    requester:
+      $ref: "#/requester"
+      path: "/"
+    record_selector:
+      $ref: "#/selector"
+  $parameters:
+    name: "lists"
+    """
+
+    configured_catalog = ConfiguredAirbyteCatalog(
+        streams=[
+            ConfiguredAirbyteStream(
+                stream=AirbyteStream(
+                    name="other_stream",
+                    json_schema={"type": "object", "properties": {"id": {"type": "string"}}},
+                    supported_sync_modes=[SyncMode.incremental, SyncMode.full_refresh],
+                ),
+                sync_mode=SyncMode.incremental,
+                destination_sync_mode=DestinationSyncMode.overwrite,
+                cursor_field=["custom_cursor_field"],
+            )
+        ]
+    )
+
+    model_to_component_factory = ModelToComponentFactory(
+        configured_catalog=configured_catalog,
+    )
+
+    parsed_manifest = YamlDeclarativeSource._parse(content)
+    resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
+    stream_manifest = transformer.propagate_types_and_parameters(
+        "", resolved_manifest["list_stream"], {}
+    )
+
+    stream: DefaultStream = model_to_component_factory.create_component(
+        model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=input_config
+    )
+
+    assert stream.cursor_field == "updated_at"
+    assert stream._cursor_field.cursor_field_key == "updated_at"
+    assert stream._cursor_field.supports_catalog_defined_cursor_field == True
+
+
 def get_schema_loader(stream: DefaultStream):
     assert isinstance(
         stream._stream_partition_generator._partition_factory._schema_loader,

unit_tests/sources/streams/concurrent/test_adapters.py

@@ -22,7 +22,7 @@
     StreamPartition,
     StreamPartitionGenerator,
 )
-from airbyte_cdk.sources.streams.concurrent.cursor import Cursor, CursorField
+from airbyte_cdk.sources.streams.concurrent.cursor import Cursor
 from airbyte_cdk.sources.streams.concurrent.exceptions import ExceptionWithDisplayMessage
 from airbyte_cdk.sources.streams.core import Stream
 from airbyte_cdk.sources.types import Record
@@ -328,7 +328,6 @@ def test_create_from_stream_stream(self):
 
         assert facade.name == "stream"
         assert facade.cursor_field == "cursor"
-        # assert facade.cursor_field == CursorField(cursor_field_key="cursor")
         assert facade._abstract_stream._primary_key == ["id"]
 
     def test_create_from_stream_stream_with_none_primary_key(self):

unit_tests/sources/streams/concurrent/test_default_stream.py

@@ -272,6 +272,43 @@ def test_as_airbyte_stream_with_file_transfer_support(self):
 
         assert actual_airbyte_stream == expected_airbyte_stream
 
+    def test_as_airbyte_stream_with_a_catalog_defined_cursor(self):
+        json_schema = {
+            "type": "object",
+            "properties": {
+                "id": {"type": ["null", "string"]},
+                "date": {"type": ["null", "string"]},
+            },
+        }
+        stream = DefaultStream(
+            self._partition_generator,
+            self._name,
+            json_schema,
+            self._primary_key,
+            CursorField(cursor_field_key="date", supports_catalog_defined_cursor_field=True),
+            self._logger,
+            FinalStateCursor(
+                stream_name=self._name,
+                stream_namespace=None,
+                message_repository=self._message_repository,
+            ),
+        )
+
+        expected_airbyte_stream = AirbyteStream(
+            name=self._name,
+            json_schema=json_schema,
+            supported_sync_modes=[SyncMode.full_refresh, SyncMode.incremental],
+            source_defined_cursor=False,
+            default_cursor_field=["date"],
+            source_defined_primary_key=None,
+            namespace=None,
+            is_resumable=True,
+            is_file_based=False,
+        )
+
+        airbyte_stream = stream.as_airbyte_stream()
+        assert airbyte_stream == expected_airbyte_stream
+
     def test_given_no_partitions_when_get_availability_then_unavailable(self) -> None:
         self._partition_generator.generate.return_value = []