src/backend/tests/integration/components/helpers/test_parse_json_data.py

from langflow.components.inputs import ChatInput
from langflow.components.processing.parse_json_data import ParseJSONDataComponent
from langflow.schema import Data

from tests.integration.components.mock_components import TextToData
from tests.integration.utils import ComponentInputHandle, run_single_component


async def test_from_data():
    outputs = await run_single_component(
        ParseJSONDataComponent,
        inputs={
            "input_value": ComponentInputHandle(
                clazz=TextToData, inputs={"text_data": ['{"key":"value1"}'], "is_json": True}, output_name="from_text"
            ),
            "query": ".[0].key",
        },
    )
    assert outputs["filtered_data"] == [Data(text="value1")]

    outputs = await run_single_component(
        ParseJSONDataComponent,
        inputs={
            "input_value": ComponentInputHandle(
                clazz=TextToData,
                inputs={"text_data": ['{"key":[{"field1": 1, "field2": 2}]}'], "is_json": True},
                output_name="from_text",
            ),
            "query": ".[0].key.[0].field2",
        },
    )
    assert outputs["filtered_data"] == [Data(text="2")]


async def test_from_message():
    outputs = await run_single_component(
        ParseJSONDataComponent,
        inputs={
            "input_value": ComponentInputHandle(clazz=ChatInput, inputs={}, output_name="message"),
            "query": ".key",
        },
        run_input="{'key':'value1'}",
    )
    assert outputs["filtered_data"] == [Data(text="value1")]

    outputs = await run_single_component(
        ParseJSONDataComponent,
        inputs={
            "input_value": ComponentInputHandle(clazz=ChatInput, inputs={}, output_name="message"),
            "query": ".key.[0].field2",
        },
        run_input='{"key":[{"field1": 1, "field2": 2}]}',
    )
    assert outputs["filtered_data"] == [Data(text="2")]