CDK: add test for concurrent sate cursor

Signed-off-by: Artem Inzhyyants <[email protected]>

Author: artem1205

unit_tests/sources/declarative/test_concurrent_declarative_source.py

@@ -4,6 +4,7 @@
 
 import copy
 import json
+import math
 from datetime import datetime, timedelta, timezone
 from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, Union
 from unittest.mock import patch
@@ -43,6 +44,9 @@
 from airbyte_cdk.sources.streams.checkpoint import Cursor
 from airbyte_cdk.sources.streams.concurrent.cursor import ConcurrentCursor
 from airbyte_cdk.sources.streams.concurrent.default_stream import DefaultStream
+from airbyte_cdk.sources.streams.concurrent.state_converters.incrementing_count_stream_state_converter import (
+    IncrementingCountStreamStateConverter,
+)
 from airbyte_cdk.sources.streams.core import StreamData
 from airbyte_cdk.sources.types import Record, StreamSlice
 from airbyte_cdk.test.mock_http import HttpMocker, HttpRequest, HttpResponse
@@ -230,6 +234,16 @@
                 "inject_into": "request_parameter",
             },
         },
+        "incremental_counting_cursor": {
+            "type": "IncrementingCountCursor",
+            "cursor_field": "id",
+            "start_value": 0,
+            "start_time_option": {
+                "type": "RequestOption",
+                "field_name": "since_id",
+                "inject_into": "request_parameter",
+            },
+        },
         "base_stream": {"retriever": {"$ref": "#/definitions/retriever"}},
         "base_incremental_stream": {
             "retriever": {
@@ -238,6 +252,13 @@
             },
             "incremental_sync": {"$ref": "#/definitions/incremental_cursor"},
         },
+        "base_incremental_counting_stream": {
+            "retriever": {
+                "$ref": "#/definitions/retriever",
+                "requester": {"$ref": "#/definitions/requester"},
+            },
+            "incremental_sync": {"$ref": "#/definitions/incremental_counting_cursor"},
+        },
         "party_members_stream": {
             "$ref": "#/definitions/base_incremental_stream",
             "retriever": {
@@ -527,6 +548,35 @@
                 },
             },
         },
+        "incremental_counting_stream": {
+            "$ref": "#/definitions/base_incremental_counting_stream",
+            "retriever": {
+                "$ref": "#/definitions/base_incremental_counting_stream/retriever",
+                "record_selector": {"$ref": "#/definitions/selector"},
+            },
+            "$parameters": {
+                "name": "incremental_counting_stream",
+                "primary_key": "id",
+                "path": "/party_members",
+            },
+            "schema_loader": {
+                "type": "InlineSchemaLoader",
+                "schema": {
+                    "$schema": "https://json-schema.org/draft-07/schema#",
+                    "type": "object",
+                    "properties": {
+                        "id": {
+                            "description": "The identifier",
+                            "type": ["null", "string"],
+                        },
+                        "name": {
+                            "description": "The name of the party member",
+                            "type": ["null", "string"],
+                        },
+                    },
+                },
+            },
+        },
     },
     "streams": [
         "#/definitions/party_members_stream",
@@ -536,6 +586,7 @@
         "#/definitions/arcana_personas_stream",
         "#/definitions/palace_enemies_stream",
         "#/definitions/async_job_stream",
+        "#/definitions/incremental_counting_stream",
     ],
     "check": {"stream_names": ["party_members", "locations"]},
     "concurrency_level": {
@@ -756,6 +807,20 @@ def test_create_concurrent_cursor():
         "state_type": "date-range",
     }
 
+    incremental_counting_stream = concurrent_streams[7]
+    assert isinstance(incremental_counting_stream, DefaultStream)
+    incremental_counting_cursor = incremental_counting_stream.cursor
+
+    assert isinstance(incremental_counting_cursor, ConcurrentCursor)
+    assert isinstance(
+        incremental_counting_cursor._connector_state_converter,
+        IncrementingCountStreamStateConverter,
+    )
+    assert incremental_counting_cursor._stream_name == "incremental_counting_stream"
+    assert incremental_counting_cursor._cursor_field.cursor_field_key == "id"
+    assert incremental_counting_cursor._start == 0
+    assert incremental_counting_cursor._end_provider() == math.inf
+
 
 def test_check():
     """