Add unit tests for error scenarios

Author: tolik0

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -3064,6 +3064,13 @@ def create_grouping_partition_router(
                 f"Underlying partition router must be a PartitionRouter subclass, got {type(underlying_router)}"
             )
 
+        if isinstance(underlying_router, SubstreamPartitionRouter):
+            if any(
+                parent_config.request_option
+                for parent_config in underlying_router.parent_stream_configs
+            ):
+                raise ValueError("Request options are not supported for GroupingPartitionRouter.")
+
         return GroupingPartitionRouter(
             group_size=model.group_size,
             underlying_partition_router=underlying_router,

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -3900,3 +3900,100 @@ def test_create_grouping_partition_router_with_underlying_router():
     assert isinstance(parent_stream_configs[0].stream, DeclarativeStream)
     assert parent_stream_configs[0].parent_key.eval({}) == "id"
     assert parent_stream_configs[0].partition_field.eval({}) == "repository_id"
+
+
+def test_create_grouping_partition_router_invalid_group_size():
+    """Test that an invalid group_size (< 1) raises a ValueError."""
+    content = """
+    schema_loader:
+      file_path: "./source_example/schemas/{{ parameters['name'] }}.yaml"
+      name: "{{ parameters['stream_name'] }}"
+    retriever:
+      requester:
+        type: "HttpRequester"
+        path: "example"
+      record_selector:
+        extractor:
+          field_path: []
+    stream_A:
+      type: DeclarativeStream
+      name: "A"
+      primary_key: "id"
+      $parameters:
+        retriever: "#/retriever"
+        url_base: "https://airbyte.io"
+        schema_loader: "#/schema_loader"
+    sub_partition_router:
+      type: SubstreamPartitionRouter
+      parent_stream_configs:
+        - stream: "#/stream_A"
+          parent_key: id
+          partition_field: repository_id
+    partition_router:
+      type: GroupingPartitionRouter
+      underlying_partition_router: "#/sub_partition_router"
+      group_size: 0
+    """
+    parsed_manifest = YamlDeclarativeSource._parse(content)
+    resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
+    partition_router_manifest = transformer.propagate_types_and_parameters(
+        "", resolved_manifest["partition_router"], {}
+    )
+
+    with pytest.raises(ValueError, match="Group size must be greater than 0, got 0"):
+        factory.create_component(
+            model_type=GroupingPartitionRouterModel,
+            component_definition=partition_router_manifest,
+            config=input_config,
+        )
+
+
+def test_create_grouping_partition_router_substream_with_request_option():
+    """Test that a SubstreamPartitionRouter with request_option raises a ValueError."""
+    content = """
+    schema_loader:
+      file_path: "./source_example/schemas/{{ parameters['name'] }}.yaml"
+      name: "{{ parameters['stream_name'] }}"
+    retriever:
+      requester:
+        type: "HttpRequester"
+        path: "example"
+      record_selector:
+        extractor:
+          field_path: []
+    stream_A:
+      type: DeclarativeStream
+      name: "A"
+      primary_key: "id"
+      $parameters:
+        retriever: "#/retriever"
+        url_base: "https://airbyte.io"
+        schema_loader: "#/schema_loader"
+    sub_partition_router:
+      type: SubstreamPartitionRouter
+      parent_stream_configs:
+        - stream: "#/stream_A"
+          parent_key: id
+          partition_field: repository_id
+          request_option:
+            inject_into: request_parameter
+            field_name: "repo_id"
+    partition_router:
+      type: GroupingPartitionRouter
+      underlying_partition_router: "#/sub_partition_router"
+      group_size: 2
+    """
+    parsed_manifest = YamlDeclarativeSource._parse(content)
+    resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
+    partition_router_manifest = transformer.propagate_types_and_parameters(
+        "", resolved_manifest["partition_router"], {}
+    )
+
+    with pytest.raises(
+        ValueError, match="Request options are not supported for GroupingPartitionRouter."
+    ):
+        factory.create_component(
+            model_type=GroupingPartitionRouterModel,
+            component_definition=partition_router_manifest,
+            config=input_config,
+        )