make SimpleRetrieverTestReadDecorator when no read enabled mode required

Author: aldogonzalez8

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -3299,24 +3299,10 @@ def _get_url() -> str:
                 parameters=model.parameters or {},
             )
 
-        if self._limit_slices_fetched or self._emit_connector_builder_messages:
-            return SimpleRetrieverTestReadDecorator(
-                name=name,
-                paginator=paginator,
-                primary_key=primary_key,
-                requester=requester,
-                record_selector=record_selector,
-                stream_slicer=stream_slicer,
-                request_option_provider=request_options_provider,
-                cursor=cursor,
-                config=config,
-                maximum_number_of_slices=self._limit_slices_fetched or 5,
-                ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests,
-                additional_query_properties=query_properties,
-                log_formatter=log_formatter,
-                parameters=model.parameters or {},
-            )
-        return SimpleRetriever(
+        test_read_enabled = self._limit_slices_fetched or self._emit_connector_builder_messages
+        maximum_number_of_slices = self._limit_slices_fetched or 5 if test_read_enabled else 0
+        retriever_log_formatter = log_formatter if test_read_enabled else None
+        return SimpleRetrieverTestReadDecorator(
             name=name,
             paginator=paginator,
             primary_key=primary_key,
@@ -3326,8 +3312,11 @@ def _get_url() -> str:
             request_option_provider=request_options_provider,
             cursor=cursor,
             config=config,
+            maximum_number_of_slices=maximum_number_of_slices,
+            emit_connector_builder_messages=self._emit_connector_builder_messages,
             ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests,
             additional_query_properties=query_properties,
+            log_formatter=retriever_log_formatter,
             parameters=model.parameters or {},
         )
 
@@ -3441,28 +3430,19 @@ def _get_download_retriever() -> SimpleRetrieverTestReadDecorator | SimpleRetrie
                 if model.download_paginator
                 else NoPagination(parameters={})
             )
-            maximum_number_of_slices = self._limit_slices_fetched or 5
+            test_read_enabled = self._limit_slices_fetched or self._emit_connector_builder_messages
+            maximum_number_of_slices = self._limit_slices_fetched or 5 if test_read_enabled else 0
 
-            if self._limit_slices_fetched or self._emit_connector_builder_messages:
-                return SimpleRetrieverTestReadDecorator(
-                    requester=download_requester,
-                    record_selector=record_selector,
-                    primary_key=None,
-                    name=job_download_components_name,
-                    paginator=paginator,
-                    config=config,
-                    parameters={},
-                    maximum_number_of_slices=maximum_number_of_slices,
-                )
-
-            return SimpleRetriever(
+            return SimpleRetrieverTestReadDecorator(
                 requester=download_requester,
                 record_selector=record_selector,
                 primary_key=None,
                 name=job_download_components_name,
                 paginator=paginator,
                 config=config,
                 parameters={},
+                maximum_number_of_slices=maximum_number_of_slices,
+                emit_connector_builder_messages=self._emit_connector_builder_messages,
             )
 
         def _get_job_timeout() -> datetime.timedelta:

airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -650,33 +650,47 @@ class SimpleRetrieverTestReadDecorator(SimpleRetriever):
     """
     In some cases, we want to limit the number of requests that are made to the backend source. This class allows for limiting the number of
     slices that are queried throughout a read command.
+
+    maximum_number_of_slices must be provided when test read is enabled.
     """
 
-    maximum_number_of_slices: int = 5
-
-    def __post_init__(self, options: Mapping[str, Any]) -> None:
-        super().__post_init__(options)
-        self.log_formatter = (
-            (
-                lambda response: format_http_message(
-                    response,
-                    f"Stream '{self.name}' request",
-                    f"Request performed in order to extract records for stream '{self.name}'",
-                    self.name,
+    maximum_number_of_slices: int = 0
+    emit_connector_builder_messages: bool = False
+
+    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
+        super().__post_init__(parameters)
+        if self.test_read_enabled():
+            self.log_formatter = (
+                (
+                    lambda response: format_http_message(
+                        response,
+                        f"Stream '{self.name}' request",
+                        f"Request performed in order to extract records for stream '{self.name}'",
+                        self.name,
+                    )
                 )
+                if not self.log_formatter
+                else self.log_formatter
             )
-            if not self.log_formatter
-            else self.log_formatter
-        )
 
-        if self.maximum_number_of_slices and self.maximum_number_of_slices < 1:
-            raise ValueError(
-                f"The maximum number of slices on a test read needs to be strictly positive. Got {self.maximum_number_of_slices}"
-            )
+            if self.maximum_number_of_slices and self.maximum_number_of_slices < 1:
+                raise ValueError(
+                    f"The maximum number of slices on a test read needs to be strictly positive. Got {self.maximum_number_of_slices}"
+                )
+
+    def test_read_enabled(self) -> bool:
+        """
+        Indicates whether the retriever is in test read mode.
+        This is used to limit the number of slices processed during a test read.
+        """
+        return bool(self.maximum_number_of_slices or self.emit_connector_builder_messages)
 
     # stream_slices is defined with arguments on http stream and fixing this has a long tail of dependencies. Will be resolved by the decoupling of http stream and simple retriever
     def stream_slices(self) -> Iterable[Optional[StreamSlice]]:  # type: ignore
-        return islice(super().stream_slices(), self.maximum_number_of_slices)
+        if not self.test_read_enabled():
+            return super().stream_slices()
+        else:
+            return islice(super().stream_slices(), self.maximum_number_of_slices)
 
 
 @deprecated(

unit_tests/sources/declarative/retrievers/test_simple_retriever.py

@@ -915,6 +915,7 @@ def test_emit_log_request_response_messages(mocker):
         stream_slicer=SinglePartitionRouter(parameters={}),
         parameters={},
         config={},
+        maximum_number_of_slices=1,
     )
 
     retriever._fetch_next_page(