diff --git a/comfy_cli/cmdline.py b/comfy_cli/cmdline.py index 9a3594f9..72a447b2 100644 --- a/comfy_cli/cmdline.py +++ b/comfy_cli/cmdline.py @@ -1052,11 +1052,12 @@ def run( @app.command( + hidden=True, help=( - "Validate a workflow without submitting (UI exports are converted to API format first). " - "Checks class_types, input shapes, enum values, edge wiring, and the dotted sub-inputs a " - "dynamic combo's selected option requires." - ) + "[DEPRECATED — use 'comfy workflow validate'] Validate a workflow without submitting " + "(UI exports are converted to API format first). Checks class_types, input shapes, enum " + "values, edge wiring, and the dotted sub-inputs a dynamic combo's selected option requires." + ), ) @tracking.track_command() def validate( @@ -1081,145 +1082,14 @@ def validate( typer.Option("--input", show_default=False, help="Path to a saved object_info JSON (offline mode)."), ] = None, ): - from pathlib import Path - - from comfy_cli.command.run import is_ui_workflow - from comfy_cli.command.run.preflight import _detect_partner_nodes - from comfy_cli.cql.engine import Graph, LoadError - from comfy_cli.workflow_to_api import WorkflowConversionError, convert_ui_to_api - - renderer = get_renderer() - - # Load workflow - wf_path = Path(workflow).expanduser() - if not wf_path.is_file(): - renderer.error(code="workflow_not_found", message=f"Workflow file not found: {workflow}", hint="check the path") - raise typer.Exit(code=1) - try: - wf_data = json.loads(wf_path.read_text(encoding="utf-8")) - except json.JSONDecodeError as e: - renderer.error(code="workflow_invalid_json", message=f"Invalid JSON: {e}", hint="re-export from ComfyUI") - raise typer.Exit(code=1) from e - if not isinstance(wf_data, dict): - renderer.error( - code="workflow_not_api_format", message="Workflow must be a JSON object", hint="use File > Export (API)" - ) - raise typer.Exit(code=1) - - # Load graph - mode = "local" - if where: - mode = where - else: - config = ConfigManager() - try: - decision = where_module.resolve(flag=None, config_value=config.get(where_module.CONFIG_KEY_WHERE_DEFAULT)) - mode = decision.target.value - except Exception: - pass - - try: - graph = Graph.load(mode=mode, input_path=input_path, host=host, port=port) - except LoadError as e: - renderer.error( - code="cql_no_graph", - message=str(e), - hint=e.details.get("hint", "pass --input , or start the server"), - details=e.details, - ) - raise typer.Exit(code=1) from e - - # Detect a UI-export (frontend/canvas) workflow and lower it to API format - # before validating — exactly as `comfy run` does. Without this the wrapper - # keys (`nodes`, `links`, `groups`, `config`, …) each emit a `non_node_key` - # warning, zero nodes are checked, and the result is a vacuous `valid:true`. - # The converter reuses the object_info the graph was already built from - # (`graph.object_info`), so offline `--input` works and no second fetch happens. - converted_from_ui = False - if is_ui_workflow(wf_data): - if renderer.is_pretty(): - rprint("[yellow]Detected UI-format workflow, converting to API format...[/yellow]") - try: - converted = convert_ui_to_api(wf_data, graph.object_info) - except WorkflowConversionError as e: - renderer.error( - code="workflow_not_api_format", - message=f"Workflow is a UI export that could not be converted to API format: {e}", - hint="use ComfyUI's 'File > Export (API)' to save as API format", - ) - raise typer.Exit(code=1) from e - except Exception as e: # noqa: BLE001 — never leak a raw traceback to the agent flow - renderer.error( - code="conversion_crash", - message=f"Workflow conversion crashed unexpectedly: {type(e).__name__}: {e}", - hint="report this at https://github.com/Comfy-Org/comfy-cli/issues", - details={"exception_type": type(e).__name__}, - ) - raise typer.Exit(code=1) from e - if not converted: - renderer.error( - code="workflow_not_api_format", - message="Workflow is a UI export that converted to no executable nodes", - hint="use ComfyUI's 'File > Export (API)' to save as API format", - ) - raise typer.Exit(code=1) - wf_data = converted - converted_from_ui = True - - result = graph.validate_workflow(wf_data) - - # Preview credit spend: partner-API (paid) nodes spend Comfy credits when the - # workflow is run. This is the same detection `comfy run` uses (authoritative - # `api_node: true`, `partner/...` category fallback), surfaced here read-only - # so agents can answer "will this spend credits?" without running. `wf_data` - # is API format at this point (any UI export already converted above), which - # is the format the detector reads. Purely informational — no exit-code gate. - partner_nodes = _detect_partner_nodes(wf_data, graph.object_info) - - payload = { - "workflow": str(wf_path), - "valid": result["valid"], - "error_count": len(result["errors"]), - "warning_count": len(result["warnings"]), - "errors": result["errors"], - "warnings": result["warnings"], - "partner_nodes": partner_nodes, - "spends_credits": bool(partner_nodes), - } - if converted_from_ui: - # Signal that validation ran against the converted graph, not the file's - # literal bytes, and report how many nodes the conversion produced. - payload["converted_from_ui"] = True - payload["converted_node_count"] = len(wf_data) - - if renderer.is_pretty(): - if result["valid"]: - rprint(f"[bold green]✓[/bold green] workflow is valid ({len(wf_data)} nodes)") - for w in result["warnings"]: - rprint(f" [yellow]⚠[/yellow] {w.get('message', '')}") - else: - rprint(f"[bold red]✗[/bold red] {len(result['errors'])} error(s)") - for e in result["errors"]: - msg = e.get("message", "") - suggestions = e.get("suggestions", []) - if suggestions: - msg += f" (did you mean: {', '.join(suggestions[:3])}?)" - rprint(f" [red]•[/red] node {e.get('node_id') or '?'}: {msg}") - for w in result["warnings"]: - rprint(f" [yellow]⚠[/yellow] {w.get('message', '')}") - if partner_nodes: - # class_type strings come from the workflow / object_info and can - # contain Rich-markup metacharacters (e.g. `[/yellow]`); escape each - # so an odd node name can't raise MarkupError and crash the command. - from rich.markup import escape as _escape - - rprint( - f"[yellow]⚠ uses partner-API (paid) nodes that spend Comfy credits: {', '.join(_escape(n) for n in partner_nodes)}[/yellow]" - ) - renderer.emit(payload, command="validate", ok=result["valid"]) - - if not result["valid"]: - raise typer.Exit(code=1) + # Deprecated alias for `comfy workflow validate` (its canonical home). Kept + # functional for backward compatibility; warns and delegates to the shared + # implementation. The warning routes to stderr in JSON modes, so structured + # output stays clean. + from comfy_cli.command.workflow import validate_api_workflow + + get_renderer().warn("`comfy validate` is deprecated — use `comfy workflow validate` instead.") + validate_api_workflow(workflow, where=where, host=host, port=port, input_path=input_path, command="validate") @app.command(help="Upload files to the ComfyUI server's input directory.") diff --git a/comfy_cli/command/custom_nodes/command.py b/comfy_cli/command/custom_nodes/command.py index aa5319f5..155d0930 100644 --- a/comfy_cli/command/custom_nodes/command.py +++ b/comfy_cli/command/custom_nodes/command.py @@ -1079,7 +1079,7 @@ def validate_node_for_publishing(): return config -@app.command("validate", help="Run validation checks for publishing") +@app.command("validate", help="Validate this custom node for registry publishing") @tracking.track_command("publish") def validate(): """ diff --git a/comfy_cli/command/workflow.py b/comfy_cli/command/workflow.py index e437ab93..185f7de0 100644 --- a/comfy_cli/command/workflow.py +++ b/comfy_cli/command/workflow.py @@ -1439,6 +1439,222 @@ def delete_cmd( renderer.emit(payload, command="workflow delete", where="cloud", changed=True) +# --------------------------------------------------------------------------- +# validate — API-format workflow validation +# --------------------------------------------------------------------------- +# The canonical home for API-format workflow validation. The top-level +# `comfy validate` is kept as a hidden deprecated alias that delegates to the +# shared implementation below (see cmdline.py). + + +def validate_api_workflow( + workflow: str, + *, + where: str | None = None, + host: str | None = None, + port: int | None = None, + input_path: str | None = None, + command: str = "workflow validate", +) -> None: + """Validate an API-format workflow without submitting it. + + Shared implementation behind ``comfy workflow validate`` (canonical) and the + deprecated top-level ``comfy validate`` alias. Checks class_types, input + shapes, enum values, and edge wiring against object_info loaded from the run + target. ``command`` labels the emitted envelope so each entry point reports + its own path. + """ + from comfy_cli import where as where_module + from comfy_cli.command.run import is_ui_workflow + from comfy_cli.command.run.preflight import _detect_partner_nodes + from comfy_cli.cql.engine import Graph, LoadError + from comfy_cli.workflow_to_api import WorkflowConversionError, convert_ui_to_api + + renderer = get_renderer() + + # Load workflow + wf_path = Path(workflow).expanduser() + if not wf_path.is_file(): + renderer.error(code="workflow_not_found", message=f"Workflow file not found: {workflow}", hint="check the path") + raise typer.Exit(code=1) + try: + wf_data = json.loads(wf_path.read_text(encoding="utf-8")) + except json.JSONDecodeError as e: + renderer.error(code="workflow_invalid_json", message=f"Invalid JSON: {e}", hint="re-export from ComfyUI") + raise typer.Exit(code=1) from e + except (OSError, UnicodeDecodeError) as e: + # e.g. a non-UTF-8 file, permission denied, or a TOCTOU race if the file + # vanished after the is_file() check above. Report structurally instead + # of crashing with a raw traceback. + renderer.error( + code="workflow_read_error", + message=f"Unable to read workflow file: {e}", + hint="check file permissions and encoding", + ) + raise typer.Exit(code=1) from e + if not isinstance(wf_data, dict): + renderer.error( + code="workflow_not_api_format", message="Workflow must be a JSON object", hint="use File > Export (API)" + ) + raise typer.Exit(code=1) + + # Load graph. Resolve routing up front so a typo (e.g. `clod`) fails here + # with an envelope: forwarded verbatim, it would surface as a bare + # ValueError from Graph.load's target resolution (online path only — + # Graph.load short-circuits to --input when supplied). + mode = "local" + try: + mode = where_module.resolve_default(flag=where).target.value + except ValueError as e: + if where: + renderer.error( + code="where_invalid", + message=str(e), + hint="use `--where local` or `--where cloud`", + ) + raise typer.Exit(code=1) from e + # A bad env/project/config value with no explicit flag never breaks the + # command — drop to the local default, as before. + except Exception: + pass + + try: + graph = Graph.load(mode=mode, input_path=input_path, host=host or "127.0.0.1", port=port or 8188) + except LoadError as e: + renderer.error( + code="cql_no_graph", + message=str(e), + hint=e.details.get("hint", "pass --input , or start the server"), + details=e.details, + ) + raise typer.Exit(code=1) from e + + # Detect a UI-export (frontend/canvas) workflow and lower it to API format + # before validating — exactly as `comfy run` does. Without this the wrapper + # keys (`nodes`, `links`, `groups`, `config`, …) each emit a `non_node_key` + # warning, zero nodes are checked, and the result is a vacuous `valid:true`. + # The converter reuses the object_info the graph was already built from + # (`graph.object_info`), so offline `--input` works and no second fetch happens. + converted_from_ui = False + if is_ui_workflow(wf_data): + if renderer.is_pretty(): + rprint("[yellow]Detected UI-format workflow, converting to API format...[/yellow]") + try: + converted = convert_ui_to_api(wf_data, graph.object_info) + except WorkflowConversionError as e: + renderer.error( + code="workflow_not_api_format", + message=f"Workflow is a UI export that could not be converted to API format: {e}", + hint="use ComfyUI's 'File > Export (API)' to save as API format", + ) + raise typer.Exit(code=1) from e + except Exception as e: # noqa: BLE001 — never leak a raw traceback to the agent flow + renderer.error( + code="conversion_crash", + message=f"Workflow conversion crashed unexpectedly: {type(e).__name__}: {e}", + hint="report this at https://github.com/Comfy-Org/comfy-cli/issues", + details={"exception_type": type(e).__name__}, + ) + raise typer.Exit(code=1) from e + if not converted: + renderer.error( + code="workflow_not_api_format", + message="Workflow is a UI export that converted to no executable nodes", + hint="use ComfyUI's 'File > Export (API)' to save as API format", + ) + raise typer.Exit(code=1) + wf_data = converted + converted_from_ui = True + + result = graph.validate_workflow(wf_data) + + # Preview credit spend: partner-API (paid) nodes spend Comfy credits when the + # workflow is run. This is the same detection `comfy run` uses (authoritative + # `api_node: true`, `partner/...` category fallback), surfaced here read-only + # so agents can answer "will this spend credits?" without running. `wf_data` + # is API format at this point (any UI export already converted above), which + # is the format the detector reads. Purely informational — no exit-code gate. + partner_nodes = _detect_partner_nodes(wf_data, graph.object_info) + + payload = { + "workflow": str(wf_path), + "valid": result["valid"], + "error_count": len(result["errors"]), + "warning_count": len(result["warnings"]), + "errors": result["errors"], + "warnings": result["warnings"], + "partner_nodes": partner_nodes, + "spends_credits": bool(partner_nodes), + } + if converted_from_ui: + # Signal that validation ran against the converted graph, not the file's + # literal bytes, and report how many nodes the conversion produced. + payload["converted_from_ui"] = True + payload["converted_node_count"] = len(wf_data) + + if renderer.is_pretty(): + # Workflow-supplied strings (node ids, messages, enum/input values echoed + # in messages) flow into rprint, which parses Rich markup. Escape them so a + # crafted file can't inject markup to spoof/hide output (e.g. fake a green ✓). + from rich.markup import escape + + if result["valid"]: + rprint(f"[bold green]✓[/bold green] workflow is valid ({len(wf_data)} nodes)") + for w in result["warnings"]: + rprint(f" [yellow]⚠[/yellow] {escape(str(w.get('message', '')))}") + else: + rprint(f"[bold red]✗[/bold red] {len(result['errors'])} error(s)") + for e in result["errors"]: + msg = str(e.get("message", "")) + suggestions = e.get("suggestions", []) + if suggestions: + msg += f" (did you mean: {', '.join(str(s) for s in suggestions[:3])}?)" + rprint(f" [red]•[/red] node {escape(str(e.get('node_id', '?')))}: {escape(msg)}") + for w in result["warnings"]: + rprint(f" [yellow]⚠[/yellow] {escape(str(w.get('message', '')))}") + if partner_nodes: + rprint( + f"[yellow]⚠ uses partner-API (paid) nodes that spend Comfy credits: " + f"{', '.join(escape(n) for n in partner_nodes)}[/yellow]" + ) + renderer.emit(payload, command=command, ok=result["valid"]) + + if not result["valid"]: + raise typer.Exit(code=1) + + +@app.command( + "validate", + help="Validate an API-format workflow without submitting. Checks class_types, input shapes, enum values, and edge wiring.", +) +@tracking.track_command("workflow") +def validate_cmd( + workflow: Annotated[ + str, + typer.Option(help="Path to the API-format workflow JSON file."), + ], + where: Annotated[ + str | None, + typer.Option("--where", show_default=False, help="Routing target for object_info: 'local' or 'cloud'."), + ] = None, + host: Annotated[ + str | None, + typer.Option(show_default=False, help="ComfyUI host (default 127.0.0.1)."), + ] = None, + port: Annotated[ + int | None, + typer.Option(show_default=False, help="ComfyUI port (default 8188)."), + ] = None, + input_path: Annotated[ + str | None, + typer.Option("--input", show_default=False, help="Path to a saved object_info JSON (offline mode)."), + ] = None, +): + validate_api_workflow( + workflow, where=where, host=host, port=port, input_path=input_path, command="workflow validate" + ) + + # --------------------------------------------------------------------------- # compose / fragment — fragment-based workflow composition # --------------------------------------------------------------------------- diff --git a/comfy_cli/command/workflow_fragments.py b/comfy_cli/command/workflow_fragments.py index e7a986fb..60a6da12 100644 --- a/comfy_cli/command/workflow_fragments.py +++ b/comfy_cli/command/workflow_fragments.py @@ -453,7 +453,7 @@ def fragment_show_cmd( renderer.emit(payload, command="workflow fragment show") -@fragment_app.command("validate", help="Validate that a fragment file is well-formed.") +@fragment_app.command("validate", help="Validate that a workflow-fragment file is well-formed") @tracking.track_command("workflow") def fragment_validate_cmd( fragment: Annotated[str, typer.Argument(help="Fragment name (looked up in --lib) or path to .json.")], diff --git a/comfy_cli/discovery.py b/comfy_cli/discovery.py index 2b7b86a9..83ce4c42 100644 --- a/comfy_cli/discovery.py +++ b/comfy_cli/discovery.py @@ -40,6 +40,9 @@ "comfy update": "update", # help / validation "comfy help": "help", + # `comfy workflow validate` is the canonical home; `comfy validate` is the + # hidden deprecated alias — both emit the same workflow-validation payload. + "comfy workflow validate": "workflow", "comfy validate": "workflow", # nodes introspection "comfy nodes ls": "nodes", diff --git a/comfy_cli/skills/comfy/SKILL.md b/comfy_cli/skills/comfy/SKILL.md index 7b2eee42..187deba9 100644 --- a/comfy_cli/skills/comfy/SKILL.md +++ b/comfy_cli/skills/comfy/SKILL.md @@ -726,7 +726,7 @@ Before `comfy run`, verify the workflow will succeed: comfy run --workflow wf.json --print-prompt # Full pre-flight: checks class_types, input shapes, enum values, edge wiring -comfy --json validate --workflow api.json +comfy --json workflow validate --workflow api.json # Spot-check a single node class exists on the target comfy --json nodes show @@ -911,7 +911,7 @@ Hard-won lessons per domain. Not a tutorial — a reference card. - Assembly: GetVideoComponents → BatchImagesNode → CreateVideo → SaveVideo (ImageBatch is deprecated) - Autogrow inputs (type COMFY_AUTOGROW_*, e.g. BatchImagesNode `images`): wire ONE slot key per connection — `"images.image0": [..], "images.image1": [..]` — never a single `images` link. - `nodes show` prints the `wire_as` form; `comfy validate` rejects the bare form before submit. + `nodes show` prints the `wire_as` form; `comfy workflow validate` rejects the bare form before submit. - I2V pattern: LoadImage → I2VNode → SaveVideo (check `nodes show` for the I2V node) - **Model enums mix t2v and i2v variants** — a node's `model` choices may include image-to-video-only models (e.g. `grok-imagine-video-1.5`) that fail at runtime diff --git a/tests/comfy_cli/command/test_workflow_validate.py b/tests/comfy_cli/command/test_workflow_validate.py new file mode 100644 index 00000000..a99b2bd6 --- /dev/null +++ b/tests/comfy_cli/command/test_workflow_validate.py @@ -0,0 +1,235 @@ +"""`comfy workflow validate` (canonical) + the deprecated `comfy validate` alias. + +Both entry points share one implementation (`workflow.validate_api_workflow`). +These tests drive them offline via `--input ` so no ComfyUI +server is needed, and assert the alias stays behavior-compatible while warning. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from typer.testing import CliRunner + +from comfy_cli.caller import Caller +from comfy_cli.cmdline import app as root_app +from comfy_cli.command import workflow as workflow_cmd +from comfy_cli.output.renderer import ( + OutputMode, + Renderer, + reset_renderer_for_testing, + set_renderer, +) + + +@pytest.fixture(autouse=True) +def reset_singleton(): + reset_renderer_for_testing() + yield + reset_renderer_for_testing() + + +def _force_json_renderer(): + r = Renderer.resolve( + is_stdout_tty=False, + env={}, + caller=Caller(kind="user", agentic=False, source_env=None), + json_flag=True, + ) + r.mode = OutputMode.JSON + set_renderer(r) + return r + + +def _object_info() -> dict: + """Minimal object_info: one output node with a single INT widget input.""" + return { + "TrivialNode": { + "input": {"required": {"value": ["INT", {"default": 0}]}}, + "input_order": {"required": ["value"]}, + "output": [], + "output_name": [], + "category": "testing", + "display_name": "Trivial", + "description": "A trivial node for tests.", + "output_node": True, + "python_module": "nodes", + } + } + + +def _valid_workflow() -> dict: + return {"1": {"class_type": "TrivialNode", "inputs": {"value": 5}}} + + +def _invalid_workflow() -> dict: + # `value` is declared INT; a string is a shape mismatch -> a validation error. + return {"1": {"class_type": "TrivialNode", "inputs": {"value": "not-an-int"}}} + + +def _write(tmp_path: Path, name: str, data: dict) -> str: + p = tmp_path / name + p.write_text(json.dumps(data), encoding="utf-8") + return str(p) + + +def _run(app, args: list[str], capsys): + """Invoke a command and return (exit_code, envelope, combined_output).""" + _force_json_renderer() + runner = CliRunner() + # Default standalone_mode so `typer.Exit(code=1)` maps to result.exit_code. + result = runner.invoke(app, args) + captured = capsys.readouterr() + combined = (captured.out or "") + (captured.err or "") + if not combined.strip(): + # CliRunner intercepted the streams itself. Include BOTH stdout and + # stderr: in JSON mode the deprecation warning is routed to stderr by + # design (so structured stdout stays clean), and newer click (>=8.2) + # captures stderr separately from stdout instead of mixing it in. + combined = result.stdout or "" + try: + combined += result.stderr or "" + except ValueError: + # Older click mixes stderr into stdout; nothing separate to add. + pass + envelope = None + for line in reversed(combined.strip().splitlines()): + line = line.strip() + if not line: + continue + try: + envelope = json.loads(line) + break + except json.JSONDecodeError: + continue + return result.exit_code, envelope, combined + + +def test_workflow_validate_valid(tmp_path, capsys): + oi = _write(tmp_path, "oi.json", _object_info()) + wf = _write(tmp_path, "wf.json", _valid_workflow()) + + code, env, _ = _run(workflow_cmd.app, ["validate", "--workflow", wf, "--input", oi], capsys) + + assert code == 0 + assert env is not None + assert env["ok"] is True + assert env["command"] == "workflow validate" + assert env["data"]["valid"] is True + assert env["data"]["error_count"] == 0 + + +def test_workflow_validate_invalid_exits_nonzero(tmp_path, capsys): + oi = _write(tmp_path, "oi.json", _object_info()) + wf = _write(tmp_path, "wf.json", _invalid_workflow()) + + code, env, _ = _run(workflow_cmd.app, ["validate", "--workflow", wf, "--input", oi], capsys) + + assert code == 1 + assert env is not None + assert env["ok"] is False + assert env["data"]["valid"] is False + assert env["data"]["error_count"] >= 1 + + +def test_deprecated_alias_still_works_and_warns(tmp_path, capsys): + oi = _write(tmp_path, "oi.json", _object_info()) + wf = _write(tmp_path, "wf.json", _valid_workflow()) + + code, env, combined = _run(root_app, ["validate", "--workflow", wf, "--input", oi], capsys) + + # Still functional... + assert code == 0 + assert env is not None + assert env["ok"] is True + assert env["data"]["valid"] is True + # ...but the alias labels itself and warns to point at the canonical home. + assert env["command"] == "validate" + assert "deprecated" in combined.lower() + assert "comfy workflow validate" in combined + + +def test_alias_matches_canonical_validation_payload(tmp_path, capsys): + """The deprecated alias must produce the same validation verdict as the + canonical command for the same inputs (only the `command` label differs).""" + oi = _write(tmp_path, "oi.json", _object_info()) + wf = _write(tmp_path, "wf.json", _valid_workflow()) + + _, canonical, _ = _run(workflow_cmd.app, ["validate", "--workflow", wf, "--input", oi], capsys) + _, alias, _ = _run(root_app, ["validate", "--workflow", wf, "--input", oi], capsys) + + for key in ("valid", "error_count", "warning_count", "errors", "warnings"): + assert canonical["data"][key] == alias["data"][key] + + +def test_invalid_where_emits_structured_error(tmp_path, capsys): + """A bad --where value fails fast with an envelope, not a raw traceback.""" + wf = _write(tmp_path, "wf.json", _valid_workflow()) + + code, env, _ = _run(workflow_cmd.app, ["validate", "--workflow", wf, "--where", "clod"], capsys) + + assert code == 1 + assert env is not None + assert env["ok"] is False + assert env["error"]["code"] == "where_invalid" + + +def test_invalid_where_default_falls_back_instead_of_failing(tmp_path, capsys, monkeypatch): + """A bad *configured* default (no explicit --where) never breaks the command. + + Only an explicit bad flag is a user error worth failing on; a stale env or + config value drops to the local default, matching the pre-existing routing + behavior of the other commands. + """ + oi = _write(tmp_path, "oi.json", _object_info()) + wf = _write(tmp_path, "wf.json", _valid_workflow()) + monkeypatch.setenv("COMFY_WHERE", "clod") + + code, env, _ = _run(workflow_cmd.app, ["validate", "--workflow", wf, "--input", oi], capsys) + + assert code == 0 + assert env is not None + assert env["ok"] is True + assert env["data"]["valid"] is True + + +def test_non_utf8_workflow_emits_structured_error(tmp_path, capsys): + """A non-UTF-8 workflow file reports an envelope instead of crashing.""" + oi = _write(tmp_path, "oi.json", _object_info()) + bad = tmp_path / "bad.json" + bad.write_bytes(b"\xff\xfe\x00not-utf8") + + code, env, _ = _run(workflow_cmd.app, ["validate", "--workflow", str(bad), "--input", oi], capsys) + + assert code == 1 + assert env is not None + assert env["ok"] is False + assert env["error"]["code"] == "workflow_read_error" + + +def test_pretty_output_escapes_workflow_markup(tmp_path, capsys): + """Workflow-supplied node ids must not inject Rich markup into pretty output.""" + from comfy_cli.caller import Caller + + oi = _write(tmp_path, "oi.json", _object_info()) + # An unknown class_type under a node id carrying Rich markup -> an error whose + # pretty rendering echoes the (escaped) node id. + wf = _write(tmp_path, "wf.json", {"[green]spoof[/green]": {"class_type": "NoSuchNode", "inputs": {}}}) + + reset_renderer_for_testing() + r = Renderer.resolve( + is_stdout_tty=True, + env={}, + caller=Caller(kind="user", agentic=False, source_env=None), + ) + r.mode = OutputMode.PRETTY + set_renderer(r) + + result = CliRunner().invoke(workflow_cmd.app, ["validate", "--workflow", wf, "--input", oi]) + out = result.stdout or "" + + assert result.exit_code == 1 + # The literal markup survives verbatim (escaped) rather than being interpreted. + assert "[green]spoof[/green]" in out