"""Unit test for custom code execution in isolation (with real Langflow code).""" from typing import Any from unittest.mock import AsyncMock import pytest from langflow_stepflow.exceptions import ExecutionError from langflow_stepflow.worker.custom_code_executor import ( CustomCodeExecutor, ) class TestCustomCodeExecutor: """Test CustomCodeExecutor functionality with real Langflow component code.""" @pytest.fixture def executor(self): """Create CustomCodeExecutor instance.""" return CustomCodeExecutor() @pytest.fixture def mock_context(self): """Create mock StepflowContext.""" context = AsyncMock() context.get_blob = AsyncMock() context.put_blob = AsyncMock() return context @pytest.fixture def simple_component_blob(self): """Simple test component blob data.""" return { "code": ''' from langflow.custom.custom_component.component import Component from langflow.io import MessageTextInput, Output from langflow.schema.message import Message class SimpleTestComponent(Component): display_name = "Simple Test" description = "A simple test component" inputs = [ MessageTextInput( name="text_input", display_name="Text Input", info="Text input for testing" ) ] outputs = [ Output(display_name="Output", name="result", method="process_text") ] async def process_text(self) -> Message: """Process the input text and return a message.""" input_text = self.text_input or "No input provided" result_text = f"Processed: {input_text}" return Message( text=result_text, sender="SimpleTestComponent", sender_name="Test Component" ) ''', "component_type": "SimpleTestComponent", "template": { "text_input": { "type": "str", "value": "", "info": "Text input for testing", "required": False, } }, "outputs": [{"name": "result", "method": "process_text", "types": ["Message"]}], "selected_output": "result", } @pytest.fixture def chat_input_component_blob(self): """Real ChatInput component blob data from Langflow.""" return { "code": ''' from langflow.base.io.chat import ChatComponent from langflow.inputs.inputs import BoolInput from langflow.io import DropdownInput, MessageTextInput, MultilineInput, Output from langflow.schema.message import Message from langflow.utils.constants import ( MESSAGE_SENDER_AI, MESSAGE_SENDER_USER, MESSAGE_SENDER_NAME_USER ) class ChatInput(ChatComponent): display_name = "Chat Input" description = "Get chat inputs from the Playground." icon = "MessagesSquare" name = "ChatInput" inputs = [ MultilineInput( name="input_value", display_name="Input Text", value="", info="Message to be passed as input.", ), DropdownInput( name="sender", display_name="Sender Type", options=[MESSAGE_SENDER_AI, MESSAGE_SENDER_USER], value=MESSAGE_SENDER_USER, info="Type of sender.", advanced=True, ), MessageTextInput( name="sender_name", display_name="Sender Name", info="Name of the sender.", value=MESSAGE_SENDER_NAME_USER, advanced=True, ), ] outputs = [ Output(display_name="Chat Message", name="message", method="message_response"), ] async def message_response(self) -> Message: """Create a message from the input.""" message = Message( text=self.input_value or "", sender=self.sender, sender_name=self.sender_name, ) self.status = message return message ''', "component_type": "ChatInput", "template": { "input_value": { "type": "str", "value": "", "info": "Message to be passed as input", "required": True, }, "sender": { "type": "str", "value": "User", "options": ["AI", "User"], "info": "Type of sender", }, "sender_name": { "type": "str", "value": "User", "info": "Name of the sender", }, }, "outputs": [{"name": "message", "method": "message_response", "types": ["Message"]}], "selected_output": "message", } @pytest.mark.asyncio async def test_execute_missing_blob_id(self, executor: CustomCodeExecutor, mock_context): """Test execution fails when blob_id is missing.""" input_data = {"input": {"text": "test"}} with pytest.raises(ExecutionError, match="No blob_id provided"): await executor.execute(input_data, mock_context) @pytest.mark.asyncio async def test_execute_simple_component( self, executor: CustomCodeExecutor, mock_context, simple_component_blob: dict[str, Any], ): """Test execution of a simple custom component.""" # Setup mock context mock_context.get_blob.return_value = simple_component_blob # Input data with blob_id and runtime inputs input_data = { "blob_id": "test_blob_123", "input": {"text_input": "Hello World"}, } # Execute result = await executor.execute(input_data, mock_context) # Verify blob was retrieved mock_context.get_blob.assert_called_once_with("test_blob_123") # Verify result structure assert "result" in result assert isinstance(result["result"], dict) # Verify the processed result result_data = result["result"] # Langflow Message: result data is nested in ["result"] field message_data = result_data["result"] if "result" in result_data else result_data assert "text" in message_data assert message_data["text"] == "Processed: Hello World" assert message_data["sender"] == "SimpleTestComponent" assert message_data["sender_name"] == "Test Component" @pytest.mark.asyncio async def test_execute_chat_input_component( self, executor: CustomCodeExecutor, mock_context, chat_input_component_blob: dict[str, Any], ): """Test execution of real ChatInput component.""" # Setup mock context mock_context.get_blob.return_value = chat_input_component_blob # Input data input_data = { "blob_id": "chatinput_blob_456", "input": { "input_value": "What is Python?", "sender": "User", "sender_name": "Test User", }, } # Execute result = await executor.execute(input_data, mock_context) # Verify result assert "result" in result result_data = result["result"] # Should have Langflow Message structure # Handle nested result structure from Langflow Messages message_data = result_data["result"] if "result" in result_data else result_data assert "text" in message_data assert message_data["text"] == "What is Python?" assert message_data["sender"] == "User" assert message_data["sender_name"] == "Test User" assert message_data.get("__langflow_type__") == "Message" @pytest.mark.asyncio async def test_execute_component_with_template_defaults( self, executor: CustomCodeExecutor, mock_context, simple_component_blob: dict[str, Any], ): """Test that template default values are used when input is not provided.""" # Modify blob to have default value simple_component_blob["template"]["text_input"]["value"] = "Default Text" mock_context.get_blob.return_value = simple_component_blob # Input data without runtime input for text_input input_data = { "blob_id": "test_blob_789", "input": {}, # No runtime inputs provided } # Execute result = await executor.execute(input_data, mock_context) # Should use template default result_data = result["result"] # Handle nested result structure from Langflow Messages message_data = result_data["result"] if "result" in result_data else result_data assert message_data["text"] == "Processed: Default Text" @pytest.mark.asyncio async def test_execute_component_runtime_overrides_template( self, executor: CustomCodeExecutor, mock_context, simple_component_blob: dict[str, Any], ): """Test that runtime inputs override template defaults.""" # Set template default simple_component_blob["template"]["text_input"]["value"] = "Template Default" mock_context.get_blob.return_value = simple_component_blob # Input data with runtime override input_data = { "blob_id": "test_blob_override", "input": {"text_input": "Runtime Override"}, } # Execute result = await executor.execute(input_data, mock_context) # Should use runtime value, not template default result_data = result["result"] # Handle nested result structure from Langflow Messages message_data = result_data["result"] if "result" in result_data else result_data assert message_data["text"] == "Processed: Runtime Override" @pytest.mark.asyncio async def test_execute_component_with_missing_code(self, executor: CustomCodeExecutor, mock_context): """Test execution fails when component code is missing.""" blob_data = { "component_type": "TestComponent", "template": {}, "outputs": [], # Missing "code" field } mock_context.get_blob.return_value = blob_data input_data = {"blob_id": "missing_code_blob", "input": {}} with pytest.raises(ExecutionError, match="No code found for component"): await executor.execute(input_data, mock_context) @pytest.mark.asyncio async def test_execute_component_with_invalid_code(self, executor: CustomCodeExecutor, mock_context): """Test execution fails when component code is invalid Python.""" blob_data = { "code": "invalid python syntax !!!", "component_type": "InvalidComponent", "template": {}, "outputs": [], } mock_context.get_blob.return_value = blob_data input_data = {"blob_id": "invalid_code_blob", "input": {}} with pytest.raises(ExecutionError, match="Failed to evaluate component code"): await executor.execute(input_data, mock_context) @pytest.mark.asyncio async def test_execute_component_class_not_found(self, executor: CustomCodeExecutor, mock_context): """Test execution fails when component class is not found in code.""" blob_data = { "code": """ # Valid Python code but no matching class def some_function(): return "not a component class" """, "component_type": "MissingComponent", "template": {}, "outputs": [], } mock_context.get_blob.return_value = blob_data input_data = {"blob_id": "no_class_blob", "input": {}} with pytest.raises( ExecutionError, match="Failed to evaluate component code for MissingComponent", ): await executor.execute(input_data, mock_context) @pytest.mark.asyncio async def test_execute_component_instantiation_fails(self, executor: CustomCodeExecutor, mock_context): """Test execution fails when component cannot be instantiated.""" blob_data = { "code": """ from langflow.custom.custom_component.component import Component from langflow.io import Output class FailingComponent(Component): display_name = "Failing Component" description = "Component that fails instantiation" outputs = [ Output(display_name="Result", name="result", method="execute") ] def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) raise ValueError("Cannot instantiate this component") def execute(self): return "should not reach this" """, "component_type": "FailingComponent", "template": {}, "outputs": [{"name": "result", "method": "execute", "types": ["str"]}], } mock_context.get_blob.return_value = blob_data input_data = {"blob_id": "failing_init_blob", "input": {}} with pytest.raises(ExecutionError, match="Failed to instantiate FailingComponent"): await executor.execute(input_data, mock_context) @pytest.mark.asyncio async def test_execute_component_method_not_found(self, executor: CustomCodeExecutor, mock_context): """Test execution fails when execution method is not found.""" blob_data = { "code": """ from langflow.custom.custom_component.component import Component class NoMethodComponent(Component): pass """, "component_type": "NoMethodComponent", "template": {}, "outputs": [{"name": "result", "method": "nonexistent_method", "types": ["str"]}], "selected_output": "result", } mock_context.get_blob.return_value = blob_data input_data = {"blob_id": "no_method_blob", "input": {}} with pytest.raises(ExecutionError, match="Method nonexistent_method not found"): await executor.execute(input_data, mock_context) @pytest.mark.asyncio async def test_execute_component_method_execution_fails(self, executor: CustomCodeExecutor, mock_context): """Test execution fails when component method throws exception.""" blob_data = { "code": """ from langflow.custom.custom_component.component import Component class FailingMethodComponent(Component): async def failing_method(self): raise RuntimeError("Method execution failed") """, "component_type": "FailingMethodComponent", "template": {}, "outputs": [{"name": "result", "method": "failing_method", "types": ["str"]}], "selected_output": "result", } mock_context.get_blob.return_value = blob_data input_data = {"blob_id": "failing_method_blob", "input": {}} with pytest.raises(ExecutionError, match="Failed to execute failing_method"): await executor.execute(input_data, mock_context) def test_environment_variable_handling_deprecated(self, executor: CustomCodeExecutor): """Test that environment variable handling is now handled via preprocessing. This test documents that environment variable resolution was moved from runtime in the UDF executor to preprocessing at the test/workflow level. The _determine_environment_variable method was removed as part of the simplification effort to rely on preprocessing instead. """ # Environment variable resolution is now handled by preprocessing # instead of runtime resolution in the UDF executor assert True # This test now just documents the architectural change def test_determine_execution_method(self, executor: CustomCodeExecutor): """Test execution method determination from outputs metadata.""" outputs = [ {"name": "result1", "method": "method1", "types": ["str"]}, {"name": "result2", "method": "method2", "types": ["int"]}, ] # Selected output specified method = executor._determine_execution_method(outputs, "result2") assert method == "method2" # No selected output, use first method = executor._determine_execution_method(outputs, None) assert method == "method1" # Empty outputs method = executor._determine_execution_method([], None) assert method is None # Selected output not found, use first method = executor._determine_execution_method(outputs, "nonexistent") assert method == "method1" class TestCustomCodeExecutorDataFrameOutput: """Test CustomCodeExecutor with components that return DataFrames (issue #673).""" @pytest.fixture def executor(self): return CustomCodeExecutor() @pytest.fixture def mock_context(self): context = AsyncMock() context.get_blob = AsyncMock() return context @pytest.fixture def pandas_dataframe_component_blob(self): """Component that returns a plain pandas DataFrame (not lfx DataFrame). This simulates the URLComponent scenario from issue #673 where compiled component code produces a plain pandas DataFrame. """ return { "code": ''' import pandas as pd from langflow.custom.custom_component.component import Component from langflow.io import Output class DataFrameComponent(Component): display_name = "DataFrame Test" description = "Returns a plain pandas DataFrame" outputs = [ Output(display_name="Content", name="content", method="fetch_content") ] def fetch_content(self) -> pd.DataFrame: """Return a plain pandas DataFrame (not lfx DataFrame).""" return pd.DataFrame([ {"text": "Hello world", "url": "http://example.com", "title": "Test"}, {"text": "Second row", "url": "http://example.org", "title": "Test 2"}, ]) ''', "component_type": "DataFrameComponent", "template": {}, "outputs": [{"name": "content", "method": "fetch_content", "types": ["DataFrame"]}], "selected_output": "content", } @pytest.mark.asyncio async def test_plain_pandas_dataframe_serializes( self, executor: CustomCodeExecutor, mock_context, pandas_dataframe_component_blob, ): """Plain pandas DataFrame from component should serialize (issue #673). Before the fix, this would raise: ValueError: Cannot serialize object of type DataFrame. Only BaseModel objects and simple types are supported. """ mock_context.get_blob.return_value = pandas_dataframe_component_blob input_data = {"blob_id": "test_df_blob", "input": {}} result = await executor.execute(input_data, mock_context) assert "result" in result output = result["result"] assert isinstance(output, dict) assert output["__langflow_type__"] == "DataFrame" assert "json_data" in output class TestCustomCodeExecutorIntegration: """Integration tests with mock Langflow imports.""" @pytest.fixture def executor_with_mocks(self): """Create CustomCodeExecutor with mocked Langflow imports.""" # We'll test this without real Langflow imports to avoid dependency issues return CustomCodeExecutor() # Registry fixture removed - no longer needed after eliminating test registry @pytest.fixture def converter(self): """Create converter instance.""" from langflow_stepflow.translation.translator import LangflowConverter return LangflowConverter() @pytest.mark.asyncio async def test_component_parameter_preparation(self, executor_with_mocks: CustomCodeExecutor): """Test component parameter preparation logic. Note: Environment variable resolution is now handled by preprocessing, not at runtime in the UDF executor. This test reflects the new approach. """ template = { "text_field": { "type": "str", "value": "template_default", "info": "Text input field", }, "api_key": { "type": "str", # In the new approach, preprocessing would have already resolved this "value": "test-api-key-123", # Already preprocessed value "_input_type": "SecretStrInput", }, "number_field": {"type": "int", "value": 42}, } runtime_inputs = { "text_field": "runtime_override", "extra_field": "extra_value", } params = await executor_with_mocks._prepare_component_parameters(template, runtime_inputs) # Should have runtime override assert params["text_field"] == "runtime_override" # Should have preprocessed API key (not runtime resolved) assert params["api_key"] == "test-api-key-123" # pragma: allowlist secret # Should have template default assert params["number_field"] == 42 # Should have runtime extra field assert params["extra_field"] == "extra_value" class TestCustomCodeExecutorWithRealLangflowComponents: """Test CustomCodeExecutor with real converted Langflow workflow components.""" @pytest.fixture def executor(self): """Create CustomCodeExecutor instance.""" return CustomCodeExecutor() @pytest.fixture def mock_context(self): """Create mock StepflowContext with blob storage.""" context = AsyncMock() context.get_blob = AsyncMock() context.put_blob = AsyncMock() return context @pytest.fixture def converter(self): """Create converter instance.""" from langflow_stepflow.translation.translator import LangflowConverter return LangflowConverter() @pytest.mark.asyncio async def test_execute_converted_workflow_components(self, executor: CustomCodeExecutor, mock_context, converter): """Test that components from converted workflows execute correctly.""" # Create simple workflow data to test UDF component creation langflow_data = { "data": { "nodes": [ { "id": "test-prompt", "data": { "type": "Prompt", "node": { "template": { "code": { "value": """ from langflow.custom.custom_component.component import Component from langflow.io import Output from langflow.schema.message import Message class TestPrompt(Component): def build_prompt(self) -> Message: return Message(text="Test prompt") """ } }, "outputs": [{"name": "prompt", "method": "build_prompt"}], "base_classes": ["Message"], "display_name": "Test Prompt", }, "outputs": [{"name": "prompt", "method": "build_prompt"}], }, } ], "edges": [], } } # Convert to find the custom code components stepflow_workflow = converter.convert(langflow_data) # Find steps that use /langflow/custom_code custom_code_steps = [step for step in stepflow_workflow.steps if step.component == "/langflow/custom_code"] assert len(custom_code_steps) > 0, "No custom code executor steps found in test workflow" # Test the first custom code step first_step = custom_code_steps[0] # The step input should have blob_id reference assert "blob_id" in str(first_step.input), f"No blob_id found in step input: {first_step.input}" print(f"✅ Found {len(custom_code_steps)} custom code components") print(f"✅ First custom code step: {first_step.id} using {first_step.component}") @pytest.mark.asyncio async def test_component_metadata_extraction_and_usage(self, executor: CustomCodeExecutor, mock_context): # Create blob with enhanced metadata (from Phase 1 improvements) enhanced_blob = { "code": """ from langflow.custom.custom_component.component import Component from langflow.io import Output from langflow.schema.message import Message class EnhancedTestComponent(Component): display_name = "Enhanced Test" description = "Test component with enhanced metadata" outputs = [ Output(display_name="Message", name="message", method="create_message") ] def create_message(self) -> Message: return Message(text="Enhanced metadata test successful") """, "template": {"input_field": {"type": "str", "value": "test"}}, "component_type": "EnhancedTestComponent", "outputs": [{"name": "message", "method": "create_message", "types": ["Message"]}], "selected_output": "message", # Enhanced metadata from Phase 1 "base_classes": ["Message"], "display_name": "Enhanced Test", "description": "Test component with enhanced metadata", "documentation": "https://docs.example.com/enhanced-test", "metadata": { "module": "test.module.EnhancedTestComponent", "code_hash": "abc123", }, "field_order": ["input_field"], "icon": "test-icon", "is_builtin": False, } mock_context.get_blob.return_value = enhanced_blob # Execute with enhanced metadata input_data = {"blob_id": "enhanced_test_blob", "input": {}} result = await executor.execute(input_data, mock_context) # Verify execution used enhanced metadata assert result["result"]["text"] == "Enhanced metadata test successful" # Verify compiled component contains enhanced metadata compiled = executor.compiled_components["enhanced_test_blob"] assert compiled["base_classes"] == ["Message"] assert compiled["display_name"] == "Enhanced Test" assert compiled["metadata"]["module"] == "test.module.EnhancedTestComponent" @pytest.fixture def basic_prompting_flow(self): """Load the basic_prompting flow fixture.""" import json from pathlib import Path flow_path = Path(__file__).parent.parent / "fixtures" / "langflow" / "basic_prompting.json" if not flow_path.exists(): pytest.skip("basic_prompting.json fixture not found") with open(flow_path) as f: return json.load(f) @pytest.fixture def prompt_component_data(self, basic_prompting_flow): """Extract Prompt component data from basic_prompting flow.""" nodes = basic_prompting_flow["data"]["nodes"] prompt_node = next(n for n in nodes if n["data"]["type"] == "Prompt") return { "code": prompt_node["data"]["node"]["template"]["code"]["value"], "component_type": "PromptComponent", "template": prompt_node["data"]["node"]["template"], "outputs": prompt_node["data"]["node"]["outputs"], "selected_output": prompt_node["data"].get("selected_output", "prompt"), } @pytest.fixture def chat_input_component_data(self, basic_prompting_flow): """Extract ChatInput component data (lfx-based) from basic_prompting flow.""" nodes = basic_prompting_flow["data"]["nodes"] chat_input_node = next(n for n in nodes if n["data"]["type"] == "ChatInput") return { "code": chat_input_node["data"]["node"]["template"]["code"]["value"], "component_type": "ChatInput", "template": chat_input_node["data"]["node"]["template"], "outputs": chat_input_node["data"]["node"]["outputs"], "selected_output": chat_input_node["data"].get("selected_output", "message"), } @pytest.fixture def agent_component_blob(self): """Agent component blob data from simple_agent.json flow. This tests the PlaceholderGraph.vertices fix for lfx Agent components. """ import json import os # Load Agent component data from simple_agent.json fixture_path = os.path.join( os.path.dirname(__file__), "..", "fixtures", "langflow", "simple_agent.json", ) if not os.path.exists(fixture_path): pytest.skip("simple_agent.json fixture not found") with open(fixture_path) as f: flow = json.load(f) # Find Agent node for node in flow["data"]["nodes"]: if "Agent" in node["data"]["node"]["display_name"]: agent_data = node["data"]["node"] return { "code": agent_data["template"]["code"]["value"], "template": agent_data["template"], "outputs": agent_data.get("outputs", []), "component_type": agent_data["display_name"], "base_classes": agent_data.get("base_classes", []), "display_name": agent_data["display_name"], "description": agent_data.get("description", ""), "selected_output": "response", } raise ValueError("Agent component not found in simple_agent.json") @pytest.mark.asyncio async def test_agent_component(self, executor: CustomCodeExecutor, mock_context, agent_component_blob): """Test that Agent component can be compiled and instantiated. This test validates the PlaceholderGraph.vertices fix by: 1. Compiling the Agent component from lfx 2. Instantiating the component 3. Accessing the graph.vertices attribute (which would fail without the fix) The fix patches lfx.custom.custom_component.component.PlaceholderGraph with an EnhancedPlaceholderGraph that includes the vertices attribute. """ mock_context.get_blob.return_value = agent_component_blob # Step 1: Compile the component compiled_component = await executor._compile_component(agent_component_blob) assert compiled_component is not None assert compiled_component["component_type"] == "Agent" # Step 2: Instantiate the component component_class = compiled_component["class"] component_instance = component_class() assert component_instance is not None # Step 3: Access graph property - this triggers PlaceholderGraph creation graph = component_instance.graph assert graph is not None # Step 4: Access vertices attribute - this would fail without the fix # The lfx Agent component code accesses graph.vertices internally vertices = graph.vertices assert vertices is not None assert isinstance(vertices, (list, tuple)) assert len(vertices) == 0 # Empty default for EnhancedPlaceholderGraph @pytest.mark.asyncio async def test_agent_component_execution_attempt( self, executor: CustomCodeExecutor, mock_context, agent_component_blob ): """Test Agent component execution (will fail without real API key). This test goes beyond just instantiation to attempt actual execution. It will fail due to missing API key, but should NOT fail with PlaceholderGraph errors. """ mock_context.get_blob.return_value = agent_component_blob # Prepare minimal input for Agent execution blob_id = "test_agent_execution" input_data = { "blob_id": blob_id, "input": { "input_value": "What is 2+2?", "session_id": "test_session", "api_key": "sk-test-fake-key", # Fake key - will fail but tests setup # pragma: allowlist secret "model_name": "gpt-4", "tools": [], # No tools to simplify test }, } # Try to execute - expect failure due to API key, NOT PlaceholderGraph try: result = await executor.execute(input_data, mock_context) # If it somehow succeeds with fake key, that's OK print(f"Unexpected success: {result}") except Exception as e: error_msg = str(e) # Should NOT have PlaceholderGraph.vertices error assert "vertices" not in error_msg.lower(), f"PlaceholderGraph.vertices error detected: {error_msg}" # Expected errors: API key, authentication, model errors error_info = f"{type(e).__name__}: {error_msg[:200]}" print(f"Expected error (not PlaceholderGraph): {error_info}") @pytest.mark.asyncio async def test_prompt_component(self, executor: CustomCodeExecutor, mock_context, prompt_component_data): """Test executing Prompt component with string-type system_message field. This test verifies that: 1. We can compile and execute a Prompt component from basic_prompting flow 2. The component successfully processes template input 3. The output is a Message object with text content """ # Create blob data for the component blob_data = { "code": prompt_component_data["code"], "component_type": prompt_component_data["component_type"], "outputs": prompt_component_data["outputs"], "selected_output": prompt_component_data["selected_output"], "template": prompt_component_data["template"], } # Set up mock context blob_id = "test-prompt-blob" mock_context.get_blob.return_value = blob_data # Prepare input data template_text = ( "Answer the user as if you were a GenAI expert, enthusiastic " "about helping them get started building something fresh." ) input_data = { "blob_id": blob_id, "input": { "template": template_text, }, } # Execute the component result = await executor.execute(input_data, mock_context) # Verify the result assert isinstance(result, dict) # Extract the actual output (may be wrapped in 'result' or 'output') if "result" in result: output = result["result"] elif "output" in result: output = result["output"] else: output = result # The Prompt component should return a Message object # Check for both old and new serialization formats has_message_marker = "__langflow_type__" in output or "__class_name__" in output output_desc = output.keys() if isinstance(output, dict) else type(output) assert has_message_marker, f"Expected Message object, got: {output_desc}" assert "text" in output assert len(output["text"]) > 0 @pytest.mark.asyncio async def test_language_model_message_to_string_conversion( self, executor: CustomCodeExecutor, mock_context, basic_prompting_flow ): """Test LanguageModelComponent receives string when Message passed. This test verifies the fix for Langflow 1.6.4+ lfx components that expect string values for MultilineInput fields, not Message objects. """ nodes = basic_prompting_flow["data"]["nodes"] lm_node = next(n for n in nodes if n["data"].get("type") == "LanguageModelComponent") # Create blob data for LanguageModelComponent component lm_blob_data = { "code": lm_node["data"]["node"]["template"]["code"]["value"], "component_type": "LanguageModelComponent", "template": lm_node["data"]["node"]["template"], "outputs": lm_node["data"]["node"]["outputs"], } # Create a Message object that will be passed to system_message field message_obj = { "__langflow_type__": "Message", "text": "You are a helpful AI assistant.", "sender": "User", "sender_name": "User", "session_id": "", } lm_blob_id = "test-lm-blob" mock_context.get_blob.return_value = lm_blob_data # Prepare input with Message object for string field input_data = { "blob_id": lm_blob_id, "input": { "input_value": "Hello, world!", "system_message": message_obj, # Message object for string field "api_key": "fake-key-for-testing", # pragma: allowlist secret }, } # The execution will fail because we don't have a real API key, # but we want to verify it doesn't fail with a validation error # about Message type being passed to MultilineInput try: await executor.execute(input_data, mock_context) # If it succeeds, great! (won't happen without real API key) except Exception as e: error_msg = str(e) # Should NOT have validation error about Message type assert "Invalid value type" not in error_msg or "Message" not in error_msg, ( f"Should not have Message type validation error. Got: {error_msg}" ) # Other errors (like missing API key) are expected assert ( "api_key" in error_msg.lower() or "openai" in error_msg.lower() or "model" in error_msg.lower() or "execute" in error_msg.lower() ), f"Expected API/model error, got: {error_msg}" @pytest.mark.asyncio async def test_chat_input_lfx_component( self, executor: CustomCodeExecutor, mock_context, chat_input_component_data ): """Test executing ChatInput component (lfx-based) from basic_prompting flow. This test verifies that: 1. We can compile and execute an lfx-based ChatInput component 2. The component accepts string input_value 3. The component returns a Message object """ # Create blob data for the component blob_data = { "code": chat_input_component_data["code"], "component_type": chat_input_component_data["component_type"], "outputs": chat_input_component_data["outputs"], "selected_output": chat_input_component_data["selected_output"], "template": chat_input_component_data["template"], } # Set up mock context blob_id = "test-chatinput-blob" mock_context.get_blob.return_value = blob_data # Prepare input data with string input_value input_data = { "blob_id": blob_id, "input": { "input_value": "Hello from the test!", "should_store_message": False, # Don't try to store without session }, } # Execute the component result = await executor.execute(input_data, mock_context) # Verify the result assert isinstance(result, dict) # Check for result/output (wrapped result) or direct Message fields if "result" in result: output = result["result"] elif "output" in result: output = result["output"] else: output = result # Should be a Message object (check for both old and new serialization) has_message_marker = "__langflow_type__" in output or "__class_name__" in output output_desc = output.keys() if isinstance(output, dict) else type(output) assert has_message_marker, f"Expected Message object, got: {output_desc}" assert "text" in output assert output["text"] == "Hello from the test!" @pytest.mark.asyncio async def test_chat_input_message_to_string_conversion( self, executor: CustomCodeExecutor, mock_context, chat_input_component_data ): """Test ChatInput component with Message object passed to string field. This is a specific test for the Langflow 1.6.4+ fix where Message objects need to be converted to strings when passed to MultilineInput fields. """ # Create blob data blob_data = { "code": chat_input_component_data["code"], "component_type": chat_input_component_data["component_type"], "outputs": chat_input_component_data["outputs"], "selected_output": chat_input_component_data["selected_output"], "template": chat_input_component_data["template"], } blob_id = "test-chatinput-message-blob" mock_context.get_blob.return_value = blob_data # Pass a Message object to the input_value field (which expects str) message_obj = { "__class_name__": "Message", "__module_name__": "lfx.schema.message", "text": "Hello from Message object!", "sender": "User", "sender_name": "User", "session_id": "", } input_data = { "blob_id": blob_id, "input": { "input_value": message_obj, # Message object for string field "should_store_message": False, }, } # Execute - should convert Message to string automatically result = await executor.execute(input_data, mock_context) # Verify execution succeeded (component should extract text from Message) assert isinstance(result, dict) if "result" in result: output = result["result"] elif "output" in result: output = result["output"] else: output = result # Should have created a new Message with the text assert "text" in output # The text should be extracted from the input Message object assert "Hello from Message object!" in output["text"]