fix(manifest-server): reset global record counter before test reads

Author: pedroslopez

airbyte_cdk/.DS_Store

@@ -7,6 +7,7 @@
 )
 from fastapi import HTTPException
 
+import airbyte_cdk.sources.declarative.stream_slicers.declarative_partition_generator
 from airbyte_cdk.connector_builder.models import StreamRead
 from airbyte_cdk.connector_builder.test_reader import TestReader
 from airbyte_cdk.entrypoint import AirbyteEntrypoint
@@ -42,6 +43,12 @@ def test_read(
         Test the read method of the source.
         """
 
+        # HACK: reset total_record_counter
+        # DeclarativePartition defines total_record_counter as a global variable, which keeps around the record count
+        # across multiple test_read calls, even if the source is different. This is a hack to reset the counter for
+        # each test_read call.
+        airbyte_cdk.sources.declarative.stream_slicers.declarative_partition_generator.total_record_counter = 0
+
         test_read_handler = TestReader(
             max_pages_per_slice=page_limit,
             max_slices=slice_limit,

airbyte_cdk/manifest_server/command_processor/processor.py

@@ -308,3 +308,57 @@ def test_discover_with_trace_error(self, command_processor, sample_config):
             # Verify exception is raised
             with pytest.raises(HTTPException):
                 command_processor.discover(sample_config)
+
+    def test_test_read_resets_global_record_counter(
+        self, command_processor, sample_config, sample_catalog
+    ):
+        """Test that test_read resets the global total_record_counter between calls."""
+        from airbyte_cdk.sources.declarative import stream_slicers
+
+        # Mock the TestReader
+        with patch(
+            "airbyte_cdk.manifest_server.command_processor.processor.TestReader"
+        ) as mock_test_reader_class:
+            mock_test_reader_instance = Mock()
+            mock_test_reader_class.return_value = mock_test_reader_instance
+            mock_stream_read = Mock()
+            mock_test_reader_instance.run_test_read.return_value = mock_stream_read
+
+            # Set initial counter value to simulate previous test_read execution
+            stream_slicers.declarative_partition_generator.total_record_counter = 100
+
+            # Execute test_read
+            result1 = command_processor.test_read(
+                config=sample_config,
+                catalog=sample_catalog,
+                state=[],
+                record_limit=50,
+                page_limit=3,
+                slice_limit=7,
+            )
+
+            # Verify counter was reset to 0 before the test_read
+            assert stream_slicers.declarative_partition_generator.total_record_counter == 0
+
+            # Set counter again to simulate state from first call
+            stream_slicers.declarative_partition_generator.total_record_counter = 200
+
+            # Execute another test_read
+            result2 = command_processor.test_read(
+                config=sample_config,
+                catalog=sample_catalog,
+                state=[],
+                record_limit=25,
+                page_limit=2,
+                slice_limit=5,
+            )
+
+            # Verify counter was reset again
+            assert stream_slicers.declarative_partition_generator.total_record_counter == 0
+
+            # Verify both calls returned the expected results
+            assert result1 == mock_stream_read
+            assert result2 == mock_stream_read
+
+            # Verify TestReader was called twice
+            assert mock_test_reader_class.call_count == 2