"""Tests for lfx.graph.flow_builder — pure flow-building utilities.

All tests use synthetic registries (plain dicts) — no lfx component loading,
no I/O, no network.
"""

import json
from pathlib import Path

import pytest
from lfx.graph.flow_builder import (
    add_component,
    add_connection,
    configure_component,
    empty_flow,
    flow_info,
    get_component,
    layout_flow,
    list_components,
    list_connections,
    needs_server_update,
    parse_flow_spec,
    remove_component,
    remove_connection,
)
from lfx.graph.flow_builder.connect import _custom_stringify, _scaped_json_stringify

# ---------------------------------------------------------------------------
# Synthetic registry — minimal templates for testing
# ---------------------------------------------------------------------------


def _make_template(display_name, outputs=None, template_fields=None):
    tmpl = {
        "display_name": display_name,
        "base_classes": ["Message"],
        "outputs": outputs or [{"name": "message", "types": ["Message"]}],
        "template": template_fields
        or {
            "input_value": {
                "display_name": "Input Text",
                "type": "str",
                "value": "",
                "input_types": ["Message"],
            },
        },
    }
    return tmpl  # noqa: RET504


REGISTRY = {
    "ChatInput": _make_template("Chat Input"),
    "ToolOutput": _make_template(
        "Tool Output",
        outputs=[{"name": "tool", "types": ["Tool"]}],
        template_fields={
            "tools": {
                "display_name": "Tools",
                "type": "other",
                "value": "",
                "input_types": ["Tool"],
            },
        },
    ),
    "ChatOutput": _make_template(
        "Chat Output",
        outputs=[{"name": "message", "types": ["Message"]}],
        template_fields={
            "input_value": {
                "display_name": "Text",
                "type": "str",
                "value": "",
                "input_types": ["Message"],
            },
        },
    ),
    "OpenAIModel": _make_template(
        "OpenAI",
        outputs=[
            {"name": "text_output", "types": ["Message"]},
            {"name": "model_output", "types": ["LanguageModel"]},
        ],
        template_fields={
            "model_name": {
                "display_name": "Model Name",
                "type": "str",
                "value": "gpt-4o-mini",
                "real_time_refresh": True,
            },
            "temperature": {
                "display_name": "Temperature",
                "type": "float",
                "value": 0.1,
            },
            "input_value": {
                "display_name": "Input",
                "type": "str",
                "value": "",
                "input_types": ["Message"],
            },
        },
    ),
}


def _fresh_flow(name="Test Flow"):
    return empty_flow(name=name)


# ---------------------------------------------------------------------------
# Flow
# ---------------------------------------------------------------------------


class TestFlow:
    def test_empty_flow_structure(self):
        flow = empty_flow(name="My Flow", description="desc")
        assert flow["name"] == "My Flow"
        assert flow["description"] == "desc"
        assert flow["data"]["nodes"] == []
        assert flow["data"]["edges"] == []
        assert flow["data"]["viewport"] == {"x": 0, "y": 0, "zoom": 1}

    def test_flow_info_empty(self):
        flow = empty_flow(name="Empty")
        info = flow_info(flow)
        assert info["name"] == "Empty"
        assert info["node_count"] == 0
        assert info["edge_count"] == 0
        assert info["components"] == []

    def test_flow_info_with_components(self):
        flow = _fresh_flow()
        add_component(flow, "ChatInput", REGISTRY)
        add_component(flow, "ChatOutput", REGISTRY)
        info = flow_info(flow)
        assert info["node_count"] == 2
        assert len(info["components"]) == 2
        assert len(info["inputs"]) == 1
        assert len(info["outputs"]) == 1


# ---------------------------------------------------------------------------
# Component
# ---------------------------------------------------------------------------


