574 lines
20 KiB
Python
574 lines
20 KiB
Python
"""Phase 2 tests: kirche CustomInput* library + intake fixture flow."""
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from lfx.custom import Component
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent.parent
|
|
sys.path.insert(0, str(REPO_ROOT / "backend" / "components"))
|
|
|
|
from kirche.form_inputs import CustomInputForm
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures" / "flows"
|
|
|
|
FORM_FIELDS = [
|
|
"full_name",
|
|
"notes",
|
|
"password",
|
|
"age",
|
|
"amount",
|
|
"newsletter",
|
|
"department",
|
|
"topics",
|
|
"satisfaction",
|
|
"attachment",
|
|
]
|
|
|
|
|
|
def test_component_contract():
|
|
assert issubclass(CustomInputForm, Component)
|
|
assert CustomInputForm.display_name.startswith("CustomInput")
|
|
inputs = {i.name: i for i in CustomInputForm.inputs}
|
|
# Ten form fields plus the hidden run-envelope plumbing input.
|
|
assert sorted(n for n in inputs if n != "form_values") == sorted(FORM_FIELDS)
|
|
assert inputs["form_values"].show is False
|
|
outputs = {o.name: o for o in CustomInputForm.outputs}
|
|
assert set(outputs) == {"form_data"}
|
|
|
|
|
|
def test_collect_merges_submitted_values_over_defaults():
|
|
from lfx.schema.data import Data
|
|
|
|
form = CustomInputForm()
|
|
default = form.collect()
|
|
assert isinstance(default, Data)
|
|
assert default.data["full_name"] == ""
|
|
assert default.data["department"] == "Music"
|
|
|
|
form.form_values = json.dumps({"full_name": "Maria", "age": 41})
|
|
merged = form.collect().data
|
|
assert merged["full_name"] == "Maria"
|
|
assert merged["age"] == 41
|
|
assert merged["department"] == "Music" # default kept
|
|
|
|
# Run envelope shape {"payload": {...}} unwraps like the other components.
|
|
form.form_values = json.dumps({"payload": {"notes": "hi"}})
|
|
assert form.collect().data["notes"] == "hi"
|
|
|
|
form.form_values = "not json"
|
|
assert form.collect().data["full_name"] == ""
|
|
|
|
|
|
def test_collect_picks_up_langflow_added_field():
|
|
"""A field added in the Langflow code editor needs no collect() change."""
|
|
from lfx.io import StrInput
|
|
|
|
form = CustomInputForm()
|
|
form._inputs["room"] = StrInput(name="room", display_name="Room", info="Where", value="Chapel")
|
|
collected = form.collect().data
|
|
assert collected["room"] == "Chapel"
|
|
assert collected["full_name"] == ""
|
|
|
|
|
|
def test_collect_survives_rename():
|
|
"""Renaming `full_name` to `name` drops the old key, keeps the new one."""
|
|
from lfx.io import StrInput
|
|
|
|
form = CustomInputForm()
|
|
del form._inputs["full_name"]
|
|
form._inputs["name"] = StrInput(name="name", display_name="Name", info="", value="Maria")
|
|
collected = form.collect().data
|
|
assert "full_name" not in collected
|
|
assert collected["name"] == "Maria"
|
|
|
|
|
|
def test_collect_keeps_unknown_envelope_keys():
|
|
"""Frontend payloads with renamed keys merge even if undeclared here."""
|
|
form = CustomInputForm()
|
|
form.form_values = json.dumps({"payload": {"nickname": "Mia"}})
|
|
collected = form.collect().data
|
|
assert collected["nickname"] == "Mia"
|
|
assert collected["department"] == "Music" # default kept
|
|
|
|
|
|
def test_collect_fields_list_restricts_defaults():
|
|
"""Asked-field list: unasked code defaults stay out of the output."""
|
|
form = CustomInputForm()
|
|
form.form_values = json.dumps(
|
|
{"payload": {"full_name": "Maria"}, "fields": ["full_name", "age"]}
|
|
)
|
|
collected = form.collect().data
|
|
assert collected == {"full_name": "Maria", "age": 18}
|
|
|
|
|
|
def test_collect_fields_list_fills_missing_from_defaults():
|
|
form = CustomInputForm()
|
|
form.form_values = json.dumps(
|
|
{"payload": {"full_name": "Maria"}, "fields": ["full_name", "department"]}
|
|
)
|
|
collected = form.collect().data
|
|
assert collected == {"full_name": "Maria", "department": "Music"}
|
|
|
|
|
|
def test_collect_fields_list_ignores_undeclared_names():
|
|
"""Asked-but-undeclared template fields contribute no Nones."""
|
|
form = CustomInputForm()
|
|
form.form_values = json.dumps({"payload": {"room": "Chapel"}, "fields": ["full_name", "room"]})
|
|
collected = form.collect().data
|
|
assert collected == {"full_name": "", "room": "Chapel"}
|
|
|
|
|
|
def test_collect_raw_dict_ignores_fields_key():
|
|
"""A raw dict is payload, not envelope: 'fields' is just a value."""
|
|
form = CustomInputForm()
|
|
form.form_values = json.dumps({"fields": ["x"], "full_name": "Maria"})
|
|
collected = form.collect().data
|
|
assert collected["fields"] == ["x"]
|
|
assert collected["full_name"] == "Maria"
|
|
assert collected["age"] == 18 # all defaults apply without envelope
|
|
|
|
|
|
def test_collect_hides_branch_default_until_trigger_matches():
|
|
"""Hidden branch fields contribute no defaults (frontend parity)."""
|
|
form = CustomInputForm()
|
|
form._inputs["notes"].info = "Anything [visible_if department=Youth]"
|
|
assert "notes" not in form.collect().data # department defaults to Music
|
|
form.form_values = json.dumps({"payload": {"department": "Youth"}})
|
|
assert form.collect().data["notes"] == ""
|
|
|
|
|
|
def test_collect_branch_visibility_uses_submitted_triggers():
|
|
form = CustomInputForm()
|
|
form._inputs["notes"].info = "[visible_if age=41]"
|
|
form.form_values = json.dumps({"payload": {"age": 41}, "fields": ["age", "notes"]})
|
|
assert form.collect().data["notes"] == ""
|
|
form.form_values = json.dumps({"payload": {"age": 18}, "fields": ["age", "notes"]})
|
|
assert "notes" not in form.collect().data
|
|
|
|
|
|
def test_collect_dangling_marker_stays_visible():
|
|
form = CustomInputForm()
|
|
form._inputs["notes"].info = "[visible_if nope=x]"
|
|
assert "notes" in form.collect().data
|
|
|
|
|
|
def test_bauvorhaben_component_contract():
|
|
from kirche.bauvorhaben_form import CustomInputBauvorhaben
|
|
|
|
assert CustomInputBauvorhaben.display_name == "Bauvorhaben Form"
|
|
inputs = {i.name: i for i in CustomInputBauvorhaben.inputs}
|
|
assert sorted(n for n in inputs if n != "form_values") == sorted(
|
|
[
|
|
"gebaeude",
|
|
"bestandteil",
|
|
"glockenmassnahme",
|
|
"gussjahr",
|
|
"gebaeudeart",
|
|
"denkmalschutz",
|
|
"denkmalnummer",
|
|
]
|
|
)
|
|
assert inputs["form_values"].show is False
|
|
outputs = {o.name: o for o in CustomInputBauvorhaben.outputs}
|
|
assert set(outputs) == {"form_data"}
|
|
|
|
|
|
def test_bauvorhaben_collect_only_own_fields():
|
|
"""Standalone component: no intake leftovers, hidden branches dropped."""
|
|
from kirche.bauvorhaben_form import CustomInputBauvorhaben
|
|
|
|
form = CustomInputBauvorhaben()
|
|
default = form.collect().data
|
|
assert "full_name" not in default and "age" not in default
|
|
assert default["gebaeude"] == "" # class default (template ships " " sentinel)
|
|
assert "bestandteil" not in default # hidden (gebaeude default "").
|
|
assert "gebaeudeart" not in default
|
|
|
|
form.form_values = json.dumps(
|
|
{
|
|
"payload": {"gebaeude": "Kirche", "bestandteil": ["Glocke"]},
|
|
"fields": [
|
|
"gebaeude",
|
|
"bestandteil",
|
|
"glockenmassnahme",
|
|
"gussjahr",
|
|
"gebaeudeart",
|
|
"denkmalschutz",
|
|
"denkmalnummer",
|
|
],
|
|
}
|
|
)
|
|
collected = form.collect().data
|
|
assert collected["bestandteil"] == ["Glocke"]
|
|
assert collected["glockenmassnahme"] == "Reinigung" # visible branch default
|
|
assert "gussjahr" not in collected # hidden branch
|
|
assert "gebaeudeart" not in collected
|
|
assert "full_name" not in collected
|
|
|
|
|
|
def _parse_chat_json(text: str) -> dict:
|
|
"""Parse terminal ChatOutput text: raw JSON or a ```json fenced block."""
|
|
text = text.strip()
|
|
if text.startswith("```"):
|
|
lines = text.splitlines()
|
|
assert lines[0].startswith("```"), text[:60]
|
|
assert lines[-1].startswith("```"), text[-60:]
|
|
text = "\n".join(lines[1:-1])
|
|
return json.loads(text)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
(
|
|
"fixture_file",
|
|
"doc_name",
|
|
"node_id",
|
|
"expected_type",
|
|
"expected_display",
|
|
"display_mode",
|
|
"expected_fields",
|
|
"expected_edges",
|
|
),
|
|
[
|
|
(
|
|
"intake_form.json",
|
|
None,
|
|
"CustomInputForm-intake",
|
|
"CustomInputForm",
|
|
"CustomInput",
|
|
"startswith",
|
|
{
|
|
"full_name",
|
|
"notes",
|
|
"password",
|
|
"age",
|
|
"amount",
|
|
"newsletter",
|
|
"department",
|
|
"topics",
|
|
"satisfaction",
|
|
"attachment",
|
|
},
|
|
[
|
|
("ChatInput-entry", "CustomInputForm-intake"),
|
|
("CustomInputForm-intake", "ChatOutput-collected"),
|
|
],
|
|
),
|
|
(
|
|
"bauvorhaben.json",
|
|
"bauvorhaben",
|
|
"CustomInputBauvorhaben-bauvorhaben",
|
|
"CustomInputBauvorhaben",
|
|
"Bauvorhaben Form",
|
|
"exact",
|
|
{
|
|
"form_values",
|
|
"gebaeude",
|
|
"bestandteil",
|
|
"glockenmassnahme",
|
|
"gussjahr",
|
|
"gebaeudeart",
|
|
"denkmalschutz",
|
|
"denkmalnummer",
|
|
"code",
|
|
},
|
|
[
|
|
("ChatInput-entry", "CustomInputBauvorhaben-bauvorhaben"),
|
|
("CustomInputBauvorhaben-bauvorhaben", "ChatOutput-collected"),
|
|
],
|
|
),
|
|
(
|
|
"mitarbeiter_einstellen.json",
|
|
"mitarbeiter_einstellen",
|
|
"CustomInputMitarbeiter-mitarbeiter",
|
|
"CustomInputMitarbeiter",
|
|
"Mitarbeiter Einstellen Form",
|
|
"exact",
|
|
{
|
|
"form_values",
|
|
"rolle",
|
|
"einstufung",
|
|
"mitarbeiter_name",
|
|
"geburtsdatum",
|
|
"code",
|
|
},
|
|
[
|
|
("ChatInput-entry", "CustomInputMitarbeiter-mitarbeiter"),
|
|
("CustomInputMitarbeiter-mitarbeiter", "Gehaltsrechner-gehalt"),
|
|
("Gehaltsrechner-gehalt", "ChatOutput-collected"),
|
|
],
|
|
),
|
|
],
|
|
ids=["intake", "bauvorhaben", "mitarbeiter"],
|
|
)
|
|
def test_fixture_shape(
|
|
fixture_file,
|
|
doc_name,
|
|
node_id,
|
|
expected_type,
|
|
expected_display,
|
|
display_mode,
|
|
expected_fields,
|
|
expected_edges,
|
|
):
|
|
doc = json.loads((FIXTURES / fixture_file).read_text(encoding="utf-8"))
|
|
if doc_name is not None:
|
|
assert doc["name"] == doc_name
|
|
nodes = {n["id"]: n for n in doc["data"]["nodes"]}
|
|
form = nodes[node_id]
|
|
assert form["data"]["type"] == expected_type
|
|
display = form["data"]["node"]["display_name"]
|
|
if display_mode == "startswith":
|
|
assert display.startswith(expected_display)
|
|
else:
|
|
assert display == expected_display
|
|
template = form["data"]["node"]["template"]
|
|
assert set(template) >= expected_fields
|
|
assert template["form_values"]["show"] is False
|
|
assert len(template["code"]["value"]) > 200
|
|
outputs = {o["name"]: o for o in form["data"]["node"]["outputs"]}
|
|
assert outputs["form_data"]["types"] == ["JSON"] # JSON object, not Message
|
|
edges = {(e["source"], e["target"]) for e in doc["data"]["edges"]}
|
|
for edge in expected_edges:
|
|
assert edge in edges
|
|
|
|
|
|
async def _check_intake_result(client, flow_id, version, run):
|
|
# collect() returns Data: the terminal ChatOutput renders it as a
|
|
# fenced ```json block, so unwrap before parsing.
|
|
result = _parse_chat_json(run["output"]["result"])
|
|
assert result["full_name"] == "Maria Muster"
|
|
assert result["age"] == 41
|
|
assert result["department"] == "Music" # default preserved
|
|
|
|
|
|
async def _check_bauvorhaben_result(client, flow_id, version, run):
|
|
# Branch rules are cached per version and served via input-schema.
|
|
cached = version["input_schema"]
|
|
assert len(cached["json_schema"]["allOf"]) == 5
|
|
schema = await client.get(f"/api/flows/{flow_id}/input-schema")
|
|
assert schema.status_code == 200, schema.text
|
|
assert set(schema.json()["json_schema"]["properties"]) == {"gebaeude", "denkmalschutz"}
|
|
result = _parse_chat_json(run["output"]["result"])
|
|
assert result["gebaeude"] == "Kirche"
|
|
assert result["bestandteil"] == ["Glocke", "Orgel"]
|
|
assert result["gussjahr"] == 1890
|
|
assert result["denkmalnummer"] == "D-123"
|
|
# Only asked fields: unasked code defaults (age, amount, ...) stay out.
|
|
assert set(result) == {
|
|
"gebaeude",
|
|
"bestandteil",
|
|
"glockenmassnahme",
|
|
"gussjahr",
|
|
"denkmalschutz",
|
|
"denkmalnummer",
|
|
}
|
|
|
|
|
|
async def _check_mitarbeiter_result(client, flow_id, version, run):
|
|
result = run["output"]["result"]
|
|
assert "65000" in result
|
|
assert "Maria Muster" in result
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("fixture_file", "flow_name", "run_input", "check_result"),
|
|
[
|
|
(
|
|
"intake_form.json",
|
|
"intake-e2e",
|
|
{"full_name": "Maria Muster", "age": 41},
|
|
_check_intake_result,
|
|
),
|
|
(
|
|
"bauvorhaben.json",
|
|
"bauvorhaben-e2e",
|
|
{
|
|
"gebaeude": "Kirche",
|
|
"bestandteil": ["Glocke", "Orgel"],
|
|
"glockenmassnahme": "Neuguss",
|
|
"gussjahr": 1890,
|
|
"denkmalschutz": True,
|
|
"denkmalnummer": "D-123",
|
|
},
|
|
_check_bauvorhaben_result,
|
|
),
|
|
(
|
|
"mitarbeiter_einstellen.json",
|
|
"mitarbeiter-e2e",
|
|
{
|
|
"rolle": "Pfarrer:in",
|
|
"einstufung": "Stufe 2",
|
|
"mitarbeiter_name": "Maria Muster",
|
|
"geburtsdatum": "1980-01-01",
|
|
},
|
|
_check_mitarbeiter_result,
|
|
),
|
|
],
|
|
ids=["intake", "bauvorhaben", "mitarbeiter"],
|
|
)
|
|
async def test_deploy_and_run(
|
|
client, lfx_server, login, fixture_file, flow_name, run_input, check_result
|
|
):
|
|
await login()
|
|
doc = json.loads((FIXTURES / fixture_file).read_text(encoding="utf-8"))
|
|
created = await client.post("/api/flows", json={"name": flow_name})
|
|
assert created.status_code == 201, created.text
|
|
flow_id = created.json()["id"]
|
|
|
|
version = await client.post(
|
|
f"/api/flows/{flow_id}/versions",
|
|
json={"langflow_json": doc, "author": "test"},
|
|
)
|
|
assert version.status_code == 201, version.text
|
|
|
|
deployed = await client.post(f"/api/flows/{flow_id}/versions/1/deploy")
|
|
assert deployed.status_code == 200, deployed.text
|
|
|
|
run = await client.post(f"/api/flows/{flow_id}/run", json={"input": run_input})
|
|
assert run.status_code == 200, run.text
|
|
assert run.json()["status"] == "completed"
|
|
await check_result(client, flow_id, version.json(), run.json())
|
|
|
|
|
|
def test_embedded_code_matches_template_fields():
|
|
"""Node template fields == code-declared inputs (no unasked defaults).
|
|
|
|
The envelope tolerates drift at runtime, but authoring must stay
|
|
honest: what Langflow shows in the code editor must be what the form
|
|
asks. Building blocks: lfx's own create_class on the embedded code.
|
|
"""
|
|
from lfx.custom import validate as validate_mod
|
|
|
|
cases = {
|
|
"intake_form.json": ("CustomInputForm-intake", "CustomInputForm"),
|
|
"bauvorhaben.json": ("CustomInputBauvorhaben-bauvorhaben", "CustomInputBauvorhaben"),
|
|
"mitarbeiter_einstellen.json": (
|
|
"CustomInputMitarbeiter-mitarbeiter",
|
|
"CustomInputMitarbeiter",
|
|
),
|
|
}
|
|
for filename, (node_id, class_name) in cases.items():
|
|
doc = json.loads((FIXTURES / filename).read_text(encoding="utf-8"))
|
|
node = next(n for n in doc["data"]["nodes"] if n["id"] == node_id)
|
|
assert node["data"]["type"] == class_name, filename
|
|
template = node["data"]["node"]["template"]
|
|
template_names = {
|
|
field["name"] for field in template.values() if isinstance(field, dict)
|
|
} - {"code"}
|
|
code = template["code"]["value"]
|
|
# Codegen-inlined core: self-contained for LFX, single truth in repo.
|
|
assert "from kirche" not in code, filename
|
|
assert "from ._" not in code, filename
|
|
assert "collect_form_data" in code, filename
|
|
assert "parse_envelope" in code, filename
|
|
cls = validate_mod.create_class(code, class_name)
|
|
code_names = {inp.name for inp in cls.inputs}
|
|
assert code_names == template_names, filename
|
|
|
|
|
|
def test_form_core_envelope_and_conditions():
|
|
"""Unit tests for the shared core (single truth for all 3 forms)."""
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "components"))
|
|
from kirche._form_core import (
|
|
_field_conditions,
|
|
collect_form_data,
|
|
parse_envelope,
|
|
)
|
|
|
|
payload, asked = parse_envelope('{"payload": {"a": 1}, "fields": ["a", ""]}')
|
|
assert payload == {"a": 1} and asked == ["a"]
|
|
payload, asked = parse_envelope('{"a": 2}')
|
|
assert payload == {"a": 2} and asked is None
|
|
payload, asked = parse_envelope("not json")
|
|
assert payload == {} and asked is None
|
|
assert _field_conditions("x [visible_if a=B]", None) == [("a", "B")]
|
|
assert _field_conditions("x", [{"field": "a", "value": 1}]) == [("a", 1)]
|
|
assert _field_conditions("x [visible_if a=B]", {"a": 2}) == [("a", 2)]
|
|
effective, scope = collect_form_data(["a", "b"], {"a": 1, "b": 2}, {"b": 3}, ["a"])
|
|
assert scope == ["a"] and effective["b"] == 3
|
|
|
|
|
|
def test_kirche_node_inlines_core():
|
|
"""Generator composes thin source + core into self-contained code."""
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
|
|
from make_flow_fixtures import _compose_component_code, _inline_core
|
|
|
|
core = _inline_core()
|
|
assert "def parse_envelope" in core and "def collect_form_data" in core
|
|
thin = "from kirche._form_core import parse_envelope\n# <-- _form_core inlined -->\nX = 1\n"
|
|
composed = _compose_component_code(thin)
|
|
assert "from kirche" not in composed and "def parse_envelope" in composed
|
|
|
|
|
|
def test_bauvorhaben_schema_branches():
|
|
from app.flows import input_schema as parser
|
|
|
|
doc = json.loads((FIXTURES / "bauvorhaben.json").read_text(encoding="utf-8"))
|
|
out = parser.extract_form_schema(doc)
|
|
schema = out["json_schema"]
|
|
assert set(schema["properties"]) == {"gebaeude", "denkmalschutz"}
|
|
assert schema["required"] == ["gebaeude"]
|
|
by_field = {next(iter(rule["then"]["properties"])): rule for rule in schema["allOf"]}
|
|
assert len(schema["allOf"]) == len(by_field) == 5
|
|
assert set(by_field) == {
|
|
"bestandteil",
|
|
"glockenmassnahme",
|
|
"gussjahr",
|
|
"gebaeudeart",
|
|
"denkmalnummer",
|
|
}
|
|
assert by_field["bestandteil"]["if"]["properties"] == {"gebaeude": {"const": "Kirche"}}
|
|
assert by_field["glockenmassnahme"]["if"]["properties"] == {
|
|
"bestandteil": {"contains": {"const": "Glocke"}}
|
|
}
|
|
assert by_field["gussjahr"]["if"]["properties"] == {"glockenmassnahme": {"const": "Neuguss"}}
|
|
assert by_field["gebaeudeart"]["if"]["properties"] == {"gebaeude": {"const": "Anderes Gebäude"}}
|
|
assert by_field["denkmalnummer"]["if"]["properties"] == {"denkmalschutz": {"const": True}}
|
|
assert by_field["bestandteil"]["then"]["required"] == ["bestandteil"]
|
|
assert by_field["denkmalnummer"]["then"]["required"] == ["denkmalnummer"]
|
|
assert "required" not in by_field["gussjahr"]["then"]
|
|
# Markers stripped from descriptions.
|
|
assert "visible_if" not in out["json_schema"]["allOf"][0]["then"]["properties"][
|
|
"bestandteil"
|
|
].get("description", "")
|
|
order = out["ui_schema"]["ui:order"]
|
|
assert order[-1] == "*"
|
|
assert set(order[:-1]) == {
|
|
"gebaeude",
|
|
"bestandteil",
|
|
"glockenmassnahme",
|
|
"gussjahr",
|
|
"gebaeudeart",
|
|
"denkmalschutz",
|
|
"denkmalnummer",
|
|
}
|
|
|
|
|
|
def test_mitarbeiter_component_contract():
|
|
from kirche.mitarbeiter_form import CustomInputMitarbeiter
|
|
|
|
assert CustomInputMitarbeiter.display_name == "Mitarbeiter Einstellen Form"
|
|
inputs = {i.name: i for i in CustomInputMitarbeiter.inputs}
|
|
assert sorted(n for n in inputs if n != "form_values") == sorted(
|
|
["rolle", "einstufung", "mitarbeiter_name", "geburtsdatum"]
|
|
)
|
|
assert inputs["form_values"].show is False
|
|
outputs = {o.name: o for o in CustomInputMitarbeiter.outputs}
|
|
assert set(outputs) == {"form_data"}
|
|
|
|
|
|
def test_mitarbeiter_collect_branch_default():
|
|
"""Einstufung default applies only when Pfarrer:in is selected."""
|
|
from kirche.mitarbeiter_form import CustomInputMitarbeiter
|
|
|
|
form = CustomInputMitarbeiter()
|
|
assert "einstufung" not in form.collect().data # rolle defaults to Assistenz
|
|
form.form_values = json.dumps({"payload": {"rolle": "Pfarrer:in"}})
|
|
assert form.collect().data["einstufung"] == "Stufe 1"
|