440 lines
15 KiB
Python
440 lines
15 KiB
Python
"""Phase 3 tests: form-schema extraction, sync cache, input-schema endpoint."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from app.flows import input_schema as parser
|
|
from app.flows import sync
|
|
|
|
from .conftest import FIXTURES_DIR as FIXTURES
|
|
|
|
|
|
def _field(name, itype, **kw):
|
|
base = {
|
|
"name": name,
|
|
"display_name": name.replace("_", " ").title(),
|
|
"info": "",
|
|
"value": None,
|
|
"required": False,
|
|
"advanced": False,
|
|
"show": True,
|
|
"_input_type": itype,
|
|
}
|
|
base.update(kw)
|
|
return base
|
|
|
|
|
|
def _doc(*fields, node_type="CustomInputForm", display="CustomInput Form"):
|
|
template = {f["name"]: f for f in fields}
|
|
node = {
|
|
"data": {
|
|
"type": node_type,
|
|
"node": {"display_name": display, "template": template},
|
|
}
|
|
}
|
|
return {"data": {"nodes": [node], "edges": []}}
|
|
|
|
|
|
def _dropdown(name, options, **kw):
|
|
return _field(name, "DropdownInput", options=options, **kw)
|
|
|
|
|
|
def _branch_schema(doc):
|
|
schema = parser.extract_form_schema(doc)["json_schema"]
|
|
assert "allOf" in schema
|
|
return schema
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("itype", "kwargs", "json_type", "widget"),
|
|
[
|
|
("StrInput", {"value": "x"}, "string", None),
|
|
("MessageTextInput", {}, "string", None),
|
|
("MultilineInput", {}, "string", "textarea"),
|
|
("SecretStrInput", {}, "string", "password"),
|
|
("IntInput", {"value": 3}, "integer", None),
|
|
("FloatInput", {"value": 1.5}, "number", None),
|
|
("BoolInput", {"value": True}, "boolean", None),
|
|
(
|
|
"DropdownInput",
|
|
{"options": ["Music", "Youth"], "value": "Youth"},
|
|
"string",
|
|
None,
|
|
),
|
|
(
|
|
"MultiselectInput",
|
|
{"options": ["Worship", "Courses"], "value": ["Worship"]},
|
|
"array",
|
|
None,
|
|
),
|
|
(
|
|
"SliderInput",
|
|
{"value": 3, "range_spec": {"min": 1, "max": 5, "step": 1}},
|
|
"number",
|
|
"range",
|
|
),
|
|
("FileInput", {}, "string", None),
|
|
("FutureFancyInput", {}, "string", None),
|
|
],
|
|
)
|
|
def test_type_mapping_table(itype, kwargs, json_type, widget):
|
|
doc = _doc(_field("a", itype, **kwargs))
|
|
schema = parser.extract_form_schema(doc)
|
|
assert "fallback" not in schema
|
|
prop = schema["json_schema"]["properties"]["a"]
|
|
assert prop["type"] == json_type
|
|
ui = schema["ui_schema"].get("a", {})
|
|
if widget is None:
|
|
assert "ui:widget" not in ui
|
|
else:
|
|
assert ui["ui:widget"] == widget
|
|
|
|
|
|
def test_type_mapping_details():
|
|
doc = _doc(
|
|
_field("a", "StrInput", value="x"),
|
|
_field("e", "IntInput", value=3),
|
|
_field("g", "BoolInput", value=True),
|
|
_field(
|
|
"h",
|
|
"DropdownInput",
|
|
options=["Music", "Youth"],
|
|
value="Youth",
|
|
),
|
|
_field(
|
|
"i",
|
|
"MultiselectInput",
|
|
options=["Worship", "Courses"],
|
|
value=["Worship"],
|
|
),
|
|
_field(
|
|
"j",
|
|
"SliderInput",
|
|
value=3,
|
|
range_spec={"min": 1, "max": 5, "step": 1},
|
|
),
|
|
)
|
|
schema = parser.extract_form_schema(doc)
|
|
props = schema["json_schema"]["properties"]
|
|
assert props["a"] == {"title": "A", "type": "string", "default": "x"}
|
|
assert props["e"] == {"title": "E", "type": "integer", "default": 3}
|
|
assert props["g"] == {"title": "G", "type": "boolean", "default": True}
|
|
assert props["h"] == {
|
|
"title": "H",
|
|
"type": "string",
|
|
"enum": ["Music", "Youth"],
|
|
"default": "Youth",
|
|
}
|
|
assert props["i"]["type"] == "array"
|
|
assert props["i"]["items"] == {"type": "string", "enum": ["Worship", "Courses"]}
|
|
assert props["i"]["uniqueItems"] is True
|
|
assert props["j"]["minimum"] == 1
|
|
assert props["j"]["maximum"] == 5
|
|
assert props["j"]["multipleOf"] == 1
|
|
|
|
|
|
def test_dropdown_default_outside_enum_omitted():
|
|
doc = _doc(_field("h", "DropdownInput", options=["Music"], value=""))
|
|
props = parser.extract_form_schema(doc)["json_schema"]["properties"]
|
|
assert props["h"] == {"title": "H", "type": "string", "enum": ["Music"]}
|
|
|
|
|
|
def test_required_advanced_hidden_code():
|
|
doc = _doc(
|
|
_field("a", "StrInput", required=True),
|
|
_field("b", "StrInput", advanced=True),
|
|
_field("c", "StrInput", show=False),
|
|
)
|
|
doc["data"]["nodes"][0]["data"]["node"]["template"]["code"] = {"value": "x"}
|
|
schema = parser.extract_form_schema(doc)
|
|
assert schema["json_schema"]["required"] == ["a"]
|
|
assert set(schema["json_schema"]["properties"]) == {"a", "b"}
|
|
order = schema["ui_schema"]["ui:order"]
|
|
assert order[-1] == "*"
|
|
assert set(order[:-1]) == {"a", "b"}
|
|
|
|
|
|
def test_display_name_match_and_fallback():
|
|
doc = _doc(_field("a", "StrInput"), node_type="Custom", display="CustomInput X")
|
|
assert set(parser.extract_form_schema(doc)["json_schema"]["properties"]) == {"a"}
|
|
plain = {
|
|
"data": {
|
|
"nodes": [{"data": {"type": "ChatInput", "node": {"template": {}}}}],
|
|
"edges": [],
|
|
}
|
|
}
|
|
schema = parser.extract_form_schema(plain)
|
|
assert schema["fallback"] == "textarea"
|
|
assert schema["json_schema"]["properties"] == {}
|
|
|
|
|
|
def test_real_intake_fixture_parses():
|
|
doc = json.loads((FIXTURES / "intake_form.json").read_text())
|
|
schema = parser.extract_form_schema(doc)
|
|
props = schema["json_schema"]["properties"]
|
|
assert "fallback" not in schema
|
|
assert set(props) == {
|
|
"full_name",
|
|
"notes",
|
|
"password",
|
|
"age",
|
|
"amount",
|
|
"newsletter",
|
|
"department",
|
|
"topics",
|
|
"satisfaction",
|
|
"attachment",
|
|
}
|
|
assert "full_name" in schema["json_schema"]["required"]
|
|
assert props["department"]["enum"] == ["Music", "Youth", "Seniors", "Admin"]
|
|
assert schema["ui_schema"]["notes"] == {"ui:widget": "textarea"}
|
|
|
|
|
|
async def test_sync_caches_schema_per_version(session, lfx_server, lfx_client, tmp_path):
|
|
(tmp_path / "intake_form.json").write_text((FIXTURES / "intake_form.json").read_text())
|
|
report = await sync.sync_from_directory(session, lfx_client, tmp_path)
|
|
assert not report.errors, report.errors
|
|
from app.flows import store
|
|
|
|
defs = store.list_definitions(session)
|
|
definition = next(d for d in defs if d.name == "intake_form")
|
|
versions = store.list_versions(session, definition.id)
|
|
assert len(versions) == 1
|
|
cached = versions[0].input_schema
|
|
assert cached["json_schema"]["properties"]["full_name"]["type"] == "string"
|
|
assert "form_values" not in cached["json_schema"]["properties"]
|
|
|
|
|
|
async def test_input_schema_endpoint(client, lfx_server, login):
|
|
assert (await client.get(f"/api/flows/{uuid4()}/input-schema")).status_code == 404
|
|
|
|
await login()
|
|
doc = json.loads((FIXTURES / "intake_form.json").read_text())
|
|
created = await client.post("/api/flows", json={"name": "schema-e2e"})
|
|
flow_id = created.json()["id"]
|
|
version = await client.post(f"/api/flows/{flow_id}/versions", json={"langflow_json": doc})
|
|
assert version.status_code == 201, version.text
|
|
assert version.json()["input_schema"]["json_schema"]["properties"]["age"]["type"] == "integer"
|
|
# Created but never deployed → 404.
|
|
assert (await client.get(f"/api/flows/{flow_id}/input-schema")).status_code == 404
|
|
|
|
deployed = await client.post(f"/api/flows/{flow_id}/versions/1/deploy")
|
|
assert deployed.status_code == 200, deployed.text
|
|
response = await client.get(f"/api/flows/{flow_id}/input-schema")
|
|
assert response.status_code == 200, response.text
|
|
body = response.json()
|
|
assert body["version_no"] == 1
|
|
assert body["fallback"] is None
|
|
assert set(body["json_schema"]["properties"]) >= {"full_name", "topics"}
|
|
assert body["ui_schema"]["password"] == {"ui:widget": "password"}
|
|
|
|
|
|
def test_free_rename_keys_kept_verbatim():
|
|
doc = _doc(
|
|
_field("name", "StrInput", display_name="Name", required=True),
|
|
_field("favorite room", "StrInput"),
|
|
)
|
|
schema = parser.extract_form_schema(doc)
|
|
assert set(schema["json_schema"]["properties"]) == {"name", "favorite room"}
|
|
assert schema["json_schema"]["required"] == ["name"]
|
|
|
|
|
|
def test_empty_name_skipped_and_collision_suffixed():
|
|
a = _field("a", "StrInput")
|
|
clash = _field("other_key", "StrInput")
|
|
clash["name"] = "a" # Langflow rename collision: second field gets a suffix
|
|
empty = _field("zzz", "StrInput")
|
|
empty["name"] = " "
|
|
node = {
|
|
"data": {
|
|
"type": "CustomInputForm",
|
|
"node": {
|
|
"display_name": "CustomInput Form",
|
|
"template": {"a": a, "other_key": clash, "zzz": empty},
|
|
},
|
|
}
|
|
}
|
|
doc = {"data": {"nodes": [node], "edges": []}}
|
|
props = parser.extract_form_schema(doc)["json_schema"]["properties"]
|
|
assert set(props) == {"a", "a_2"}
|
|
|
|
|
|
def test_visible_if_single_branch():
|
|
doc = _doc(
|
|
_dropdown("gebaeude", ["Kirche", "Halle"], required=True),
|
|
_field(
|
|
"bestandteil",
|
|
"MultiselectInput",
|
|
options=["Glocke", "Orgel"],
|
|
info="Bitte wählen [visible_if gebaeude=Kirche]",
|
|
required=True,
|
|
),
|
|
)
|
|
out = parser.extract_form_schema(doc)
|
|
schema = out["json_schema"]
|
|
# Conditional field lives only in its branch, not at root.
|
|
assert set(schema["properties"]) == {"gebaeude"}
|
|
assert schema["required"] == ["gebaeude"]
|
|
assert len(schema["allOf"]) == 1
|
|
rule = schema["allOf"][0]
|
|
assert rule["if"] == {
|
|
"properties": {"gebaeude": {"const": "Kirche"}},
|
|
"required": ["gebaeude"],
|
|
}
|
|
assert rule["then"]["properties"]["bestandteil"]["type"] == "array"
|
|
assert rule["then"]["required"] == ["bestandteil"]
|
|
# Marker stripped from the shown description.
|
|
assert schema["properties"]["gebaeude"]["title"] == "Gebaeude"
|
|
order = out["ui_schema"]["ui:order"]
|
|
assert order[-1] == "*"
|
|
assert set(order[:-1]) == {"gebaeude", "bestandteil"}
|
|
|
|
|
|
def test_visible_if_nested_three_levels():
|
|
doc = _doc(
|
|
_dropdown("a", ["x", "y"]),
|
|
_dropdown("b", ["p", "q"], info="[visible_if a=x]"),
|
|
_field("c", "IntInput", info="[visible_if b=p]"),
|
|
_field("d", "StrInput", info="[visible_if c=3]"),
|
|
)
|
|
schema = _branch_schema(doc)
|
|
assert set(schema["properties"]) == {"a"}
|
|
assert len(schema["allOf"]) == 3
|
|
by_field = {next(iter(rule["then"]["properties"])): rule for rule in schema["allOf"]}
|
|
assert by_field["b"]["if"] == {
|
|
"properties": {"a": {"const": "x"}},
|
|
"required": ["a"],
|
|
}
|
|
assert by_field["c"]["if"]["properties"] == {"b": {"const": "p"}}
|
|
assert by_field["d"]["if"]["properties"] == {"c": {"const": 3}}
|
|
|
|
|
|
def test_visible_if_and_quoted_bool_multiselect():
|
|
doc = _doc(
|
|
_dropdown("a", ["x", "y"]),
|
|
_field("flag", "BoolInput"),
|
|
_field(
|
|
"m",
|
|
"MultiselectInput",
|
|
options=["Glocke", "Orgel"],
|
|
info='[visible_if a="x" & flag=true]',
|
|
),
|
|
)
|
|
schema = _branch_schema(doc)
|
|
rule = schema["allOf"][0]
|
|
assert rule["if"] == {
|
|
"properties": {"a": {"const": "x"}, "flag": {"const": True}},
|
|
"required": ["a", "flag"],
|
|
}
|
|
assert rule["then"]["properties"]["m"]["type"] == "array"
|
|
|
|
|
|
def test_visible_if_multiselect_trigger_uses_contains():
|
|
doc = _doc(
|
|
_field("tags", "MultiselectInput", options=["Glocke", "Orgel"]),
|
|
_field("detail", "StrInput", info="[visible_if tags=Glocke]"),
|
|
)
|
|
schema = _branch_schema(doc)
|
|
assert schema["allOf"][0]["if"]["properties"] == {"tags": {"contains": {"const": "Glocke"}}}
|
|
|
|
|
|
def test_visible_if_dangling_unsupported_cycle_fail_open():
|
|
doc = _doc(
|
|
_dropdown("a", ["x", "y"]),
|
|
_field("ok", "StrInput", info="[visible_if a=x]"),
|
|
_field("dangling", "StrInput", info="[visible_if nope=x]"),
|
|
_field("slider_dep", "StrInput", info="[visible_if slid=x]"),
|
|
_field(
|
|
"slid",
|
|
"SliderInput",
|
|
value=3,
|
|
range_spec={"min": 1, "max": 5, "step": 1},
|
|
),
|
|
_field("c1", "StrInput", info="[visible_if c2=x]"),
|
|
_field("c2", "StrInput", info="[visible_if c1=x]"),
|
|
_field("selfish", "StrInput", info="[visible_if selfish=x]"),
|
|
)
|
|
schema = parser.extract_form_schema(doc)["json_schema"]
|
|
# Only the valid marker produces a branch; the rest stay at root.
|
|
assert len(schema["allOf"]) == 1
|
|
assert set(schema["properties"]) >= {
|
|
"a",
|
|
"dangling",
|
|
"slider_dep",
|
|
"slid",
|
|
"c1",
|
|
"c2",
|
|
"selfish",
|
|
}
|
|
assert "ok" not in schema["properties"]
|
|
|
|
|
|
def test_no_markers_no_allof():
|
|
doc = _doc(_dropdown("a", ["x", "y"]))
|
|
schema = parser.extract_form_schema(doc)["json_schema"]
|
|
assert "allOf" not in schema
|
|
assert set(schema["properties"]) == {"a"}
|
|
|
|
|
|
def test_custom_component_form_display_match():
|
|
"""User-built * Form components parse without a kirche base class."""
|
|
doc = _doc(
|
|
_field("mitarbeiter_name", "StrInput", info="Vor- und Nachname"),
|
|
_dropdown("rolle", ["Assistenz", "Pfarrer:in"]),
|
|
node_type="CustomComponent",
|
|
display="Mitarbeiter Einstellen Form",
|
|
)
|
|
schema = parser.extract_form_schema(doc)
|
|
assert "fallback" not in schema
|
|
props = schema["json_schema"]["properties"]
|
|
assert set(props) == {"mitarbeiter_name", "rolle"}
|
|
assert props["rolle"]["enum"] == ["Assistenz", "Pfarrer:in"]
|
|
assert props["mitarbeiter_name"]["description"] == "Vor- und Nachname"
|
|
|
|
|
|
def test_custom_component_non_form_falls_back():
|
|
doc = _doc(
|
|
_field("payload", "StrInput"),
|
|
node_type="CustomComponent",
|
|
display="Rule Check",
|
|
)
|
|
assert parser.extract_form_schema(doc)["fallback"] == "textarea"
|
|
|
|
|
|
def test_real_mitarbeiter_file_parses():
|
|
path = Path(__file__).parent.parent / "flows" / "mitarbeiter_einstellen.json"
|
|
doc = json.loads(path.read_text(encoding="utf-8"))
|
|
schema = parser.extract_form_schema(doc)
|
|
assert "fallback" not in schema
|
|
props = schema["json_schema"]["properties"]
|
|
assert set(props) == {"rolle", "mitarbeiter_name", "geburtsdatum"}
|
|
assert schema["json_schema"]["required"] == ["rolle", "mitarbeiter_name"]
|
|
# Einstufung is branched (visible only for Pfarrer:in).
|
|
assert len(schema["json_schema"]["allOf"]) == 1
|
|
rule = schema["json_schema"]["allOf"][0]
|
|
assert rule["if"]["properties"] == {"rolle": {"const": "Pfarrer:in"}}
|
|
assert set(rule["then"]["properties"]) == {"einstufung"}
|
|
|
|
|
|
def test_schema_field_names_covers_branches():
|
|
doc = _doc(
|
|
_dropdown("a", ["x", "y"]),
|
|
_dropdown("b", ["p", "q"], info="[visible_if a=x]"),
|
|
_field("c", "IntInput", info="[visible_if b=p]"),
|
|
)
|
|
cached = parser.extract_form_schema(doc)
|
|
assert parser.schema_field_names(cached) == ["a", "b", "c"]
|
|
plain = parser.extract_form_schema(
|
|
{
|
|
"data": {
|
|
"nodes": [{"data": {"type": "ChatInput", "node": {"template": {}}}}],
|
|
"edges": [],
|
|
}
|
|
}
|
|
)
|
|
assert parser.schema_field_names(plain) == []
|
|
assert parser.schema_field_names({}) == []
|
|
assert parser.schema_field_names({"json_schema": None}) == []
|