class TestComponent:
    def test_add_component(self):
        flow = _fresh_flow()
        result = add_component(flow, "ChatInput", REGISTRY)
        assert result["id"].startswith("ChatInput-")
        assert result["display_name"] == "Chat Input"
        assert len(flow["data"]["nodes"]) == 1

    def test_add_component_with_explicit_id(self):
        flow = _fresh_flow()
        result = add_component(flow, "ChatInput", REGISTRY, component_id="ChatInput-fixed")
        assert result["id"] == "ChatInput-fixed"

    def test_add_unknown_component_raises(self):
        flow = _fresh_flow()
        with pytest.raises(ValueError, match="Unknown component"):
            add_component(flow, "TotallyFake", REGISTRY)

    def test_remove_component(self):
        flow = _fresh_flow()
        r = add_component(flow, "ChatInput", REGISTRY)
        assert len(flow["data"]["nodes"]) == 1
        remove_component(flow, r["id"])
        assert len(flow["data"]["nodes"]) == 0

    def test_remove_nonexistent_raises(self):
        flow = _fresh_flow()
        with pytest.raises(ValueError, match="Component not found"):
            remove_component(flow, "NonExistent-12345")

    def test_remove_component_also_removes_edges(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        assert len(flow["data"]["edges"]) == 1
        remove_component(flow, r1["id"])
        assert len(flow["data"]["edges"]) == 0

    def test_configure_component(self):
        flow = _fresh_flow()
        r = add_component(flow, "ChatInput", REGISTRY)
        configure_component(flow, r["id"], {"input_value": "Hello"})
        node = flow["data"]["nodes"][0]
        assert node["data"]["node"]["template"]["input_value"]["value"] == "Hello"

    def test_configure_nonexistent_raises(self):
        flow = _fresh_flow()
        with pytest.raises(ValueError, match="Component not found"):
            configure_component(flow, "NonExistent-12345", {"key": "val"})

    def test_configure_unknown_field_raises(self):
        flow = _fresh_flow()
        r = add_component(flow, "ChatInput", REGISTRY)
        with pytest.raises(ValueError, match="Unknown parameter"):
            configure_component(flow, r["id"], {"nonexistent_field": "val"})

    def test_get_component(self):
        flow = _fresh_flow()
        r = add_component(flow, "ChatInput", REGISTRY)
        info = get_component(flow, r["id"])
        assert info["id"] == r["id"]
        assert info["type"] == "ChatInput"
        assert info["display_name"] == "Chat Input"
        assert "input_value" in info["params"]

    def test_get_nonexistent_raises(self):
        flow = _fresh_flow()
        with pytest.raises(ValueError, match="Component not found"):
            get_component(flow, "NonExistent-12345")

    def test_list_components(self):
        flow = _fresh_flow()
        add_component(flow, "ChatInput", REGISTRY)
        add_component(flow, "ChatOutput", REGISTRY)
        comps = list_components(flow)
        assert len(comps) == 2
        types = {c["type"] for c in comps}
        assert types == {"ChatInput", "ChatOutput"}

    def test_unique_ids(self):
        flow = _fresh_flow()
        ids = set()
        for _ in range(10):
            r = add_component(flow, "ChatInput", REGISTRY)
            assert r["id"] not in ids
            ids.add(r["id"])

    def test_node_structure(self):
        flow = _fresh_flow()
        add_component(flow, "ChatInput", REGISTRY)
        node = flow["data"]["nodes"][0]
        assert node["type"] == "genericNode"
        assert node["position"] == {"x": 0, "y": 0}
        assert node["selected"] is False
        assert node["data"]["showNode"] is True
        assert "template" in node["data"]["node"]
        assert "outputs" in node["data"]["node"]


# ---------------------------------------------------------------------------
# NeedsServerUpdate
# ---------------------------------------------------------------------------


class TestNeedsServerUpdate:
    def test_real_time_refresh_triggers(self):
        template = {"agent_llm": {"real_time_refresh": True, "type": "str"}}
        assert needs_server_update(template, "agent_llm") is True

    def test_no_refresh_does_not_trigger(self):
        template = {"input_value": {"type": "str", "value": ""}}
        assert needs_server_update(template, "input_value") is False

    def test_tool_mode_always_triggers(self):
        assert needs_server_update({}, "tool_mode") is True

    def test_missing_field_does_not_trigger(self):
        assert needs_server_update({}, "nonexistent") is False

    def test_non_dict_field_does_not_trigger(self):
        template = {"some_field": "just a string"}
        assert needs_server_update(template, "some_field") is False


# ---------------------------------------------------------------------------
# Connect
# ---------------------------------------------------------------------------


class TestConnect:
    def test_add_connection(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        edge = add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        assert len(flow["data"]["edges"]) == 1
        assert edge["source"] == r1["id"]
        assert edge["target"] == r2["id"]

    def test_edge_has_reactflow_id(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        edge = add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        assert edge["id"].startswith("reactflow__edge-")

    def test_edge_handle_strings_use_oe(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        edge = add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        assert isinstance(edge["sourceHandle"], str)
        assert isinstance(edge["targetHandle"], str)
        assert "\u0153" in edge["sourceHandle"]
        assert "\u0153" in edge["targetHandle"]

    def test_edge_data_has_handle_dicts(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        edge = add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        assert isinstance(edge["data"]["sourceHandle"], dict)
        assert isinstance(edge["data"]["targetHandle"], dict)
        assert edge["data"]["sourceHandle"]["name"] == "message"
        assert edge["data"]["targetHandle"]["fieldName"] == "input_value"

    def test_edge_resolves_output_types(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        edge = add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        assert edge["data"]["sourceHandle"]["output_types"] == ["Message"]

    def test_edge_resolves_input_types(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        edge = add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        assert edge["data"]["targetHandle"]["inputTypes"] == ["Message"]

    def test_edge_animated_and_selected(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        edge = add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        assert edge["animated"] is False
        assert edge["selected"] is False

    def test_add_connection_is_idempotent(self):
        """Repeating add_connection for the same pair must not duplicate the edge.

        Regression for LE-866: connecting components via batch (or after the
        edge was already created in the UI) appended a second edge with the
        same connection, double-wiring the flow at runtime.
        """
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        first = add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        second = add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        assert len(flow["data"]["edges"]) == 1
        assert first is second  # returns the existing edge dict

    def test_add_connection_dedupes_against_ui_saved_edges(self):
        """Re-adding connections to a real UI-saved flow must not grow the edge list.

        Regression for LE-866 against an actual UI-exported fixture. UI-saved
        edges from older Langflow versions use the `xy-edge__` id prefix
        instead of `reactflow__edge-`, so dedup must be structural (source,
        target, handle name, handle fieldName) rather than by edge id.
        """
        fixture = Path(__file__).parent.parent / "data" / "MemoryChatbotNoLLM.json"
        flow = json.loads(fixture.read_text())
        original_count = len(flow["data"]["edges"])
        assert original_count > 0, "fixture should have edges to test against"

        # Replay every existing connection through add_connection — exactly what
        # batch -> connect_components does when called for an already-wired pair.
        for conn in list_connections(flow):
            add_connection(
                flow,
                conn["source_id"],
                conn["source_output"],
                conn["target_id"],
                conn["target_input"],
                source_types=conn["source_types"],
                target_types=conn["target_types"],
            )

        assert len(flow["data"]["edges"]) == original_count

    def test_remove_connection(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        removed = remove_connection(flow, r1["id"], r2["id"])
        assert removed == 1
        assert len(flow["data"]["edges"]) == 0

    def test_remove_connection_with_filter(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        removed = remove_connection(flow, r1["id"], r2["id"], source_output="message")
        assert removed == 1

    def test_remove_connection_no_match(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        removed = remove_connection(flow, r1["id"], r2["id"], source_output="nonexistent")
        assert removed == 0
        assert len(flow["data"]["edges"]) == 1

    def test_list_connections(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        conns = list_connections(flow)
        assert len(conns) == 1
        assert conns[0]["source_id"] == r1["id"]
        assert conns[0]["target_id"] == r2["id"]
        assert conns[0]["source_output"] == "message"
        assert conns[0]["target_input"] == "input_value"


# ---------------------------------------------------------------------------
# CustomStringify
# ---------------------------------------------------------------------------


class TestCustomStringify:
    def test_sorted_keys(self):
        result = _custom_stringify({"z": 1, "a": 2})
        assert result == '{"a":2,"z":1}'

    def test_nested_list(self):
        result = _custom_stringify({"list": ["a", "b"]})
        assert result == '{"list":["a","b"]}'

    def test_scaped_replaces_quotes(self):
        result = _scaped_json_stringify({"key": "val"})
        assert '"' not in result
        assert "\u0153" in result

    def test_bool(self):
        assert _custom_stringify(True) == "true"  # noqa: FBT003
        assert _custom_stringify(False) == "false"  # noqa: FBT003

    def test_null(self):
        assert _custom_stringify(None) == "null"

    def test_number(self):
        assert _custom_stringify(42) == "42"
        assert _custom_stringify(3.14) == "3.14"


# ---------------------------------------------------------------------------
# Layout
# ---------------------------------------------------------------------------


class TestLayout:
    def test_layout_empty_flow(self):
        flow = _fresh_flow()
        layout_flow(flow)  # should not error

    def test_layout_single_node(self):
        flow = _fresh_flow()
        add_component(flow, "ChatInput", REGISTRY)
        layout_flow(flow)
        pos = flow["data"]["nodes"][0]["position"]
        assert "x" in pos
        assert "y" in pos

    def test_layout_chain_assigns_layers(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "OpenAIModel", REGISTRY)
        r3 = add_component(flow, "ChatOutput", REGISTRY)
        add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        add_connection(flow, r2["id"], "text_output", r3["id"], "input_value")
        layout_flow(flow)

        by_id = {n["data"]["id"]: n for n in flow["data"]["nodes"]}
        x1 = by_id[r1["id"]]["position"]["x"]
        x2 = by_id[r2["id"]]["position"]["x"]
        x3 = by_id[r3["id"]]["position"]["x"]
        assert x1 < x2 < x3

    def test_layout_branch_same_x(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY, component_id="ChatOutput-aaa")
        r3 = add_component(flow, "ChatOutput", REGISTRY, component_id="ChatOutput-bbb")
        add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        add_connection(flow, r1["id"], "message", r3["id"], "input_value")
        layout_flow(flow)

        by_id = {n["data"]["id"]: n for n in flow["data"]["nodes"]}
        assert by_id[r2["id"]]["position"]["x"] == by_id[r3["id"]]["position"]["x"]

    def test_layout_disconnected_distinct(self):
        flow = _fresh_flow()
        add_component(flow, "ChatInput", REGISTRY)
        add_component(flow, "ChatOutput", REGISTRY)
        layout_flow(flow)

        positions = [(n["position"]["x"], n["position"]["y"]) for n in flow["data"]["nodes"]]
        assert positions[0] != positions[1]

    def test_layout_uses_spacing_constants(self):
        from lfx.graph.flow_builder.layout import LAYER_SPACING_X

        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        layout_flow(flow)

        by_id = {n["data"]["id"]: n for n in flow["data"]["nodes"]}
        x_diff = by_id[r2["id"]]["position"]["x"] - by_id[r1["id"]]["position"]["x"]
        assert x_diff == LAYER_SPACING_X

    def test_layout_diamond_graph(self):
        """A -> B, A -> C, B -> D, C -> D. D should be at layer 2."""
        from lfx.graph.flow_builder.layout import LAYER_SPACING_X

        flow = _fresh_flow()
        a = add_component(flow, "ChatInput", REGISTRY, component_id="A-00001")
        b = add_component(flow, "ChatOutput", REGISTRY, component_id="B-00001")
        c = add_component(flow, "ChatOutput", REGISTRY, component_id="C-00001")
        d = add_component(flow, "ChatOutput", REGISTRY, component_id="D-00001")
        add_connection(flow, a["id"], "message", b["id"], "input_value")
        add_connection(flow, a["id"], "message", c["id"], "input_value")
        add_connection(flow, b["id"], "message", d["id"], "input_value")
        add_connection(flow, c["id"], "message", d["id"], "input_value")
        layout_flow(flow)

        by_id = {n["data"]["id"]: n for n in flow["data"]["nodes"]}
        # A at layer 0, B and C at layer 1, D at layer 2
        assert by_id["A-00001"]["position"]["x"] == 0
        assert by_id["B-00001"]["position"]["x"] == LAYER_SPACING_X
        assert by_id["C-00001"]["position"]["x"] == LAYER_SPACING_X
        assert by_id["D-00001"]["position"]["x"] == 2 * LAYER_SPACING_X

    def test_layout_long_chain(self):
        """5-node linear chain: each node should be one layer further right."""
        from lfx.graph.flow_builder.layout import LAYER_SPACING_X

        flow = _fresh_flow()
        ids = []
        for i in range(5):
            r = add_component(flow, "ChatInput", REGISTRY, component_id=f"N-{i:05d}")
            ids.append(r["id"])
        for i in range(4):
            add_connection(flow, ids[i], "message", ids[i + 1], "input_value")
        layout_flow(flow)

        by_id = {n["data"]["id"]: n for n in flow["data"]["nodes"]}
        for i, nid in enumerate(ids):
            assert by_id[nid]["position"]["x"] == i * LAYER_SPACING_X

    def test_layout_y_spacing_multiple_at_same_layer(self):
        """3 nodes at the same layer should be spaced by NODE_SPACING_Y."""
        from lfx.graph.flow_builder.layout import NODE_SPACING_Y

        flow = _fresh_flow()
        src = add_component(flow, "ChatInput", REGISTRY, component_id="Src-00001")
        t1 = add_component(flow, "ChatOutput", REGISTRY, component_id="T1-00001")
        t2 = add_component(flow, "ChatOutput", REGISTRY, component_id="T2-00001")
        t3 = add_component(flow, "ChatOutput", REGISTRY, component_id="T3-00001")
        add_connection(flow, src["id"], "message", t1["id"], "input_value")
        add_connection(flow, src["id"], "message", t2["id"], "input_value")
        add_connection(flow, src["id"], "message", t3["id"], "input_value")
        layout_flow(flow)

        by_id = {n["data"]["id"]: n for n in flow["data"]["nodes"]}
        ys = sorted([by_id[t["id"]]["position"]["y"] for t in [t1, t2, t3]])
        # Check spacing between consecutive nodes
        assert ys[1] - ys[0] == NODE_SPACING_Y
        assert ys[2] - ys[1] == NODE_SPACING_Y

    def test_layout_idempotent(self):
        """Running layout twice produces the same positions."""
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        add_connection(flow, r1["id"], "message", r2["id"], "input_value")

        layout_flow(flow)
        positions_1 = [(n["data"]["id"], n["position"].copy()) for n in flow["data"]["nodes"]]

        layout_flow(flow)
        positions_2 = [(n["data"]["id"], n["position"].copy()) for n in flow["data"]["nodes"]]

        assert positions_1 == positions_2

    def test_layout_single_node_centered_y(self):
        """A single node in a layer should be at y=0."""
        flow = _fresh_flow()
        add_component(flow, "ChatInput", REGISTRY)
        layout_flow(flow)
        assert flow["data"]["nodes"][0]["position"]["y"] == 0.0

    def test_layout_two_disconnected_subgraphs(self):
        """Two independent chains should both get correct layering."""
        from lfx.graph.flow_builder.layout import LAYER_SPACING_X

        flow = _fresh_flow()
        # Chain 1: A -> B
        a = add_component(flow, "ChatInput", REGISTRY, component_id="A-00001")
        b = add_component(flow, "ChatOutput", REGISTRY, component_id="B-00001")
        add_connection(flow, a["id"], "message", b["id"], "input_value")
        # Chain 2: C -> D
        c = add_component(flow, "ChatInput", REGISTRY, component_id="C-00001")
        d = add_component(flow, "ChatOutput", REGISTRY, component_id="D-00001")
        add_connection(flow, c["id"], "message", d["id"], "input_value")
        layout_flow(flow)

        by_id = {n["data"]["id"]: n for n in flow["data"]["nodes"]}
        # Both chains: source at layer 0, target at layer 1
        assert by_id["A-00001"]["position"]["x"] == 0
        assert by_id["B-00001"]["position"]["x"] == LAYER_SPACING_X
        assert by_id["C-00001"]["position"]["x"] == 0
        assert by_id["D-00001"]["position"]["x"] == LAYER_SPACING_X

    def test_assign_layers_directly(self):
        """Test _assign_layers with a known graph."""
        from lfx.graph.flow_builder.layout import _assign_layers

        node_ids = ["A", "B", "C", "D"]
        successors = {"A": ["B", "C"], "B": ["D"], "C": ["D"]}
        predecessors = {"B": ["A"], "C": ["A"], "D": ["B", "C"]}
        layers = _assign_layers(node_ids, successors, predecessors)
        assert layers["A"] == 0
        assert layers["B"] == 1
        assert layers["C"] == 1
        assert layers["D"] == 2


# ---------------------------------------------------------------------------
# Component — additional tests
# ---------------------------------------------------------------------------


class TestComponentExtended:
    def test_configure_overwrites_existing_value(self):
        flow = _fresh_flow()
        r = add_component(flow, "OpenAIModel", REGISTRY)
        # Template starts with value "gpt-4o-mini"
        info = get_component(flow, r["id"])
        assert info["params"]["model_name"] == "gpt-4o-mini"
        # Overwrite
        configure_component(flow, r["id"], {"model_name": "gpt-4o"})
        info2 = get_component(flow, r["id"])
        assert info2["params"]["model_name"] == "gpt-4o"

    def test_add_component_deep_copies_template(self):
        """Modifying a node should not affect the registry or other nodes."""
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatInput", REGISTRY)
        configure_component(flow, r1["id"], {"input_value": "modified"})
        info2 = get_component(flow, r2["id"])
        assert info2["params"]["input_value"] == ""  # unmodified

    def test_list_components_empty_flow(self):
        flow = _fresh_flow()
        assert list_components(flow) == []

    def test_get_component_outputs(self):
        flow = _fresh_flow()
        r = add_component(flow, "OpenAIModel", REGISTRY)
        info = get_component(flow, r["id"])
        output_names = [o["name"] for o in info["outputs"]]
        assert "text_output" in output_names
        assert "model_output" in output_names


# ---------------------------------------------------------------------------
# Connect — additional tests
# ---------------------------------------------------------------------------


class TestConnectErrorCases:
    def test_connect_nonexistent_source_raises(self):
        flow = _fresh_flow()
        add_component(flow, "ChatOutput", REGISTRY, component_id="Target-00001")
        with pytest.raises(ValueError, match="Component not found in flow"):
            add_connection(flow, "NoSuch-00001", "message", "Target-00001", "input_value")

    def test_connect_nonexistent_target_raises(self):
        flow = _fresh_flow()
        add_component(flow, "ChatInput", REGISTRY, component_id="Source-00001")
        with pytest.raises(ValueError, match="Component not found in flow"):
            add_connection(flow, "Source-00001", "message", "NoSuch-00001", "input_value")

    def test_connect_nonexistent_output_raises(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        with pytest.raises(ValueError, match="Output 'bad_output' not found"):
            add_connection(flow, r1["id"], "bad_output", r2["id"], "input_value")

    def test_connect_nonexistent_input_raises(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        with pytest.raises(ValueError, match="Input 'bad_input' not found"):
            add_connection(flow, r1["id"], "message", r2["id"], "bad_input")

    def test_connect_incompatible_types_raises(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ToolOutput", REGISTRY)
        with pytest.raises(ValueError, match="Type mismatch"):
            add_connection(flow, r1["id"], "message", r2["id"], "tools")
        assert len(flow["data"]["edges"]) == 0

    def test_explicit_types_bypass_validation(self):
        """When explicit types are provided, type-compatibility is not enforced."""
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        # Incompatible explicit types -- succeeds because validation is skipped
        edge = add_connection(
            flow,
            r1["id"],
            "message",
            r2["id"],
            "input_value",
            source_types=["Incompatible"],
            target_types=["Mismatch"],
        )
        assert edge["data"]["sourceHandle"]["output_types"] == ["Incompatible"]
        assert edge["data"]["targetHandle"]["inputTypes"] == ["Mismatch"]


class TestConnectExtended:
    def test_explicit_types_override(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        edge = add_connection(
            flow,
            r1["id"],
            "message",
            r2["id"],
            "input_value",
            source_types=["CustomType"],
            target_types=["OtherType"],
        )
        assert edge["data"]["sourceHandle"]["output_types"] == ["CustomType"]
        assert edge["data"]["targetHandle"]["inputTypes"] == ["OtherType"]

    def test_multiple_connections_between_same_pair(self):
        """Two different output->input connections between the same components."""
        flow = _fresh_flow()
        r1 = add_component(flow, "OpenAIModel", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        add_connection(flow, r1["id"], "text_output", r2["id"], "input_value")
        # This would be unusual but should work structurally
        add_connection(
            flow,
            r1["id"],
            "model_output",
            r2["id"],
            "input_value",
            source_types=["LanguageModel"],
        )
        assert len(flow["data"]["edges"]) == 2
        conns = list_connections(flow)
        outputs = {c["source_output"] for c in conns}
        assert outputs == {"text_output", "model_output"}

    def test_remove_connection_by_target_input(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        removed = remove_connection(flow, r1["id"], r2["id"], target_input="input_value")
        assert removed == 1

    def test_list_connections_empty(self):
        flow = _fresh_flow()
        assert list_connections(flow) == []

    def test_source_type_extracted_from_id(self):
        """The source handle dataType should be the component type prefix."""
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY, component_id="ChatInput-abc12")
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        edge = add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        assert edge["data"]["sourceHandle"]["dataType"] == "ChatInput"


# ---------------------------------------------------------------------------
# Flow — additional tests
# ---------------------------------------------------------------------------


class TestFlowExtended:
    def test_flow_info_with_edges(self):
        flow = _fresh_flow()
        r1 = add_component(flow, "ChatInput", REGISTRY)
        r2 = add_component(flow, "ChatOutput", REGISTRY)
        add_connection(flow, r1["id"], "message", r2["id"], "input_value")
        info = flow_info(flow)
        assert info["edge_count"] == 1

    def test_flow_info_description(self):
        flow = empty_flow(name="Named", description="A description")
        info = flow_info(flow)
        assert info["description"] == "A description"

    def test_flow_info_no_inputs_outputs_for_model(self):
        """OpenAIModel is not a ChatInput or ChatOutput."""
        flow = _fresh_flow()
        add_component(flow, "OpenAIModel", REGISTRY)
        info = flow_info(flow)
        assert info["inputs"] == []
        assert info["outputs"] == []
        assert len(info["components"]) == 1


# ---------------------------------------------------------------------------
# CustomStringify — additional tests
# ---------------------------------------------------------------------------


class TestCustomStringifyExtended:
    def test_empty_dict(self):
        assert _custom_stringify({}) == "{}"

    def test_empty_list(self):
        assert _custom_stringify([]) == "[]"

    def test_nested_dict(self):
        result = _custom_stringify({"a": {"b": 1}})
        assert result == '{"a":{"b":1}}'

    def test_string_value(self):
        assert _custom_stringify("hello") == '"hello"'

    def test_scaped_round_trips_handle_format(self):
        """A source handle dict should produce a string that matches Langflow's format."""
        handle = {
            "dataType": "ChatInput",
            "id": "ChatInput-abc12",
            "name": "message",
            "output_types": ["Message"],
        }
        result = _scaped_json_stringify(handle)
        # Should have no double quotes
        assert '"' not in result
        # Should contain the field names with oe
        assert "dataType" in result.replace("\u0153", '"')
        assert "output_types" in result.replace("\u0153", '"')


# ---------------------------------------------------------------------------
# parse_flow_spec — text parser tests
# ---------------------------------------------------------------------------


class TestParseFlowSpec:
    def test_basic_spec(self):
        spec = """\
name: Test Flow
description: A test

nodes:
  A: ChatInput
  B: ChatOutput

edges:
  A.message -> B.input_value
"""
        result = parse_flow_spec(spec)
        assert result["name"] == "Test Flow"
        assert result["description"] == "A test"
        assert len(result["nodes"]) == 2
        assert result["nodes"][0] == {"id": "A", "type": "ChatInput"}
        assert result["nodes"][1] == {"id": "B", "type": "ChatOutput"}
        assert len(result["edges"]) == 1
        assert result["edges"][0] == {
            "source_id": "A",
            "source_output": "message",
            "target_id": "B",
            "target_input": "input_value",
        }

    def test_no_description(self):
        spec = """\
name: Minimal

nodes:
  A: ChatInput
"""
        result = parse_flow_spec(spec)
        assert result["name"] == "Minimal"
        assert result["description"] == ""

    def test_no_edges(self):
        spec = """\
name: No Edges

nodes:
  A: ChatInput
  B: ChatOutput
"""
        result = parse_flow_spec(spec)
        assert len(result["nodes"]) == 2
        assert len(result["edges"]) == 0

    def test_config_simple_values(self):
        spec = """\
name: Config Test

nodes:
  A: ChatInput

config:
  A.input_value: Hello world
  A.sender_name: Bot
"""
        result = parse_flow_spec(spec)
        assert result["config"] == {"A": {"input_value": "Hello world", "sender_name": "Bot"}}

    def test_config_multiline(self):
        spec = """\
name: Multiline

nodes:
  A: ChatInput

config:
  A.system_prompt: |
    You are a helpful assistant.
    Be concise.
"""
        result = parse_flow_spec(spec)
        assert "You are a helpful assistant." in result["config"]["A"]["system_prompt"]
        assert "Be concise." in result["config"]["A"]["system_prompt"]

    def test_config_multiple_nodes(self):
        spec = """\
name: Multi Config

nodes:
  A: ChatInput
  B: OpenAIModel

config:
  A.input_value: test
  B.model_name: gpt-4o
  B.temperature: 0.5
"""
        result = parse_flow_spec(spec)
        assert result["config"]["A"] == {"input_value": "test"}
        assert result["config"]["B"] == {"model_name": "gpt-4o", "temperature": 0.5}

    def test_multiple_edges(self):
        spec = """\
name: Chain

nodes:
  A: ChatInput
  B: OpenAIModel
  C: ChatOutput

edges:
  A.message -> B.input_value
  B.text_output -> C.input_value
"""
        result = parse_flow_spec(spec)
        assert len(result["edges"]) == 2
        assert result["edges"][0]["source_id"] == "A"
        assert result["edges"][0]["target_id"] == "B"
        assert result["edges"][1]["source_id"] == "B"
        assert result["edges"][1]["target_id"] == "C"

    def test_tool_mode_edge(self):
        spec = """\
name: Agent

nodes:
  A: URLComponent
  B: Agent

edges:
  A.component_as_tool -> B.tools
"""
        result = parse_flow_spec(spec)
        assert result["edges"][0]["source_output"] == "component_as_tool"
        assert result["edges"][0]["target_input"] == "tools"

    def test_empty_nodes_raises(self):
        spec = """\
name: Empty
"""
        with pytest.raises(ValueError, match="No nodes found"):
            parse_flow_spec(spec)

    def test_blank_lines_between_sections(self):
        spec = """\
name: Spaced



nodes:

  A: ChatInput

  B: ChatOutput


edges:

  A.message -> B.input_value

"""
        result = parse_flow_spec(spec)
        assert len(result["nodes"]) == 2
        assert len(result["edges"]) == 1

    def test_leading_trailing_whitespace(self):
        spec = """
    name: Indented

    nodes:
      A: ChatInput

    edges:
      A.message -> A.input_value
"""
        result = parse_flow_spec(spec)
        assert result["name"] == "Indented"

    def test_many_nodes(self):
        lines = ["name: Big", "", "nodes:"]
        lines.extend(f"  N{i}: ChatInput" for i in range(10))
        spec = "\n".join(lines)
        result = parse_flow_spec(spec)
        assert len(result["nodes"]) == 10

    def test_config_preserves_multiline_indentation(self):
        spec = """\
name: Indent Test

nodes:
  A: ChatInput

config:
  A.template: |
    Line 1
      Indented line 2
    Line 3
"""
        result = parse_flow_spec(spec)
        tmpl = result["config"]["A"]["template"]
        assert "  Indented line 2" in tmpl

    def test_edge_with_underscores_in_names(self):
        spec = """\
name: Underscores

nodes:
  A: OpenAIModel
  B: Agent

edges:
  A.model_output -> B.model
"""
        result = parse_flow_spec(spec)
        assert result["edges"][0]["source_output"] == "model_output"

    def test_roundtrip_node_ids_preserved(self):
        spec = """\
name: IDs

nodes:
  X: ChatInput
  Y: ChatOutput
  Z: OpenAIModel
"""
        result = parse_flow_spec(spec)
        ids = [n["id"] for n in result["nodes"]]
        assert ids == ["X", "Y", "Z"]

    def test_config_multiline_with_dot_and_colon(self):
        """Continuation lines with dots and colons must not be mistaken for keys."""
        spec = """\
name: DotColon Test

nodes:
  A: ChatInput

config:
  A.system_prompt: |
    You are an assistant. Follow these rules: be polite.
    Also see http://example.com: it has info.
"""
        result = parse_flow_spec(spec)
        prompt = result["config"]["A"]["system_prompt"]
        assert "You are an assistant" in prompt
        assert "Also see http://example.com" in prompt

    def test_config_coerces_int(self):
        spec = """\
name: Coerce

nodes:
  A: ChatInput

config:
  A.max_tokens: 100
"""
        result = parse_flow_spec(spec)
        assert result["config"]["A"]["max_tokens"] == 100
        assert isinstance(result["config"]["A"]["max_tokens"], int)

    def test_config_coerces_float(self):
        spec = """\
name: Coerce

nodes:
  A: ChatInput

config:
  A.temperature: 0.7
"""
        result = parse_flow_spec(spec)
        assert result["config"]["A"]["temperature"] == 0.7
        assert isinstance(result["config"]["A"]["temperature"], float)

    def test_config_coerces_bool(self):
        spec = """\
name: Coerce

nodes:
  A: ChatInput

config:
  A.stream: true
  A.verbose: False
"""
        result = parse_flow_spec(spec)
        assert result["config"]["A"]["stream"] is True
        assert result["config"]["A"]["verbose"] is False

    def test_config_multiline_stays_string(self):
        """Multi-line values (with |) should not be coerced."""
        spec = """\
name: NoCoerce

nodes:
  A: ChatInput

config:
  A.template: |
    42
"""
        result = parse_flow_spec(spec)
        assert result["config"]["A"]["template"] == "42"
        assert isinstance(result["config"]["A"]["template"], str)

    def test_config_coerces_null(self):
        spec = """\
name: Null

nodes:
  A: ChatInput

config:
  A.value: null
"""
        result = parse_flow_spec(spec)
        assert result["config"]["A"]["value"] is None

    def test_edge_unknown_source_in_spec(self):
        """Parser doesn't validate refs, but the structure should be correct."""
        spec = """\
name: Bad Edge

nodes:
  A: ChatInput

edges:
  Z.message -> A.input_value
"""
        result = parse_flow_spec(spec)
        assert result["edges"][0]["source_id"] == "Z"

    def test_config_key_with_long_field_name(self):
        spec = """\
name: Long

nodes:
  A: ChatInput

config:
  A.very_long_field_name_with_underscores: hello
"""
        result = parse_flow_spec(spec)
        assert result["config"]["A"]["very_long_field_name_with_underscores"] == "hello"

    def test_multiline_then_single_line(self):
        """A multi-line value followed by a single-line value."""
        spec = """\
name: Mixed

nodes:
  A: ChatInput

config:
  A.template: |
    Line 1
    Line 2
  A.name: simple
"""
        result = parse_flow_spec(spec)
        assert "Line 1" in result["config"]["A"]["template"]
        assert "Line 2" in result["config"]["A"]["template"]
        assert result["config"]["A"]["name"] == "simple"

    def test_config_multiline_with_key_like_pattern(self):
        """Continuation line that looks like a config key (word.word: value)."""
        spec = """\
name: KeyLike

nodes:
  A: ChatInput

config:
  A.prompt: |
    hello.world: this should stay in the prompt
    config.timeout: 30 seconds
"""
        result = parse_flow_spec(spec)
        prompt = result["config"]["A"]["prompt"]
        assert "hello.world: this should stay" in prompt
        assert "config.timeout: 30" in prompt
        assert len(result["config"]) == 1  # only A, no hello or config node

    def test_sections_in_any_order(self):
        """Config before edges should work."""
        spec = """\
name: Reordered

nodes:
  A: ChatInput
  B: ChatOutput

config:
  A.input_value: test

edges:
  A.message -> B.input_value
"""
        result = parse_flow_spec(spec)
        assert result["config"]["A"]["input_value"] == "test"
        assert len(result["edges"]) == 1
