add testing for factory, simple retriever, request options provider and fix bug emitting as Record instead of mapping

Author: brianjlai

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2856,7 +2856,7 @@ def create_simple_retriever(
 
             if len(query_properties_definitions) > 1:
                 raise ValueError(
-                    f"request_parameters should only define one QueryProperties field, but found {len(query_properties_definitions)}"
+                    f"request_parameters only supports defining one QueryProperties field, but found {len(query_properties_definitions)} usages"
                 )
 
             if len(query_properties_definitions) == 1:
@@ -3000,27 +3000,6 @@ def _remove_query_properties(
             or not request_parameter.get("type") == "QueryProperties"
         }
 
-    @staticmethod
-    def _translate_query_properties_to_interpolated_strings(
-        request_parameters: Mapping[str, Union[Any, str]],
-    ) -> Mapping[str, Union[Any, str]]:
-        # todo blai: remove this since unused
-        new_request_parameters = dict()
-        for key, request_parameter in request_parameters.items():
-            if (
-                isinstance(request_parameter, Mapping)
-                and request_parameter.get("type") == "QueryProperties"
-            ):
-                # This may seem like this could be combined into the above conditional, but this is separated
-                # so that we do not add the properties into the new request_parameters mapping
-                if request_parameter.get("inject_into"):
-                    new_request_parameters[key] = (
-                        "{{ stream_partition.extra_fields['query_properties'] }}"
-                    )
-            else:
-                new_request_parameters[key] = request_parameter
-        return new_request_parameters
-
     def create_state_delegating_stream(
         self,
         model: StateDelegatingStreamModel,

airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -515,7 +515,10 @@ def read_records(
                 self.cursor.close_slice(_slice, most_recent_record_from_slice)
 
             if has_multiple_chunks:
-                yield from merged_records.values()
+                yield from [
+                    Record(data=merged_record, stream_name=self.name, associated_slice=stream_slice)
+                    for merged_record in merged_records.values()
+                ]
         else:
             _slice = stream_slice or StreamSlice(partition={}, cursor_slice={})  # None-check