diff --git a/koolook_setups.py b/koolook_setups.py index 36dbd2f..68bcbb2 100644 --- a/koolook_setups.py +++ b/koolook_setups.py @@ -548,6 +548,8 @@ def _validate_persisted_setup_surface( items = surface.get(key) if isinstance(items, list): _validate_surface_entries(items, f"setupSurface.{key}", diagnostics) + if "app" in surface: + _validate_surface_app(surface["app"], "setupSurface.app", diagnostics) def _validate_surface_entries( @@ -576,6 +578,77 @@ def _validate_surface_entries( diagnostics.append(f"{node_path}.{field} must be non-empty text") +def _validate_surface_app(app: Any, path: str, diagnostics: list[str]) -> None: + if not isinstance(app, dict): + diagnostics.append(f"{path} must be a JSON object") + return + for key in ("inputs", "outputs", "results"): + value = app.get(key) + if value is None: + continue + if not isinstance(value, list): + diagnostics.append(f"{path}.{key} must be a list") + continue + _validate_surface_app_fields(value, f"{path}.{key}", diagnostics) + if "switch" in app: + _validate_surface_app_switch(app["switch"], f"{path}.switch", diagnostics) + + +def _validate_surface_app_fields(fields: list[Any], path: str, diagnostics: list[str]) -> None: + for field_index, field in enumerate(fields): + field_path = f"{path}[{field_index}]" + if not isinstance(field, dict): + diagnostics.append(f"{field_path} must be a JSON object") + continue + _validate_non_empty_text(field.get("key"), f"{field_path}.key", diagnostics) + _validate_non_empty_text(field.get("label"), f"{field_path}.label", diagnostics) + if "visible" in field and not isinstance(field["visible"], bool): + diagnostics.append(f"{field_path}.visible must be a boolean") + _validate_surface_app_target(field.get("target"), f"{field_path}.target", diagnostics) + + +def _validate_surface_app_switch(switch: Any, path: str, diagnostics: list[str]) -> None: + if not isinstance(switch, dict): + diagnostics.append(f"{path} must be a JSON object") + return + _validate_non_empty_text(switch.get("key"), f"{path}.key", diagnostics) + _validate_non_empty_text(switch.get("label"), f"{path}.label", diagnostics) + if "visible" in switch and not isinstance(switch["visible"], bool): + diagnostics.append(f"{path}.visible must be a boolean") + _validate_surface_app_target(switch.get("target"), f"{path}.target", diagnostics) + default = switch.get("default") + if default is not None and (isinstance(default, bool) or not isinstance(default, int)): + diagnostics.append(f"{path}.default must be an integer") + options = switch.get("options") + if not isinstance(options, list): + diagnostics.append(f"{path}.options must be a list") + return + for option_index, option in enumerate(options): + option_path = f"{path}.options[{option_index}]" + if not isinstance(option, dict): + diagnostics.append(f"{option_path} must be a JSON object") + continue + if isinstance(option.get("value"), bool) or not isinstance(option.get("value"), int): + diagnostics.append(f"{option_path}.value must be an integer") + _validate_non_empty_text(option.get("label"), f"{option_path}.label", diagnostics) + _validate_non_empty_text(option.get("input"), f"{option_path}.input", diagnostics) + if "visible" in option and not isinstance(option["visible"], bool): + diagnostics.append(f"{option_path}.visible must be a boolean") + + +def _validate_surface_app_target(target: Any, path: str, diagnostics: list[str]) -> None: + if not isinstance(target, dict): + diagnostics.append(f"{path} must be a JSON object") + return + _validate_non_empty_text(target.get("node"), f"{path}.node", diagnostics) + _validate_non_empty_text(target.get("input"), f"{path}.input", diagnostics) + + +def _validate_non_empty_text(value: Any, path: str, diagnostics: list[str]) -> None: + if not isinstance(value, str) or not value.strip(): + diagnostics.append(f"{path} must be non-empty text") + + def _uses_group_authored_surface( input_contract: dict[str, Any], output_contract: dict[str, Any], diff --git a/tests/server/test_published_setup_registry.py b/tests/server/test_published_setup_registry.py index d363854..e038d3e 100644 --- a/tests/server/test_published_setup_registry.py +++ b/tests/server/test_published_setup_registry.py @@ -196,6 +196,86 @@ def test_registry_omits_malformed_persisted_setup_surface() -> None: assert "ltx-director-demo: setupSurface.outputs must be a list" in registry.diagnostics +def test_registry_omits_malformed_persisted_setup_surface_app() -> None: + setup = deepcopy(_valid_setup()) + setup["inputContract"] = {"inputs": []} + setup["outputContract"] = {"outputs": []} + setup["setupSurface"] = { + "sourceInputs": [ + { + "group": "Koolook Input", + "nodes": [{"id": "12", "type": "Koolook_PublishInput", "title": "Input"}], + } + ], + "outputs": [ + { + "group": "Koolook Output", + "nodes": [{"id": "20", "type": "Koolook_PublishOutput", "title": "Output"}], + } + ], + "controls": [], + "app": "not-object", + } + registry = PublishedSetupRegistry(StaticSetupStorage([setup])) + + assert registry.getSetup("ltx-director-demo") is None + assert "ltx-director-demo: setupSurface.app must be a JSON object" in registry.diagnostics + + +def test_registry_omits_malformed_persisted_setup_surface_app_fields() -> None: + setup = deepcopy(_valid_setup()) + setup["inputContract"] = {"inputs": []} + setup["outputContract"] = {"outputs": []} + setup["setupSurface"] = { + "sourceInputs": [ + { + "group": "Koolook Input", + "nodes": [{"id": "12", "type": "Koolook_PublishInput", "title": "Input"}], + } + ], + "outputs": [ + { + "group": "Koolook Output", + "nodes": [{"id": "20", "type": "Koolook_PublishOutput", "title": "Output"}], + } + ], + "controls": [], + "app": { + "inputs": [ + { + "key": "single_file", + "label": "Single file", + "visible": True, + "target": {"node": "12"}, + "default": "/image.png", + } + ], + "outputs": "not-list", + "results": [], + "switch": { + "key": "switch", + "label": "Input type", + "visible": True, + "target": {"node": "12", "input": "mode"}, + "default": 2, + "options": [{"value": True, "label": "Img", "visible": True, "input": "single_file"}], + }, + }, + } + registry = PublishedSetupRegistry(StaticSetupStorage([setup])) + + assert registry.getSetup("ltx-director-demo") is None + assert "ltx-director-demo: setupSurface.app.outputs must be a list" in registry.diagnostics + assert ( + "ltx-director-demo: setupSurface.app.inputs[0].target.input must be non-empty text" + in registry.diagnostics + ) + assert ( + "ltx-director-demo: setupSurface.app.switch.options[0].value must be an integer" + in registry.diagnostics + ) + + def test_file_storage_loads_setups_object_and_uses_sample_fallback(tmp_path) -> None: primary = tmp_path / "setups.json" fallback = tmp_path / "sample.json"