ci: add tests for text input/output components (langflow-ai#3649)

* ci: add tests for text input/output components

* ci: add tests for text input/output components

* [autofix.ci] apply automated fixes

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

src/backend/tests/integration/components/inputs/test_text_input.py

@@ -0,0 +1,21 @@
+from langflow.schema.message import Message
+from tests.integration.utils import run_single_component
+
+from langflow.components.inputs import TextInputComponent
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_text_input():
+    outputs = await run_single_component(TextInputComponent, run_input="sample text", input_type="text")
+    print(outputs)
+    assert isinstance(outputs["text"], Message)
+    assert outputs["text"].text == "sample text"
+    assert outputs["text"].sender is None
+    assert outputs["text"].sender_name is None
+
+    outputs = await run_single_component(TextInputComponent, run_input="", input_type="text")
+    assert isinstance(outputs["text"], Message)
+    assert outputs["text"].text == ""
+    assert outputs["text"].sender is None
+    assert outputs["text"].sender_name is None

src/backend/tests/integration/components/outputs/test_text_output.py

@@ -0,0 +1,23 @@
+from langflow.components.outputs import TextOutputComponent
+from langflow.schema.message import Message
+from tests.integration.utils import run_single_component
+
+import pytest
+
+
+@pytest.mark.asyncio
+async def test():
+    outputs = await run_single_component(TextOutputComponent, inputs={"input_value": "hello"})
+    assert isinstance(outputs["text"], Message)
+    assert outputs["text"].text == "hello"
+    assert outputs["text"].sender is None
+    assert outputs["text"].sender_name is None
+
+
+@pytest.mark.asyncio
+async def test_message():
+    outputs = await run_single_component(TextOutputComponent, inputs={"input_value": Message(text="hello")})
+    assert isinstance(outputs["text"], Message)
+    assert outputs["text"].text == "hello"
+    assert outputs["text"].sender is None
+    assert outputs["text"].sender_name is None

src/backend/tests/integration/utils.py

@@ -136,7 +136,11 @@ class ComponentInputHandle:
 
 
 async def run_single_component(
-    clazz: type, inputs: dict = None, run_input: Optional[Any] = None, session_id: Optional[str] = None
+    clazz: type,
+    inputs: dict = None,
+    run_input: Optional[Any] = None,
+    session_id: Optional[str] = None,
+    input_type: Optional[str] = "chat",
 ) -> dict[str, Any]:
     user_id = str(uuid.uuid4())
     flow_id = str(uuid.uuid4())
@@ -162,7 +166,7 @@ def _add_component(clazz: type, inputs: Optional[dict] = None) -> str:
     component_id = _add_component(clazz, inputs)
     graph.prepare()
     if run_input:
-        graph_run_inputs = [InputValueRequest(input_value=run_input, type="chat")]
+        graph_run_inputs = [InputValueRequest(input_value=run_input, type=input_type)]
     else:
         graph_run_inputs = []