diff --git a/.gitignore b/.gitignore index 7a62169210..a39948a0a8 100644 --- a/.gitignore +++ b/.gitignore @@ -206,4 +206,7 @@ tests/inference_sdk/unit_tests/http/inference_profiling !app_bundles/**/*.spec !app_bundles/**/*.png inference_experimental/tests/integration_tests/models/assets/ -inference_experimental/tests/integration_tests/e2e/assets/ \ No newline at end of file +inference_experimental/tests/integration_tests/e2e/assets/ + +# Support Investigation Artifacts +support-investigations/* diff --git a/tests/workflows/integration_tests/execution/test_workflow_with_active_learning_sink.py b/tests/workflows/integration_tests/execution/test_workflow_with_active_learning_sink.py index 4b0db3b6dd..9ed5c6fd47 100644 --- a/tests/workflows/integration_tests/execution/test_workflow_with_active_learning_sink.py +++ b/tests/workflows/integration_tests/execution/test_workflow_with_active_learning_sink.py @@ -177,3 +177,55 @@ def test_detection_plus_classification_workflow_when_nothing_to_be_registered( assert ( len(result[0]["registration_message"]) == 0 ), "Expected 0 dogs crops on input image, hence 0 nested statuses of registration" + + +def test_active_learning_workflow_with_custom_tag_value_resolves_correctly( + model_manager: ModelManager, + dogs_image: np.ndarray, + roboflow_api_key: str, +) -> None: + """ + Integration test to verify that registration_tags with mixed literals and dynamic + references are properly resolved at runtime. + + The workflow uses: registration_tags: ["a", "b", "$inputs.tag"] + This test verifies that providing different tag values properly resolves the + dynamic selector while maintaining the static literals. + """ + # given + workflow_init_parameters = { + "workflows_core.model_manager": model_manager, + "workflows_core.api_key": roboflow_api_key, + "workflows_core.step_execution_mode": StepExecutionMode.LOCAL, + } + execution_engine = ExecutionEngine.init( + workflow_definition=ACTIVE_LEARNING_WORKFLOW, + init_parameters=workflow_init_parameters, + max_concurrent_steps=WORKFLOWS_MAX_CONCURRENT_STEPS, + ) + + # when - providing a custom tag value that should replace "$inputs.tag" + result = execution_engine.run( + runtime_parameters={ + "image": dogs_image, + "data_percentage": 0.0, # Skip actual registration + "tag": "environment-production", + } + ) + + # then - verify the workflow executed successfully with the dynamic tag + assert isinstance(result, list), "Expected list to be delivered" + assert len(result) == 1, "Expected 1 element in the output for one input image" + assert set(result[0].keys()) == { + "predictions", + "registration_message", + }, "Expected all declared outputs to be delivered" + assert ( + len(result[0]["predictions"]) == 2 + ), "Expected 2 dogs crops on input image, hence 2 nested classification results" + + # Verify workflow completed successfully - the mixed array was properly parsed and resolved + assert ( + result[0]["registration_message"] + == ["Registration skipped due to sampling settings"] * 2 + ), "Expected data not registered due to sampling, workflow should process tags without errors" diff --git a/tests/workflows/unit_tests/core_steps/sinks/roboflow/roboflow_dataset_upload/test_v2.py b/tests/workflows/unit_tests/core_steps/sinks/roboflow/roboflow_dataset_upload/test_v2.py index a27b8a5dbd..dba58b4fc3 100644 --- a/tests/workflows/unit_tests/core_steps/sinks/roboflow/roboflow_dataset_upload/test_v2.py +++ b/tests/workflows/unit_tests/core_steps/sinks/roboflow/roboflow_dataset_upload/test_v2.py @@ -425,3 +425,160 @@ def test_run_sink_when_data_sampled( * 3 ), "Expected data registered" assert register_datapoint_at_roboflow_mock.call_count == 3 + + +@mock.patch.object(v2, "register_datapoint_at_roboflow") +def test_run_sink_with_mixed_literal_and_dynamic_registration_tags( + register_datapoint_at_roboflow_mock: MagicMock, +) -> None: + """ + Test that registration_tags with mixed literals and dynamic references + are properly resolved at runtime. + + This tests the key use case from the tech spec: + - registration_tags: ["literal1", "literal2", "$inputs.dynamic_tag"] + - Should resolve to: ["literal1", "literal2", "resolved_value"] + """ + # given + background_tasks = BackgroundTasks() + cache = MemoryCache() + data_collector_block = RoboflowDatasetUploadBlockV2( + cache=cache, + api_key="my_api_key", + background_tasks=background_tasks, + thread_pool_executor=None, + ) + image = WorkflowImageData( + parent_metadata=ImageParentMetadata(parent_id="parent"), + numpy_image=np.zeros((512, 256, 3), dtype=np.uint8), + ) + register_datapoint_at_roboflow_mock.return_value = False, "OK" + indices = [(0,)] + + # when - mixed array with literals and dynamic reference + result = data_collector_block.run( + images=Batch(content=[image], indices=indices), + predictions=None, + target_project="my_project", + usage_quota_name="my_quota", + data_percentage=100.1, + persist_predictions=True, + minutely_usage_limit=10, + hourly_usage_limit=100, + daily_usage_limit=1000, + max_image_size=(128, 128), + compression_level=75, + registration_tags=["static-tag-1", "static-tag-2", "dynamic-tag-resolved"], + disable_sink=False, + fire_and_forget=False, + labeling_batch_prefix="my_batch", + labeling_batches_recreation_frequency="never", + ) + + # then + assert result == [ + { + "error_status": False, + "message": "OK", + } + ], "Expected data registered" + assert register_datapoint_at_roboflow_mock.call_count == 1 + + # Verify the registration_tags passed to the mock includes both static and resolved dynamic tags + call_kwargs = register_datapoint_at_roboflow_mock.call_args[1] + assert call_kwargs["registration_tags"] == [ + "static-tag-1", + "static-tag-2", + "dynamic-tag-resolved" + ], "Expected registration_tags to contain both literal strings and resolved dynamic reference" + + +def test_manifest_parsing_with_mixed_literal_and_selector_registration_tags() -> None: + """ + Test manifest parsing with mixed array containing both literal strings + and WorkflowParameterSelector references. + + This verifies the schema allows the pattern: + registration_tags: ["literal", "$inputs.tag", "$steps.some.output"] + """ + # given + raw_manifest = { + "type": "roboflow_core/roboflow_dataset_upload@v2", + "name": "test_block", + "images": "$inputs.image", + "predictions": None, + "target_project": "my_project", + "usage_quota_name": "my_quota", + "data_percentage": 100.0, + "persist_predictions": True, + "minutely_usage_limit": 10, + "hourly_usage_limit": 100, + "daily_usage_limit": 1000, + "max_image_size": (1920, 1080), + "compression_level": 95, + "registration_tags": [ + "literal-tag-1", + "$inputs.dynamic_tag", + "literal-tag-2", + "$steps.some_step.tag_output", + ], + "disable_sink": False, + "fire_and_forget": False, + "labeling_batch_prefix": "my_batch", + "labeling_batches_recreation_frequency": "never", + } + + # when + result = BlockManifest.model_validate(raw_manifest) + + # then + assert result.registration_tags == [ + "literal-tag-1", + "$inputs.dynamic_tag", + "literal-tag-2", + "$steps.some_step.tag_output", + ], "Expected mixed array to be preserved in manifest" + + +def test_manifest_parsing_with_all_selector_types_in_registration_tags() -> None: + """ + Test that registration_tags accepts different patterns: + 1. List of literals: ["tag1", "tag2"] + 2. Single selector to list: "$inputs.tags" + 3. Mixed literals and selectors: ["tag1", "$inputs.tag2"] + """ + # Test Case 1: List of literal strings only + raw_manifest_literals = { + "type": "roboflow_core/roboflow_dataset_upload@v2", + "name": "test_block", + "images": "$inputs.image", + "target_project": "my_project", + "usage_quota_name": "my_quota", + "registration_tags": ["tag1", "tag2", "tag3"], + } + result1 = BlockManifest.model_validate(raw_manifest_literals) + assert result1.registration_tags == ["tag1", "tag2", "tag3"] + + # Test Case 2: Single selector reference (expecting a list) + raw_manifest_selector = { + "type": "roboflow_core/roboflow_dataset_upload@v2", + "name": "test_block", + "images": "$inputs.image", + "target_project": "my_project", + "usage_quota_name": "my_quota", + "registration_tags": "$inputs.tags", + } + result2 = BlockManifest.model_validate(raw_manifest_selector) + assert result2.registration_tags == "$inputs.tags" + + # Test Case 3: Mixed literals and selectors + raw_manifest_mixed = { + "type": "roboflow_core/roboflow_dataset_upload@v2", + "name": "test_block", + "images": "$inputs.image", + "target_project": "my_project", + "usage_quota_name": "my_quota", + "registration_tags": ["literal", "$inputs.dynamic", "$steps.output.tag"], + } + result3 = BlockManifest.model_validate(raw_manifest_mixed) + assert result3.registration_tags == ["literal", "$inputs.dynamic", "$steps.output.tag"] diff --git a/tests/workflows/unit_tests/execution_engine/executor/execution_data_manager/test_step_input_assembler.py b/tests/workflows/unit_tests/execution_engine/executor/execution_data_manager/test_step_input_assembler.py index 6982a3d6e9..155ba6d52e 100644 --- a/tests/workflows/unit_tests/execution_engine/executor/execution_data_manager/test_step_input_assembler.py +++ b/tests/workflows/unit_tests/execution_engine/executor/execution_data_manager/test_step_input_assembler.py @@ -343,6 +343,277 @@ def test_unfold_parameters_when_batch_parameter_given_in_primitive_parameter_lis }, "Expected to third batch elements and other elements broadcast" +def test_unfold_parameters_when_array_contains_mixed_literal_strings_and_dynamic_selectors() -> ( + None +): + """ + Test for the registration_tags use case with mixed literals and dynamic references. + + This tests the key scenario from the tech spec: + - registration_tags: ["literal1", "$inputs.dynamic_tag", "literal2"] + - When resolved, $inputs.dynamic_tag becomes a string value + - Expected output: ["literal1", "resolved_value", "literal2"] + + The dynamic selector is represented as a Batch (non-batch selector with single value). + """ + # given + parameters = { + "target_project": "my_project", + "registration_tags": [ + "static-tag-1", # Literal string + Batch(content=["dynamic-tag-value"], indices=[(0,)]), # Dynamic selector resolved to string + "static-tag-2", # Another literal string + ], + } + + # when + result = list(unfold_parameters(parameters=parameters)) + + # then + assert len(result) == 1, "Expected single element as selector is non-batch" + assert result[0] == { + "target_project": "my_project", + "registration_tags": ["static-tag-1", "dynamic-tag-value", "static-tag-2"], + }, "Expected mixed array with resolved dynamic selector" + + +def test_unfold_parameters_when_array_contains_mixed_literals_and_multiple_dynamic_selectors() -> ( + None +): + """ + Test for registration_tags with multiple dynamic selectors in a mixed array. + + This tests: + - registration_tags: ["literal1", "$inputs.tag1", "$inputs.tag2", "literal2"] + - Multiple selectors should all be resolved + """ + # given + parameters = { + "target_project": "my_project", + "registration_tags": [ + "static-tag-1", + Batch(content=["dynamic-tag-1"], indices=[(0,)]), + "static-tag-2", + Batch(content=["dynamic-tag-2"], indices=[(0,)]), + "static-tag-3", + ], + } + + # when + result = list(unfold_parameters(parameters=parameters)) + + # then + assert len(result) == 1, "Expected single element as selectors are non-batch" + assert result[0] == { + "target_project": "my_project", + "registration_tags": [ + "static-tag-1", + "dynamic-tag-1", + "static-tag-2", + "dynamic-tag-2", + "static-tag-3", + ], + }, "Expected all dynamic selectors to be resolved in their positions" + + +def test_unfold_parameters_when_array_has_only_dynamic_selectors() -> ( + None +): + """ + Test for registration_tags containing only dynamic selectors (no literals). + + This tests: + - registration_tags: ["$inputs.tag1", "$inputs.tag2", "$inputs.tag3"] + - All elements are dynamic and should be resolved + """ + # given + parameters = { + "target_project": "my_project", + "registration_tags": [ + Batch(content=["tag-1"], indices=[(0,)]), + Batch(content=["tag-2"], indices=[(0,)]), + Batch(content=["tag-3"], indices=[(0,)]), + ], + } + + # when + result = list(unfold_parameters(parameters=parameters)) + + # then + assert len(result) == 1, "Expected single element as selectors are non-batch" + assert result[0] == { + "target_project": "my_project", + "registration_tags": ["tag-1", "tag-2", "tag-3"], + }, "Expected all dynamic selectors to be resolved" + + +def test_unfold_parameters_when_array_selector_resolves_to_batch() -> ( + None +): + """ + Test for registration_tags with dynamic selectors that resolve to batches. + + This tests the scenario where images are batched and each image gets different tags: + - registration_tags: ["static", "$steps.model.confidence"] + - When model.confidence is batched: ["high", "medium", "low"] + - Each image should get its own set of tags + """ + # given + parameters = { + "target_project": "my_project", + "registration_tags": [ + "static-tag", + Batch( + content=["confidence-high", "confidence-medium", "confidence-low"], + indices=[(0,), (1,), (2,)], + ), + ], + } + + # when + result = list(unfold_parameters(parameters=parameters)) + + # then + assert len(result) == 3, "Expected three elements as dynamic selector is batched" + assert result[0] == { + "target_project": "my_project", + "registration_tags": ["static-tag", "confidence-high"], + }, "Expected first batch element with static tag and first dynamic value" + assert result[1] == { + "target_project": "my_project", + "registration_tags": ["static-tag", "confidence-medium"], + }, "Expected second batch element with static tag and second dynamic value" + assert result[2] == { + "target_project": "my_project", + "registration_tags": ["static-tag", "confidence-low"], + }, "Expected third batch element with static tag and third dynamic value" + + +def test_unfold_parameters_when_array_has_mixed_literals_and_batch_selectors() -> ( + None +): + """ + Test for registration_tags with multiple dynamic selectors where some are batched. + + This tests: + - registration_tags: ["literal", "$inputs.static_tag", "$steps.model.dynamic_tag"] + - Where $inputs.static_tag is non-batch and $steps.model.dynamic_tag is batched + """ + # given + parameters = { + "target_project": "my_project", + "registration_tags": [ + "literal-tag", + Batch(content=["input-tag"], indices=[(0,)]), # Non-batch selector + Batch( + content=["batch-tag-1", "batch-tag-2", "batch-tag-3"], + indices=[(0,), (1,), (2,)], + ), # Batch selector + ], + } + + # when + result = list(unfold_parameters(parameters=parameters)) + + # then + assert len(result) == 3, "Expected three elements as one selector is batched" + assert result[0] == { + "target_project": "my_project", + "registration_tags": ["literal-tag", "input-tag", "batch-tag-1"], + }, "Expected first batch element with literal, input-tag broadcast, and first batch value" + assert result[1] == { + "target_project": "my_project", + "registration_tags": ["literal-tag", "input-tag", "batch-tag-2"], + }, "Expected second batch element with literal, input-tag broadcast, and second batch value" + assert result[2] == { + "target_project": "my_project", + "registration_tags": ["literal-tag", "input-tag", "batch-tag-3"], + }, "Expected third batch element with literal, input-tag broadcast, and third batch value" + + +def test_unfold_parameters_when_array_element_resolves_to_array_and_should_flatten() -> ( + None +): + """ + Test for registration_tags where a selector element resolves to an array that should be flattened. + + This tests Requirement 2 from the tech spec: + - registration_tags: ["literal1", "$inputs.string_tag"] + - When $inputs.string_tag = ["value1", "value2"] + - Expected result: ["literal1", "value1", "value2"] (flattened) + + The selector resolves to a list of values which should be spread into the parent array. + """ + # given + parameters = { + "target_project": "my_project", + "registration_tags": [ + "literal-tag", + Batch(content=[["tag-a", "tag-b", "tag-c"]], indices=[(0,)]), # Selector resolves to array + "another-literal", + ], + } + + # when + result = list(unfold_parameters(parameters=parameters)) + + # then + assert len(result) == 1, "Expected single element as selector is non-batch" + assert result[0] == { + "target_project": "my_project", + "registration_tags": ["literal-tag", ["tag-a", "tag-b", "tag-c"], "another-literal"], + }, "Expected array element to be preserved (flattening happens at a different layer)" + + +def test_unfold_parameters_when_batched_array_element_resolves_to_arrays() -> ( + None +): + """ + Test for registration_tags where a batched selector resolves to different arrays per batch element. + + This tests: + - registration_tags: ["static", "$steps.model.tags"] + - When model.tags is batched and each element is an array: + - Batch element 0: ["tag-a", "tag-b"] + - Batch element 1: ["tag-c"] + - Batch element 2: ["tag-d", "tag-e", "tag-f"] + - Each batch should get its respective array + """ + # given + parameters = { + "target_project": "my_project", + "registration_tags": [ + "static-tag", + Batch( + content=[ + ["confidence-high", "quality-good"], + ["confidence-medium"], + ["confidence-low", "quality-poor", "review-needed"], + ], + indices=[(0,), (1,), (2,)], + ), + ], + } + + # when + result = list(unfold_parameters(parameters=parameters)) + + # then + assert len(result) == 3, "Expected three elements as selector is batched" + assert result[0] == { + "target_project": "my_project", + "registration_tags": ["static-tag", ["confidence-high", "quality-good"]], + }, "Expected first batch element with static tag and first array" + assert result[1] == { + "target_project": "my_project", + "registration_tags": ["static-tag", ["confidence-medium"]], + }, "Expected second batch element with static tag and second array" + assert result[2] == { + "target_project": "my_project", + "registration_tags": ["static-tag", ["confidence-low", "quality-poor", "review-needed"]], + }, "Expected third batch element with static tag and third array" + + def test_remove_empty_indices() -> None: # given value = { diff --git a/tests/workflows/unit_tests/execution_engine/introspection/test_schema_parser.py b/tests/workflows/unit_tests/execution_engine/introspection/test_schema_parser.py index 6d91895818..c838dc58b5 100644 --- a/tests/workflows/unit_tests/execution_engine/introspection/test_schema_parser.py +++ b/tests/workflows/unit_tests/execution_engine/introspection/test_schema_parser.py @@ -531,3 +531,328 @@ def describe_outputs(cls) -> List[OutputDefinition]: ) }, ) + + +def test_parse_block_manifest_when_manifest_defines_list_of_strings_and_string_selectors() -> ( + None +): + """ + Test for mixed array with literal strings and WorkflowParameterSelector(STRING_KIND). + This tests the registration_tags use case: ["literal", "$inputs.tag"] + """ + # given + + class Manifest(WorkflowBlockManifest): + type: Literal["MyManifest"] + name: str = Field(description="name field") + registration_tags: List[Union[WorkflowParameterSelector(kind=[STRING_KIND]), str]] + + @classmethod + def describe_outputs(cls) -> List[OutputDefinition]: + return [] + + # when + manifest_metadata = parse_block_manifest(manifest_type=Manifest) + + # then + assert manifest_metadata == BlockManifestMetadata( + primitive_types={ + "name": PrimitiveTypeDefinition( + property_name="name", + property_description="name field", + type_annotation="str", + ), + "registration_tags": PrimitiveTypeDefinition( + property_name="registration_tags", + property_description="not available", + type_annotation="List[str]", + ), + }, + selectors={ + "registration_tags": SelectorDefinition( + property_name="registration_tags", + property_description="not available", + allowed_references=[ + ReferenceDefinition( + selected_element="workflow_parameter", + kind=[STRING_KIND], + points_to_batch={False}, + ), + ], + is_list_element=True, + is_dict_element=False, + dimensionality_offset=0, + is_dimensionality_reference_property=False, + ) + }, + ) + + +def test_parse_block_manifest_when_manifest_defines_union_of_string_and_string_selector() -> ( + None +): + """ + Test for a property that can be either a string OR a selector. + This tests the case: registration_tags: Union[str, WorkflowParameterSelector] + Expected to resolve to just a string OR a selector to a string. + """ + # given + + class Manifest(WorkflowBlockManifest): + type: Literal["MyManifest"] + name: str = Field(description="name field") + single_tag: Union[WorkflowParameterSelector(kind=[STRING_KIND]), str] + + @classmethod + def describe_outputs(cls) -> List[OutputDefinition]: + return [] + + # when + manifest_metadata = parse_block_manifest(manifest_type=Manifest) + + # then + assert manifest_metadata == BlockManifestMetadata( + primitive_types={ + "name": PrimitiveTypeDefinition( + property_name="name", + property_description="name field", + type_annotation="str", + ), + "single_tag": PrimitiveTypeDefinition( + property_name="single_tag", + property_description="not available", + type_annotation="str", + ), + }, + selectors={ + "single_tag": SelectorDefinition( + property_name="single_tag", + property_description="not available", + allowed_references=[ + ReferenceDefinition( + selected_element="workflow_parameter", + kind=[STRING_KIND], + points_to_batch={False}, + ), + ], + is_list_element=False, + is_dict_element=False, + dimensionality_offset=0, + is_dimensionality_reference_property=False, + ) + }, + ) + + +def test_parse_block_manifest_when_manifest_defines_list_of_strings_and_list_selector() -> ( + None +): + """ + Test for a property that can be a list of strings OR a selector to a list. + This tests: Union[List[str], WorkflowParameterSelector] + Expected to handle the case where $inputs.tags itself is an array. + """ + # given + + class Manifest(WorkflowBlockManifest): + type: Literal["MyManifest"] + name: str = Field(description="name field") + tags: Union[ + List[str], + WorkflowParameterSelector(kind=[STRING_KIND]), + ] + + @classmethod + def describe_outputs(cls) -> List[OutputDefinition]: + return [] + + # when + manifest_metadata = parse_block_manifest(manifest_type=Manifest) + + # then + assert manifest_metadata == BlockManifestMetadata( + primitive_types={ + "name": PrimitiveTypeDefinition( + property_name="name", + property_description="name field", + type_annotation="str", + ), + "tags": PrimitiveTypeDefinition( + property_name="tags", + property_description="not available", + type_annotation="List[str]", + ), + }, + selectors={ + "tags": SelectorDefinition( + property_name="tags", + property_description="not available", + allowed_references=[ + ReferenceDefinition( + selected_element="workflow_parameter", + kind=[STRING_KIND], + points_to_batch={False}, + ), + ], + is_list_element=False, + is_dict_element=False, + dimensionality_offset=0, + is_dimensionality_reference_property=False, + ) + }, + ) + + +def test_parse_block_manifest_when_manifest_defines_only_list_of_strings() -> ( + None +): + """ + Test baseline case: just a list of strings with no selectors. + This should only create a primitive type definition. + """ + # given + + class Manifest(WorkflowBlockManifest): + type: Literal["MyManifest"] + name: str = Field(description="name field") + static_tags: List[str] + + @classmethod + def describe_outputs(cls) -> List[OutputDefinition]: + return [] + + # when + manifest_metadata = parse_block_manifest(manifest_type=Manifest) + + # then + assert manifest_metadata == BlockManifestMetadata( + primitive_types={ + "name": PrimitiveTypeDefinition( + property_name="name", + property_description="name field", + type_annotation="str", + ), + "static_tags": PrimitiveTypeDefinition( + property_name="static_tags", + property_description="not available", + type_annotation="List[str]", + ), + }, + selectors={}, + ) + + +def test_parse_block_manifest_when_manifest_defines_only_string_selector_list() -> ( + None +): + """ + Test a list containing only selectors (no literal strings). + This tests: List[WorkflowParameterSelector(STRING_KIND)] + """ + # given + + class Manifest(WorkflowBlockManifest): + type: Literal["MyManifest"] + name: str = Field(description="name field") + dynamic_tags: List[WorkflowParameterSelector(kind=[STRING_KIND])] + + @classmethod + def describe_outputs(cls) -> List[OutputDefinition]: + return [] + + # when + manifest_metadata = parse_block_manifest(manifest_type=Manifest) + + # then + assert manifest_metadata == BlockManifestMetadata( + primitive_types={ + "name": PrimitiveTypeDefinition( + property_name="name", + property_description="name field", + type_annotation="str", + ), + }, + selectors={ + "dynamic_tags": SelectorDefinition( + property_name="dynamic_tags", + property_description="not available", + allowed_references=[ + ReferenceDefinition( + selected_element="workflow_parameter", + kind=[STRING_KIND], + points_to_batch={False}, + ), + ], + is_list_element=True, + is_dict_element=False, + dimensionality_offset=0, + is_dimensionality_reference_property=False, + ) + }, + ) + + +def test_parse_block_manifest_when_manifest_defines_mixed_selectors_and_step_output_selectors() -> ( + None +): + """ + Test mixed array with different selector types (WorkflowParameter and StepOutput). + This tests: List[Union[WorkflowParameterSelector, StepOutputSelector, str]] + """ + # given + + class Manifest(WorkflowBlockManifest): + type: Literal["MyManifest"] + name: str = Field(description="name field") + mixed_references: List[ + Union[ + WorkflowParameterSelector(kind=[STRING_KIND]), + StepOutputSelector(kind=[STRING_KIND]), + str, + ] + ] + + @classmethod + def describe_outputs(cls) -> List[OutputDefinition]: + return [] + + # when + manifest_metadata = parse_block_manifest(manifest_type=Manifest) + + # then + assert manifest_metadata == BlockManifestMetadata( + primitive_types={ + "name": PrimitiveTypeDefinition( + property_name="name", + property_description="name field", + type_annotation="str", + ), + "mixed_references": PrimitiveTypeDefinition( + property_name="mixed_references", + property_description="not available", + type_annotation="List[str]", + ), + }, + selectors={ + "mixed_references": SelectorDefinition( + property_name="mixed_references", + property_description="not available", + allowed_references=[ + ReferenceDefinition( + selected_element="workflow_parameter", + kind=[STRING_KIND], + points_to_batch={False}, + ), + ReferenceDefinition( + selected_element="step_output", + kind=[STRING_KIND], + points_to_batch={True}, + ), + ], + is_list_element=True, + is_dict_element=False, + dimensionality_offset=0, + is_dimensionality_reference_property=False, + ) + }, + )