from __future__ import annotations from collections.abc import Callable import io import importlib import os import subprocess import sys from pathlib import Path from types import SimpleNamespace from typing import NoReturn, cast import apprc as rc import pytest from rich.console import Console from typer.testing import CliRunner import kneiff.cli.comfy as cli_comfy import kneiff.cli.dataset as cli_dataset from kneiff.config import ( COMFY_I2I_LORA_1_ENV_KEY, COMFY_I2I_LORA_2_ENV_KEY, COMFY_I2I_LORA_ENV_KEY, COMFY_I2I_LORA_STRENGTH_1_ENV_KEY, COMFY_I2I_LORA_STRENGTH_2_ENV_KEY, COMFY_I2I_LORA_STRENGTH_ENV_KEY, COMFY_I2I_MODEL_DUO_ENV_KEY, COMFY_I2I_MODEL_SOLO_ENV_KEY, COMFY_LORAS_DIR_1_ENV_KEY, COMFY_LORAS_DIR_2_ENV_KEY, COMFY_MODELS_DIR_ENV_KEY, COMFY_OUTPUT_DIR_ENV_KEY, COMFY_POLL_INTERVAL_SECONDS_ENV_KEY, COMFY_PROMPT_TIMEOUT_SECONDS_ENV_KEY, COMFY_UPSCALE_MODEL_ENV_KEY, COMFY_UPSCALE_DENOISE_BASE_ENV_KEY, COMFY_UPSCALE_LORA_ENV_KEY, COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY, COMFY_UPSCALE_REFINER_UNET_ENV_KEY, COMFY_UPSCALE_REFINER_VAE_ENV_KEY, COMFY_URL_ENV_KEY, COMFY_T2I_LORA_DUO_1_ENV_KEY, COMFY_T2I_LORA_DUO_2_ENV_KEY, COMFY_T2I_LORA_SOLO_ENV_KEY, COMFY_T2I_LORA_STRENGTH_DUO_1_ENV_KEY, COMFY_T2I_LORA_STRENGTH_DUO_2_ENV_KEY, COMFY_T2I_LORA_STRENGTH_SOLO_ENV_KEY, COMFY_T2I_MODEL_DUO_ENV_KEY, COMFY_T2I_MODEL_SOLO_ENV_KEY, DEFAULT_COMFY_UPSCALE_MODEL, DEFAULT_COMFY_UPSCALE_DENOISE_BASE, DEFAULT_COMFY_UPSCALE_LORA, DEFAULT_COMFY_UPSCALE_LORA_STRENGTH, DEFAULT_COMFY_UPSCALE_LORA_TOKEN, DEFAULT_COMFY_UPSCALE_REFINER_CLIP, DEFAULT_COMFY_UPSCALE_REFINER_CLIP_TYPE, DEFAULT_COMFY_UPSCALE_REFINER_UNET, DEFAULT_COMFY_UPSCALE_REFINER_VAE, KNF_APPRC_TOML_ENV_KEY, KNEIFF_RC, ComfyConfig, KneiffConfig, LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP, LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP_TYPE, LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_UNET, LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_VAE, ) from kneiff.infer.comfy.client import ComfyUploadedImage from kneiff.progress import ProgressUpdate import kneiff.utils.image.tag_jtp3 as tag_jtp3 from tests._cli_helpers import invoke_cli pytestmark = pytest.mark.usefixtures("isolated_cli_project") SHOWCASE_PROMPTS_PATH = ( Path(__file__).parent / "fixtures" / "projects" / "aurora" / "prompts.knf.yaml" ) def _write_noob_showcase_prompts(path: Path) -> None: """Write one standalone catalog usable by NOOB CLI routing tests.""" path.write_text( """ version: 1 defaults: seed: 123 prompts: - id: aurora_portrait uses: [showcase, training_validation] captions: nlg: - Aurora_Fox portrait pony: - score_9, Aurora_Fox, portrait noob: - Aurora_Fox, portrait """, encoding="utf-8", ) _DEFAULT_COMFY_INPUT_OPTIONS = { ("UpscaleModelLoader", "model_name"): ( DEFAULT_COMFY_UPSCALE_MODEL, "local-upscaler.safetensors", "cli-upscaler.safetensors", "4x-test.safetensors", ), ("UNETLoader", "unet_name"): ( DEFAULT_COMFY_UPSCALE_REFINER_UNET, LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_UNET, "z-image.safetensors", ), ("CLIPLoader", "clip_name"): ( DEFAULT_COMFY_UPSCALE_REFINER_CLIP, LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP, "qwen.safetensors", ), ("CLIPLoader", "type"): ( DEFAULT_COMFY_UPSCALE_REFINER_CLIP_TYPE, LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP_TYPE, ), ("VAELoader", "vae_name"): ( DEFAULT_COMFY_UPSCALE_REFINER_VAE, LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_VAE, ), } class _FakeComfyModelClient: def __init__( self, server_url: str, options: dict[tuple[str, str], tuple[str, ...]], ) -> None: self.server_url = server_url.rstrip("/") self.options = options self.input_option_requests: list[tuple[str, str]] = [] def input_options(self, node_class: str, input_name: str) -> tuple[str, ...]: key = (node_class, input_name) self.input_option_requests.append(key) return self.options.get(key, ()) class _CapturedShowcaseProgress: """Record aggregate CLI progress without rendering a terminal bar.""" def __init__( self, title: str, total: int, console: Console | None, ) -> None: self.title = title self.total = total self.console = console self.advances: list[str | None] = [] self.exit_error: type[BaseException] | None = None def __enter__(self) -> _CapturedShowcaseProgress: return self def __exit__(self, *exc_info: object) -> None: error_type = exc_info[0] self.exit_error = ( error_type if isinstance(error_type, type) and issubclass(error_type, BaseException) else None ) def advance(self, text: str | None = None, *, amount: int = 1) -> None: self.advances.extend([text] * amount) def step(self, text: str, *, amount: int = 1) -> None: self.total += amount self.advance(text, amount=amount) def apply(self, update: ProgressUpdate) -> None: """Record one shared progress update.""" self.total += update.additional_total self.advance(update.description, amount=update.advance) def _capturing_showcase_progress_factory( reporters: list[_CapturedShowcaseProgress], ) -> Callable[..., _CapturedShowcaseProgress]: def create( title: str, total: int = 0, console: Console | None = None, ) -> _CapturedShowcaseProgress: reporter = _CapturedShowcaseProgress(title, total, console) reporters.append(reporter) return reporter return create def _install_fake_comfy_model_client( monkeypatch: pytest.MonkeyPatch, *, options: dict[tuple[str, str], tuple[str, ...]] | None = None, ) -> _FakeComfyModelClient: client = _FakeComfyModelClient( "http://127.0.0.1:8188", options or dict(_DEFAULT_COMFY_INPUT_OPTIONS), ) monkeypatch.setattr(cli_comfy, "ComfyUiClient", lambda server_url: client) return client @pytest.fixture(autouse=True) def _isolate_comfy_env(monkeypatch: pytest.MonkeyPatch) -> None: import kneiff.infer.comfy.showcase_grid as showcase_grid_module owner = rc.schema.owner_for(ComfyConfig) for field in owner.fields: monkeypatch.delenv(owner.env_key(field.name), raising=False) monkeypatch.setattr( showcase_grid_module, "write_showcase_grid", lambda rows, output_subfolder, filename, **kwargs: ComfyUploadedImage( name=filename, subfolder=output_subfolder, type="output", ), ) def test_cli_help_smoke() -> None: runner = CliRunner() for args in ( ["--help"], ["comfy", "--help"], ["comfy", "showcase", "--help"], ["comfy", "upscale", "--help"], ["comfy", "outpaint", "--help"], ["comfy", "t2i", "--help"], ["comfy", "t2i", "solo", "--help"], ["comfy", "t2i", "duo", "--help"], ["config", "--help"], ["config", "paths", "--help"], ["config", "show", "--help"], ["config", "doctor", "--help"], ["config", "setup", "--help"], ["config", "set", "--help"], ["config", "edit", "--help"], ["config", "app", "--help"], ["config", "app", "init", "--help"], ["config", "storage", "--help"], ["config", "storage", "add", "--help"], ["config", "storage", "list", "--help"], ["config", "storage", "remove", "--help"], ["img", "--help"], ["img", "tag", "--help"], ["llm", "--help"], ["llm", "prompt", "--help"], ["dataset", "--help"], ["dataset", "aspects", "--help"], ["dataset", "describe", "--help"], ["model", "--help"], ["model", "convert", "--help"], ["train", "--help"], ["train", "start", "--help"], ["train", "prepare", "--help"], ["train", "runs", "--help"], ["train", "review", "--help"], ["train", "grid", "--help"], ): result = invoke_cli(runner, args) assert result.exit_code == 0 def test_comfy_t2i_help_exposes_preset_specific_options() -> None: runner = CliRunner() solo = invoke_cli(runner, ["comfy", "t2i", "solo", "--help"]) duo = invoke_cli(runner, ["comfy", "t2i", "duo", "--help"]) assert solo.exit_code == 0 assert "--pipeline" in solo.output assert "--activation-token-1" in solo.output assert "--t2i-lora-strength-1" in solo.output assert "--num-cleanups" in solo.output assert "--i2i-model" in solo.output assert "--i2i-lora-strength" in solo.output assert duo.exit_code == 0 assert "--activation-token-2" in duo.output assert "--num-cleanups" in duo.output assert "--i2i-lora-strength-2" in duo.output def test_comfy_t2i_solo_requires_pipeline_outside_an_interactive_terminal() -> None: runner = CliRunner() result = invoke_cli(runner, ["comfy", "t2i", "solo", "studio portrait"]) assert result.exit_code == 2 assert "--pipeline" in result.output def test_showcase_workflow_picker_keeps_project_order_without_duplicates() -> None: picker_names = cli_comfy._workflow_picker_preset_names( ["pony", "project-only", "anima", "pony", "noob", "project-only"] ) assert picker_names == ["pony", "project-only", "anima", "noob"] def test_cli_root_help_shows_canonical_image_group_only() -> None: runner = CliRunner() result = invoke_cli(runner, ["--help"]) assert result.exit_code == 0 assert "│ img" in result.output assert "│ image" not in result.output assert "infer" not in result.output def test_image_group_alias_is_removed() -> None: runner = CliRunner() result = invoke_cli(runner, ["image", "--help"]) assert result.exit_code != 0 assert "No such command 'image'" in result.output def test_infer_group_is_removed() -> None: runner = CliRunner() result = invoke_cli(runner, ["infer", "--help"]) assert result.exit_code != 0 assert "No such command 'infer'" in result.output def test_config_uses_native_apprc_index_path( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.delenv(KNF_APPRC_TOML_ENV_KEY, raising=False) monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config-home")) expected_path = tmp_path / "config-home" / "knf" / "knf.apprc.toml" assert KNEIFF_RC.spec.index_env_key == KNF_APPRC_TOML_ENV_KEY assert KNEIFF_RC.spec.default_index_path() == expected_path assert KNEIFF_RC.spec.index_path() == expected_path def test_cli_help_shows_batch_config_and_resume_run_usage() -> None: runner = CliRunner() start_result = invoke_cli(runner, ["train", "start", "--help"]) prepare_result = invoke_cli(runner, ["train", "prepare", "--help"]) dataset_result = invoke_cli(runner, ["dataset", "plan", "--help"]) assert start_result.exit_code == 0 assert "[CONFIG]..." in start_result.output assert "knf train start CONFIG..." in start_result.output assert "knf train start CONFIG... --resume RUN" in start_result.output assert "Pass RUN after --resume" in start_result.output assert prepare_result.exit_code == 0 assert "[CONFIG]..." in prepare_result.output assert dataset_result.exit_code == 0 assert "[CONFIG]..." in dataset_result.output def test_comfy_showcase_help_shows_prompt_field() -> None: runner = CliRunner() output_option = "--" + "output" result = invoke_cli(runner, ["comfy", "showcase", "--help"]) assert result.exit_code == 0 assert "--lora" in result.output assert "--training" in result.output assert "[STEPS]..." in result.output assert "--prompts" in result.output assert "--prompt-field" in result.output assert output_option not in result.output assert "auto" in result.output assert "packaged preset" in result.output assert "noob-nova" in result.output assert "showcase_.workflow.json" in result.output def test_comfy_upscale_help_describes_directory_inputs() -> None: runner = CliRunner() result = invoke_cli(runner, ["comfy", "upscale", "--help"]) normalized_help = " ".join(result.output.replace("│", " ").split()) assert result.exit_code == 0 assert "One or more image files or directories to upscale." in normalized_help assert "Directory inputs are scanned recursively by default." in normalized_help assert "Include images in nested subdirectories" in normalized_help assert "--recursive" in result.output assert "--no-recursive" in result.output def test_comfy_outpaint_help_describes_packaged_file_only_workflows() -> None: runner = CliRunner() result = invoke_cli(runner, ["comfy", "outpaint", "--help"]) normalized_help = " ".join(result.output.replace("│", " ").split()) assert result.exit_code == 0 assert "Omit them to consume newline-delimited SOURCE/... paths from stdin" in ( normalized_help ) assert "Directory inputs are rejected" in normalized_help assert "--workflow" in result.output assert "--safe-border" in result.output assert "without ComfyUI inference" in normalized_help assert "--seed" in result.output def test_comfy_t2i_solo_help_uses_positional_prompt() -> None: runner = CliRunner() result = invoke_cli(runner, ["comfy", "t2i", "solo", "--help"]) normalized_help = " ".join(result.output.replace("│", " ").split()) assert result.exit_code == 0 assert "comfy t2i solo [OPTIONS] PROMPT" in normalized_help assert "Scene request used to generate reviewed prompts." in normalized_help def test_model_showcase_command_is_removed() -> None: runner = CliRunner() result = invoke_cli(runner, ["model", "showcase", "--help"]) assert result.exit_code != 0 def test_comfy_config_schema_declares_custom_prefix_fields() -> None: owner = rc.schema.owner_for(ComfyConfig) expected_env_keys = { "url": COMFY_URL_ENV_KEY, "models_dir": COMFY_MODELS_DIR_ENV_KEY, "loras_dir_1": COMFY_LORAS_DIR_1_ENV_KEY, "loras_dir_2": COMFY_LORAS_DIR_2_ENV_KEY, "output_dir": COMFY_OUTPUT_DIR_ENV_KEY, "t2i_model_solo": COMFY_T2I_MODEL_SOLO_ENV_KEY, "t2i_lora_solo": COMFY_T2I_LORA_SOLO_ENV_KEY, "t2i_lora_strength_solo": COMFY_T2I_LORA_STRENGTH_SOLO_ENV_KEY, "i2i_model_solo": COMFY_I2I_MODEL_SOLO_ENV_KEY, "i2i_lora": COMFY_I2I_LORA_ENV_KEY, "i2i_lora_strength": COMFY_I2I_LORA_STRENGTH_ENV_KEY, "t2i_model_duo": COMFY_T2I_MODEL_DUO_ENV_KEY, "t2i_lora_duo_1": COMFY_T2I_LORA_DUO_1_ENV_KEY, "t2i_lora_duo_2": COMFY_T2I_LORA_DUO_2_ENV_KEY, "t2i_lora_strength_duo_1": COMFY_T2I_LORA_STRENGTH_DUO_1_ENV_KEY, "t2i_lora_strength_duo_2": COMFY_T2I_LORA_STRENGTH_DUO_2_ENV_KEY, "i2i_model_duo": COMFY_I2I_MODEL_DUO_ENV_KEY, "i2i_lora_1": COMFY_I2I_LORA_1_ENV_KEY, "i2i_lora_2": COMFY_I2I_LORA_2_ENV_KEY, "i2i_lora_strength_1": COMFY_I2I_LORA_STRENGTH_1_ENV_KEY, "i2i_lora_strength_2": COMFY_I2I_LORA_STRENGTH_2_ENV_KEY, "upscale_model": COMFY_UPSCALE_MODEL_ENV_KEY, "upscale_refiner_unet": COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "upscale_refiner_clip": COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "upscale_refiner_clip_type": COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY, "upscale_refiner_vae": COMFY_UPSCALE_REFINER_VAE_ENV_KEY, "upscale_lora": COMFY_UPSCALE_LORA_ENV_KEY, "upscale_lora_token": COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, "upscale_lora_strength": COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, "upscale_denoise_base": COMFY_UPSCALE_DENOISE_BASE_ENV_KEY, "prompt_timeout_seconds": COMFY_PROMPT_TIMEOUT_SECONDS_ENV_KEY, "poll_interval_seconds": COMFY_POLL_INTERVAL_SECONDS_ENV_KEY, } assert owner in KNEIFF_RC.spec.owners assert { field.name: owner.env_key(field.name) for field in owner.fields } == expected_env_keys for field_name in expected_env_keys: assert owner.config_path_text(field_name) == f"comfy.{field_name}" def test_kneiff_config_uses_comfy_defaults( monkeypatch: pytest.MonkeyPatch, ) -> None: for env_key in ( COMFY_URL_ENV_KEY, COMFY_MODELS_DIR_ENV_KEY, COMFY_LORAS_DIR_1_ENV_KEY, COMFY_LORAS_DIR_2_ENV_KEY, COMFY_OUTPUT_DIR_ENV_KEY, COMFY_T2I_MODEL_SOLO_ENV_KEY, COMFY_T2I_LORA_SOLO_ENV_KEY, COMFY_T2I_LORA_STRENGTH_SOLO_ENV_KEY, COMFY_I2I_MODEL_SOLO_ENV_KEY, COMFY_I2I_LORA_ENV_KEY, COMFY_I2I_LORA_STRENGTH_ENV_KEY, COMFY_T2I_MODEL_DUO_ENV_KEY, COMFY_T2I_LORA_DUO_1_ENV_KEY, COMFY_T2I_LORA_DUO_2_ENV_KEY, COMFY_T2I_LORA_STRENGTH_DUO_1_ENV_KEY, COMFY_T2I_LORA_STRENGTH_DUO_2_ENV_KEY, COMFY_I2I_MODEL_DUO_ENV_KEY, COMFY_I2I_LORA_1_ENV_KEY, COMFY_I2I_LORA_2_ENV_KEY, COMFY_I2I_LORA_STRENGTH_1_ENV_KEY, COMFY_I2I_LORA_STRENGTH_2_ENV_KEY, COMFY_UPSCALE_MODEL_ENV_KEY, COMFY_UPSCALE_REFINER_UNET_ENV_KEY, COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY, COMFY_UPSCALE_REFINER_VAE_ENV_KEY, COMFY_UPSCALE_LORA_ENV_KEY, COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, COMFY_UPSCALE_DENOISE_BASE_ENV_KEY, COMFY_PROMPT_TIMEOUT_SECONDS_ENV_KEY, COMFY_POLL_INTERVAL_SECONDS_ENV_KEY, ): monkeypatch.delenv(env_key, raising=False) config = KneiffConfig().comfy assert config.url == "http://127.0.0.1:8188" assert config.models_dir == "" assert config.loras_dir_1 == "" assert config.loras_dir_2 == "" assert config.output_dir == "" assert config.t2i_lora_strength_solo == 1.0 assert config.i2i_lora_strength == 1.0 assert config.t2i_lora_strength_duo_1 == 1.0 assert config.t2i_lora_strength_duo_2 == 1.0 assert config.i2i_lora_strength_1 == 1.0 assert config.i2i_lora_strength_2 == 1.0 assert config.upscale_model == DEFAULT_COMFY_UPSCALE_MODEL assert config.upscale_refiner_unet == DEFAULT_COMFY_UPSCALE_REFINER_UNET assert config.upscale_refiner_clip == DEFAULT_COMFY_UPSCALE_REFINER_CLIP assert config.upscale_refiner_clip_type == DEFAULT_COMFY_UPSCALE_REFINER_CLIP_TYPE assert config.upscale_refiner_vae == DEFAULT_COMFY_UPSCALE_REFINER_VAE assert config.upscale_lora == DEFAULT_COMFY_UPSCALE_LORA assert config.upscale_lora_token == DEFAULT_COMFY_UPSCALE_LORA_TOKEN assert config.upscale_lora_strength == DEFAULT_COMFY_UPSCALE_LORA_STRENGTH assert config.upscale_denoise_base == DEFAULT_COMFY_UPSCALE_DENOISE_BASE assert config.prompt_timeout_seconds == 3600.0 assert config.poll_interval_seconds == 1.0 def test_kneiff_config_uses_comfy_env_values( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv(COMFY_URL_ENV_KEY, "http://comfy.example") monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, "/models") monkeypatch.setenv(COMFY_LORAS_DIR_1_ENV_KEY, "/models/models/loras/project") monkeypatch.setenv(COMFY_UPSCALE_MODEL_ENV_KEY, "4x-test.safetensors") monkeypatch.setenv(COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "z-image.safetensors") monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "qwen.safetensors") monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY, "lumina2") monkeypatch.setenv(COMFY_UPSCALE_REFINER_VAE_ENV_KEY, "ae.safetensors") monkeypatch.setenv(COMFY_UPSCALE_LORA_ENV_KEY, "project/character.safetensors") monkeypatch.setenv(COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, "char_token") monkeypatch.setenv(COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, "0.75") monkeypatch.setenv(COMFY_UPSCALE_DENOISE_BASE_ENV_KEY, "0.18") monkeypatch.setenv(COMFY_PROMPT_TIMEOUT_SECONDS_ENV_KEY, "12.5") monkeypatch.setenv(COMFY_POLL_INTERVAL_SECONDS_ENV_KEY, "0.25") config = KneiffConfig().comfy assert config == ComfyConfig( url="http://comfy.example", models_dir="/models", loras_dir_1="/models/models/loras/project", upscale_model="4x-test.safetensors", upscale_refiner_unet="z-image.safetensors", upscale_refiner_clip="qwen.safetensors", upscale_refiner_clip_type="lumina2", upscale_refiner_vae="ae.safetensors", upscale_lora="project/character.safetensors", upscale_lora_token="char_token", upscale_lora_strength=0.75, upscale_denoise_base=0.18, prompt_timeout_seconds=12.5, poll_interval_seconds=0.25, ) def test_kneiff_config_loads_every_t2i_environment_field( monkeypatch: pytest.MonkeyPatch, ) -> None: values = { COMFY_LORAS_DIR_1_ENV_KEY: "primary", COMFY_LORAS_DIR_2_ENV_KEY: "partner", COMFY_OUTPUT_DIR_ENV_KEY: "/comfy/output", COMFY_T2I_MODEL_SOLO_ENV_KEY: "solo-model.safetensors", COMFY_T2I_LORA_SOLO_ENV_KEY: "solo-lora.safetensors", COMFY_T2I_LORA_STRENGTH_SOLO_ENV_KEY: "0.11", COMFY_I2I_MODEL_SOLO_ENV_KEY: "solo-edit.safetensors", COMFY_I2I_LORA_ENV_KEY: "solo-edit-lora.safetensors", COMFY_I2I_LORA_STRENGTH_ENV_KEY: "0.22", COMFY_T2I_MODEL_DUO_ENV_KEY: "duo-model.safetensors", COMFY_T2I_LORA_DUO_1_ENV_KEY: "duo-one.safetensors", COMFY_T2I_LORA_DUO_2_ENV_KEY: "duo-two.safetensors", COMFY_T2I_LORA_STRENGTH_DUO_1_ENV_KEY: "0.33", COMFY_T2I_LORA_STRENGTH_DUO_2_ENV_KEY: "0.44", COMFY_I2I_MODEL_DUO_ENV_KEY: "duo-edit.safetensors", COMFY_I2I_LORA_1_ENV_KEY: "edit-one.safetensors", COMFY_I2I_LORA_2_ENV_KEY: "edit-two.safetensors", COMFY_I2I_LORA_STRENGTH_1_ENV_KEY: "0.55", COMFY_I2I_LORA_STRENGTH_2_ENV_KEY: "0.66", } for key, value in values.items(): monkeypatch.setenv(key, value) config = KneiffConfig().comfy assert config.loras_dir_1 == "primary" assert config.loras_dir_2 == "partner" assert config.output_dir == "/comfy/output" assert config.t2i_model_solo == "solo-model.safetensors" assert config.t2i_lora_solo == "solo-lora.safetensors" assert config.t2i_lora_strength_solo == 0.11 assert config.i2i_model_solo == "solo-edit.safetensors" assert config.i2i_lora == "solo-edit-lora.safetensors" assert config.i2i_lora_strength == 0.22 assert config.t2i_model_duo == "duo-model.safetensors" assert config.t2i_lora_duo_1 == "duo-one.safetensors" assert config.t2i_lora_duo_2 == "duo-two.safetensors" assert config.t2i_lora_strength_duo_1 == 0.33 assert config.t2i_lora_strength_duo_2 == 0.44 assert config.i2i_model_duo == "duo-edit.safetensors" assert config.i2i_lora_1 == "edit-one.safetensors" assert config.i2i_lora_2 == "edit-two.safetensors" assert config.i2i_lora_strength_1 == 0.55 assert config.i2i_lora_strength_2 == 0.66 def test_removed_comfy_loras_dir_environment_key_is_ignored( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.delenv(COMFY_LORAS_DIR_1_ENV_KEY, raising=False) monkeypatch.setenv("COMFY_LORAS_DIR", "legacy-directory") assert KneiffConfig().comfy.loras_dir_1 == "" def test_kneiff_config_rejects_invalid_comfy_float_env( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv(COMFY_PROMPT_TIMEOUT_SECONDS_ENV_KEY, "slow") with pytest.raises(ExceptionGroup, match="ComfySettings") as exc_info: KneiffConfig() assert "prompt_timeout_seconds" in repr(exc_info.value) def test_cli_config_error_is_concise_and_does_not_expose_secrets( tmp_path: Path, isolated_cli_storage: Path, ) -> None: secret = "CLI_CONFIG_SECRET_SENTINEL_7a91" env = os.environ.copy() env.update( { "COMFY_PROMPT_TIMEOUT_SECONDS": "slow", "KNF_LMSTUDIO_API_KEY": secret, "KNF_STORAGE": str(isolated_cli_storage), "XDG_CONFIG_HOME": str(tmp_path / "config-home"), } ) result = subprocess.run( [ sys.executable, "-m", "kneiff.cli.app", "--skip-dotenv-layers", "dataset", "plan", ], capture_output=True, text=True, check=False, env=env, ) output = result.stdout + result.stderr assert result.returncode == 2 assert "Invalid Kneiff runtime config" in output assert "prompt_timeout_seconds" in output assert secret not in output assert "Traceback" not in output def test_non_cli_bundle_resolves_named_storage_after_apprc_bootstrap( tmp_path: Path, ) -> None: config_home = tmp_path / "config-home" storage_root = tmp_path / "named-storage" storage_root.mkdir() (storage_root / ".env.apprc-storage").touch() index_path = config_home / "knf" / "knf.apprc.toml" rc.storage.register_storage( name="demo", root=storage_root, path=index_path, storage_env_filename=KNEIFF_RC.spec.storage_env_filename, ) env = os.environ.copy() env.update( { "KNF_STORAGE": "demo", "XDG_CONFIG_HOME": str(config_home), } ) result = subprocess.run( [ sys.executable, "-c", ( "from kneiff.config import KNEIFF_RC, KneiffConfig; " "KNEIFF_RC.bootstrap(); " "print(KneiffConfig().storage.root)" ), ], capture_output=True, text=True, check=False, env=env, ) assert result.returncode == 0, result.stdout + result.stderr assert result.stdout.strip() == str(storage_root.resolve()) def test_cli_rejects_unknown_log_level() -> None: runner = CliRunner() result = invoke_cli( runner, ["--log-level", "verbose-ish", "dataset", "plan"], ) assert result.exit_code == 2 assert "Unknown logging level: verbose-ish" in result.output def test_comfy_showcase_requires_storage( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.delenv("KNF_STORAGE", raising=False) monkeypatch.delenv(KNF_APPRC_TOML_ENV_KEY, raising=False) monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config-home")) prompts_path = tmp_path / "prompts.yaml" prompts_path.write_text( """ version: 1 defaults: seed: 123 prompts: - id: first uses: [showcase] captions: nlg: - prose prompt pony: - pony tags """, encoding="utf-8", ) runner = CliRunner() result = invoke_cli( runner, [ "comfy", "showcase", "-w", "pony", "-p", str(prompts_path), ], ) assert result.exit_code != 0 assert "KNF_STORAGE is required" in result.output def test_comfy_upscale_uses_storage_local_env( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, isolated_cli_storage: Path, ) -> None: import kneiff.infer.comfy.upscale as upscale_module from kneiff.infer.comfy.upscale import ( ComfyUpscaleRequest, ComfyUpscaleResult, ) (isolated_cli_storage / ".env.apprc-storage").write_text( ( f'{COMFY_UPSCALE_MODEL_ENV_KEY}="local-upscaler.safetensors"\n' f'{COMFY_UPSCALE_LORA_ENV_KEY}="missing.safetensors"\n' f'{COMFY_UPSCALE_LORA_TOKEN_ENV_KEY}="env_token"\n' ), encoding="utf-8", ) monkeypatch.delenv(COMFY_UPSCALE_MODEL_ENV_KEY, raising=False) image_path = tmp_path / "image.png" image_path.write_bytes(b"image") requests: list[ComfyUpscaleRequest] = [] def fake_run_upscale( request: ComfyUpscaleRequest, **kwargs: object, ) -> tuple[ComfyUpscaleResult, ...]: requests.append(request) return () monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale) _install_fake_comfy_model_client(monkeypatch) runner = CliRunner() result = invoke_cli(runner, ["comfy", "upscale", str(image_path), "--fast"]) assert result.exit_code == 0, result.output assert requests[0].upscale_model == "local-upscaler.safetensors" assert requests[0].fast is True assert requests[0].lora is None def test_comfy_upscale_uses_one_aggregate_prompt_progress_task( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import kneiff.infer.comfy.upscale as upscale_module from kneiff.infer.comfy.upscale import ( ComfyUpscaleRequest, ComfyUpscaleResult, ProgressCallback as ResultCallback, ) from kneiff.progress import ProgressCallback as WorkProgressCallback first = tmp_path / "first.png" second = tmp_path / "second.png" first.write_bytes(b"image") second.write_bytes(b"image") reporters: list[_CapturedShowcaseProgress] = [] def fake_run_upscale( request: ComfyUpscaleRequest, *, progress: ResultCallback | None = None, work_progress: WorkProgressCallback | None = None, **kwargs: object, ) -> tuple[ComfyUpscaleResult, ...]: del kwargs results: list[ComfyUpscaleResult] = [] for index, path in enumerate(request.inputs, start=1): result = ComfyUpscaleResult( input_path=path, prompt_id=f"prompt-{index}", filename_prefix=f"output-{index}", history={}, ) results.append(result) if work_progress is not None: work_progress( ProgressUpdate( f"Image {index}/2 · {path.name}", advance=1, ) ) if progress is not None: progress(result) return tuple(results) monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale) monkeypatch.setattr( cli_comfy, "CliProgress", _capturing_showcase_progress_factory(reporters), ) _install_fake_comfy_model_client(monkeypatch) result = invoke_cli( CliRunner(), [ "comfy", "upscale", str(first), str(second), "--fast", "--model", "cli-upscaler.safetensors", ], ) assert result.exit_code == 0, result.output assert len(reporters) == 1 assert reporters[0].title == "ComfyUI upscale" assert reporters[0].total == 2 assert reporters[0].advances == [ "Image 1/2 · first.png", "Image 2/2 · second.png", ] assert f"{first}: output-1" in result.output assert f"{second}: output-2" in result.output def test_dataset_aspects_streams_pruned_paths_with_dynamic_progress( monkeypatch: pytest.MonkeyPatch, ) -> None: import kneiff.training.lora.simpletuner_resolution_report as aspect_report reporters: list[_CapturedShowcaseProgress] = [] progress_descriptions = [ "Validated dataset paths", "Checking manifest image paths 1/2", "Built 2 planned geometry variants", "Read source dimensions 1/1", "Built SimpleTuner artifacts", "Analyzing backend fullbody 1/1", ] def fake_dataset_plan(*_args: object) -> cli_dataset.DatasetSyncPlan: return cast( cli_dataset.DatasetSyncPlan, SimpleNamespace(dataset_name="ready"), ) def fake_build_aspects_analysis( plan: cli_dataset.DatasetSyncPlan, *, process_count: int, progress: _CapturedShowcaseProgress, ) -> aspect_report.SimpleTunerAspectAnalysis: assert plan.dataset_name == "ready" assert process_count == 2 progress.step("Validated dataset paths") progress.apply( ProgressUpdate( "Checking manifest image paths 1/2", advance=1, additional_total=2, ) ) progress.apply( ProgressUpdate( "Built 2 planned geometry variants", advance=1, additional_total=1, ) ) progress.apply( ProgressUpdate( "Read source dimensions 1/1", advance=1, additional_total=1, ) ) progress.step("Built SimpleTuner artifacts") progress.apply( ProgressUpdate( "Analyzing backend fullbody 1/1", advance=1, additional_total=1, ) ) return cast(aspect_report.SimpleTunerAspectAnalysis, object()) monkeypatch.setattr(cli_dataset, "_dataset_plan_from_cli", fake_dataset_plan) monkeypatch.setattr( cli_dataset, "_build_dataset_aspects_analysis", fake_build_aspects_analysis, ) monkeypatch.setattr( aspect_report, "format_simpletuner_aspect_analysis", lambda _analysis: "Aspect report", ) monkeypatch.setattr( aspect_report, "pruned_aspect_bucket_source_paths", lambda _analysis: ("SOURCE/0-FULLBODY/scene.png",), ) monkeypatch.setattr( cli_dataset, "CliProgress", _capturing_showcase_progress_factory(reporters), ) result = invoke_cli( CliRunner(), ["dataset", "aspects", "ready", "--processes", "2"], ) assert result.exit_code == 0, result.output assert result.stdout == "SOURCE/0-FULLBODY/scene.png\n" assert "Aspect report" in result.stderr assert len(reporters) == 1 assert reporters[0].title == "dataset aspects: ready" assert reporters[0].total == 7 assert reporters[0].advances == progress_descriptions assert reporters[0].exit_error is None def test_comfy_outpaint_consumes_dataset_aspect_source_stream( isolated_cli_project: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: from PIL import Image import kneiff.infer.comfy.outpaint as outpaint_module source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png" source_path.parent.mkdir(parents=True, exist_ok=True) Image.new("RGB", (640, 960), color="white").save(source_path) requests: list[outpaint_module.ComfyOutpaintRequest] = [] def fake_run_outpaint( request: outpaint_module.ComfyOutpaintRequest, **_kwargs: object, ) -> tuple[outpaint_module.ComfyOutpaintResult, ...]: requests.append(request) return () monkeypatch.setattr(outpaint_module, "run_outpaint", fake_run_outpaint) result = invoke_cli( CliRunner(), ["comfy", "outpaint", "-w", "willy"], input="SOURCE/0-FULLBODY/scene.png\n", ) assert result.exit_code == 0, result.output assert requests[0].inputs == (source_path.resolve(),) assert requests[0].square_fraction == 0.25 def test_comfy_outpaint_safe_border_writes_pipeable_local_outputs_without_comfyui( isolated_cli_project: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: from PIL import Image source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png" source_path.parent.mkdir(parents=True, exist_ok=True) Image.new("RGB", (601, 800), color="white").save(source_path) output_root = isolated_cli_project / "safe-border-output" monkeypatch.setenv(COMFY_OUTPUT_DIR_ENV_KEY, str(output_root)) def fail_comfy_client(*_args: object, **_kwargs: object) -> NoReturn: raise AssertionError("--safe-border must not construct a ComfyUI client") monkeypatch.setattr(cli_comfy, "ComfyUiClient", fail_comfy_client) result = invoke_cli( CliRunner(), ["comfy", "outpaint", "--safe-border"], input="SOURCE/0-FULLBODY/scene.png\n", ) assert result.exit_code == 0, result.output written_paths = tuple(Path(line) for line in result.stdout.splitlines()) assert len(written_paths) == 1 assert written_paths[0].parent.parent == output_root assert written_paths[0].parent.name.count("-") == 2 assert "-safe-border-1" in written_paths[0].name assert written_paths[0].is_file() assert "Wrote 1 safe-border result" in result.stderr explicit_result = invoke_cli( CliRunner(), ["comfy", "outpaint", str(source_path), "--safe-border"], input="", ) assert explicit_result.exit_code == 0, explicit_result.output explicit_path = Path(explicit_result.stdout.strip()) assert explicit_path.is_file() assert explicit_path.parent.parent == output_root def test_comfy_outpaint_safe_border_from_pruned_aspects_needs_no_workflow( isolated_cli_project: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: from PIL import Image import kneiff.datasets.export.aspect_analysis as aspect_analysis import kneiff.infer.comfy.outpaint as outpaint_module import kneiff.training.lora.simpletuner_resolution_report as aspect_report source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png" source_path.parent.mkdir(parents=True, exist_ok=True) Image.new("RGB", (640, 960), color="white").save(source_path) config_path = isolated_cli_project / "configs" / "ANIMA.knf.yaml" config_path.parent.mkdir(parents=True, exist_ok=True) config_path.write_text( 'mappings:\n fullbody:\n - "0-FULLBODY"\n', encoding="utf-8", ) output_root = isolated_cli_project / "safe-border-output" monkeypatch.setenv(COMFY_OUTPUT_DIR_ENV_KEY, str(output_root)) requests: list[outpaint_module.SafeBorderRequest] = [] def fake_build_aspect_analysis( plan: object, *, process_count: int, progress_callback: object, ) -> object: assert getattr(plan, "dataset_name") == "ANIMA" assert process_count == 1 assert callable(progress_callback) return object() def fake_run_safe_border( request: outpaint_module.SafeBorderRequest, **_kwargs: object, ) -> tuple[outpaint_module.SafeBorderResult, ...]: requests.append(request) return () monkeypatch.setattr( aspect_analysis, "build_dataset_aspect_analysis", fake_build_aspect_analysis, ) monkeypatch.setattr( aspect_report, "pruned_aspect_bucket_source_paths", lambda _analysis: ("SOURCE/0-FULLBODY/scene.png",), ) monkeypatch.setattr(outpaint_module, "run_safe_border", fake_run_safe_border) result = invoke_cli( CliRunner(), [ "comfy", "outpaint", "-a", "--aspects-config", "ANIMA", "--safe-border", "--square-fraction", "0.5", ], input="", ) assert result.exit_code == 0, result.output assert requests[0].inputs == (source_path.resolve(),) assert requests[0].output_dir == output_root.resolve() assert requests[0].square_fraction == 0.5 @pytest.mark.parametrize( "inference_option", [ ("--workflow", "willy"), ("--url", "http://comfy.test"), ("--seed", "42"), ], ) def test_comfy_outpaint_safe_border_rejects_inference_only_options( isolated_cli_project: Path, inference_option: tuple[str, str], ) -> None: from PIL import Image source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png" source_path.parent.mkdir(parents=True, exist_ok=True) Image.new("RGB", (640, 960), color="white").save(source_path) result = invoke_cli( CliRunner(), [ "comfy", "outpaint", str(source_path), "--safe-border", *inference_option, ], input="", ) assert result.exit_code != 0 normalized_output = " ".join(result.output.replace("│", " ").split()) assert "--safe-border cannot be combined" in normalized_output def test_comfy_outpaint_safe_border_requires_configured_local_output_root( isolated_cli_project: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: from PIL import Image source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png" source_path.parent.mkdir(parents=True, exist_ok=True) Image.new("RGB", (640, 960), color="white").save(source_path) monkeypatch.setenv(COMFY_OUTPUT_DIR_ENV_KEY, "") result = invoke_cli( CliRunner(), ["comfy", "outpaint", str(source_path), "--safe-border"], input="", ) assert result.exit_code != 0 assert f"Set {COMFY_OUTPUT_DIR_ENV_KEY} to use --safe-border" in result.output def test_comfy_outpaint_interrupt_exits_after_requesting_cancellation( isolated_cli_project: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: from PIL import Image import kneiff.infer.comfy.outpaint as outpaint_module source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png" source_path.parent.mkdir(parents=True, exist_ok=True) Image.new("RGB", (640, 960), color="white").save(source_path) def interrupted_run_outpaint(_request: object, **_kwargs: object) -> None: raise KeyboardInterrupt monkeypatch.setattr(outpaint_module, "run_outpaint", interrupted_run_outpaint) result = invoke_cli( CliRunner(), ["comfy", "outpaint", str(source_path), "-w", "willy"], input="", ) assert result.exit_code == 130, result.output assert "requested cancellation of its ComfyUI jobs" in result.stderr def test_comfy_outpaint_from_pruned_aspects_resolves_and_outpaints_sources( isolated_cli_project: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: from PIL import Image import kneiff.datasets.export.aspect_analysis as aspect_analysis import kneiff.infer.comfy.outpaint as outpaint_module import kneiff.training.lora.simpletuner_resolution_report as aspect_report source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png" source_path.parent.mkdir(parents=True, exist_ok=True) Image.new("RGB", (640, 960), color="white").save(source_path) config_path = isolated_cli_project / "configs" / "ANIMA.knf.yaml" config_path.parent.mkdir(parents=True, exist_ok=True) config_path.write_text( 'mappings:\n fullbody:\n - "0-FULLBODY"\n', encoding="utf-8", ) requests: list[outpaint_module.ComfyOutpaintRequest] = [] def fake_build_aspect_analysis( plan: object, *, process_count: int, progress_callback: object, ) -> object: assert getattr(plan, "dataset_name") == "ANIMA" assert process_count == 1 assert callable(progress_callback) return object() def fake_run_outpaint( request: outpaint_module.ComfyOutpaintRequest, **_kwargs: object, ) -> tuple[outpaint_module.ComfyOutpaintResult, ...]: requests.append(request) return () monkeypatch.setattr( aspect_analysis, "build_dataset_aspect_analysis", fake_build_aspect_analysis, ) monkeypatch.setattr( aspect_report, "pruned_aspect_bucket_source_paths", lambda _analysis: ("SOURCE/0-FULLBODY/scene.png",), ) monkeypatch.setattr(outpaint_module, "run_outpaint", fake_run_outpaint) result = invoke_cli( CliRunner(), [ "comfy", "outpaint", "-a", "--aspects-config", "ANIMA", "-w", "willy", "--square-fraction", "0.5", ], input="", ) assert result.exit_code == 0, result.output assert "ANIMA: found 1 physically pruned source image(s)." in result.stderr assert requests[0].inputs == (source_path.resolve(),) assert requests[0].square_fraction == 0.5 def test_comfy_outpaint_from_pruned_aspects_empty_result_is_a_noop( isolated_cli_project: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: import kneiff.datasets.export.aspect_analysis as aspect_analysis import kneiff.training.lora.simpletuner_resolution_report as aspect_report config_path = isolated_cli_project / "configs" / "ANIMA.knf.yaml" config_path.parent.mkdir(parents=True, exist_ok=True) config_path.write_text( 'mappings:\n fullbody:\n - "0-FULLBODY"\n', encoding="utf-8", ) monkeypatch.setattr( aspect_analysis, "build_dataset_aspect_analysis", lambda *_args, **_kwargs: object(), ) monkeypatch.setattr( aspect_report, "pruned_aspect_bucket_source_paths", lambda _analysis: (), ) result = invoke_cli( CliRunner(), ["comfy", "outpaint", "-a"], input="", ) assert result.exit_code == 0, result.output assert "No source images are physically pruned" in result.stderr def test_comfy_outpaint_rejects_pruned_aspects_with_other_input_modes( isolated_cli_project: Path, ) -> None: from PIL import Image source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png" source_path.parent.mkdir(parents=True, exist_ok=True) Image.new("RGB", (640, 960), color="white").save(source_path) config_path = isolated_cli_project / "configs" / "ANIMA.knf.yaml" config_path.parent.mkdir(parents=True, exist_ok=True) config_path.write_text( 'mappings:\n fullbody:\n - "0-FULLBODY"\n', encoding="utf-8", ) result = invoke_cli( CliRunner(), ["comfy", "outpaint", "-a", str(source_path), "-w", "willy"], input="", ) assert result.exit_code != 0 normalized_output = " ".join(result.output.replace("│", " ").split()) assert "cannot be combined with positional image inputs" in normalized_output piped_result = invoke_cli( CliRunner(), ["comfy", "outpaint", "-a", "-w", "willy"], input="SOURCE/0-FULLBODY/scene.png\n", ) assert piped_result.exit_code != 0 normalized_piped_output = " ".join(piped_result.output.replace("│", " ").split()) assert "cannot be combined with positional image inputs" in normalized_piped_output def test_comfy_outpaint_empty_source_stream_is_a_successful_noop( isolated_cli_project: Path, ) -> None: del isolated_cli_project result = invoke_cli( CliRunner(), ["comfy", "outpaint"], input="", ) assert result.exit_code == 0, result.output assert "No outpaint source images received on stdin" in result.stderr def test_comfy_outpaint_rejects_positional_images_and_source_stream( isolated_cli_project: Path, ) -> None: from PIL import Image source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png" source_path.parent.mkdir(parents=True, exist_ok=True) Image.new("RGB", (640, 960), color="white").save(source_path) result = invoke_cli( CliRunner(), ["comfy", "outpaint", str(source_path), "-w", "willy"], input="SOURCE/0-FULLBODY/scene.png\n", ) assert result.exit_code != 0 normalized_output = " ".join(result.output.replace("│", " ").split()) assert ( "either positional image files or piped SOURCE/... paths" in normalized_output ) def test_comfy_upscale_requires_storage( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: monkeypatch.delenv("KNF_STORAGE", raising=False) monkeypatch.delenv(KNF_APPRC_TOML_ENV_KEY, raising=False) monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config-home")) image_path = tmp_path / "image.png" image_path.write_bytes(b"image") runner = CliRunner() result = invoke_cli( runner, [ "comfy", "upscale", str(image_path), "--model", "cli-upscaler.safetensors", "--fast", ], ) assert result.exit_code != 0 assert "KNF_STORAGE is required" in result.output def test_comfy_upscale_quality_uses_brain_off_defaults( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import kneiff.infer.comfy.upscale as upscale_module from kneiff.infer.comfy.upscale import ( ComfyUpscaleRequest, ComfyUpscaleResult, ) for env_key in ( COMFY_UPSCALE_MODEL_ENV_KEY, COMFY_UPSCALE_REFINER_UNET_ENV_KEY, COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY, COMFY_UPSCALE_REFINER_VAE_ENV_KEY, COMFY_UPSCALE_LORA_ENV_KEY, COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, COMFY_UPSCALE_DENOISE_BASE_ENV_KEY, ): monkeypatch.delenv(env_key, raising=False) image_path = tmp_path / "image.png" image_path.write_bytes(b"image") requests: list[ComfyUpscaleRequest] = [] def fake_run_upscale( request: ComfyUpscaleRequest, **kwargs: object, ) -> tuple[ComfyUpscaleResult, ...]: requests.append(request) return () monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale) _install_fake_comfy_model_client(monkeypatch) runner = CliRunner() result = invoke_cli( runner, ["--skip-dotenv-layers", "comfy", "upscale", str(image_path)] ) assert result.exit_code == 0, result.output assert requests[0].upscale_model == DEFAULT_COMFY_UPSCALE_MODEL assert requests[0].quality_workflow == "krea2" assert requests[0].refiner is not None assert requests[0].refiner.unet == DEFAULT_COMFY_UPSCALE_REFINER_UNET assert requests[0].refiner.clip == DEFAULT_COMFY_UPSCALE_REFINER_CLIP def test_comfy_upscale_z_image_uses_legacy_defaults( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import kneiff.infer.comfy.upscale as upscale_module from kneiff.infer.comfy.upscale import ( ComfyUpscaleRequest, ComfyUpscaleResult, ) for env_key in ( COMFY_UPSCALE_MODEL_ENV_KEY, COMFY_UPSCALE_REFINER_UNET_ENV_KEY, COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY, COMFY_UPSCALE_REFINER_VAE_ENV_KEY, COMFY_UPSCALE_LORA_ENV_KEY, COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, COMFY_UPSCALE_DENOISE_BASE_ENV_KEY, ): monkeypatch.delenv(env_key, raising=False) image_path = tmp_path / "image.png" image_path.write_bytes(b"image") requests: list[ComfyUpscaleRequest] = [] def fake_run_upscale( request: ComfyUpscaleRequest, **kwargs: object, ) -> tuple[ComfyUpscaleResult, ...]: requests.append(request) return () monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale) _install_fake_comfy_model_client(monkeypatch) runner = CliRunner() result = invoke_cli( runner, ["--skip-dotenv-layers", "comfy", "upscale", str(image_path), "-z"], ) assert result.exit_code == 0, result.output assert requests[0].quality_workflow == "z-image" assert requests[0].refiner is not None assert requests[0].refiner.unet == LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_UNET assert requests[0].refiner.clip == LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP assert requests[0].refiner.clip_type == ( LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP_TYPE ) assert requests[0].refiner.vae == LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_VAE def test_comfy_upscale_quality_rejects_unavailable_default_non_interactive( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: for env_key in ( COMFY_UPSCALE_MODEL_ENV_KEY, COMFY_UPSCALE_REFINER_UNET_ENV_KEY, COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY, COMFY_UPSCALE_REFINER_VAE_ENV_KEY, COMFY_UPSCALE_LORA_ENV_KEY, COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, COMFY_UPSCALE_DENOISE_BASE_ENV_KEY, ): monkeypatch.delenv(env_key, raising=False) image_path = tmp_path / "image.png" image_path.write_bytes(b"image") options = dict(_DEFAULT_COMFY_INPUT_OPTIONS) options[("UNETLoader", "unet_name")] = ("other-unet.safetensors",) _install_fake_comfy_model_client(monkeypatch, options=options) runner = CliRunner() result = invoke_cli( runner, ["--skip-dotenv-layers", "comfy", "upscale", str(image_path)] ) assert result.exit_code != 0 assert COMFY_UPSCALE_REFINER_UNET_ENV_KEY in result.output assert DEFAULT_COMFY_UPSCALE_REFINER_UNET in result.output assert "--fast" in result.output def test_comfy_upscale_quality_cli_passes_refiner_and_lora( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import kneiff.infer.comfy.upscale as upscale_module from kneiff.infer.comfy.upscale import ( ComfyUpscaleRequest, ComfyUpscaleResult, ) models_dir = tmp_path / "models" lora_path = models_dir / "models" / "loras" / "character.safetensors" lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") env_lora_path = models_dir / "models" / "loras" / "env-character.safetensors" env_lora_path.write_bytes(b"lora") monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) monkeypatch.setenv(COMFY_UPSCALE_MODEL_ENV_KEY, "4x-test.safetensors") monkeypatch.setenv(COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "z-image.safetensors") monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "qwen.safetensors") monkeypatch.setenv(COMFY_UPSCALE_LORA_ENV_KEY, "env-character.safetensors") monkeypatch.setenv(COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, "env_token") monkeypatch.setenv(COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, "0.9") image_path = tmp_path / "image.png" image_path.write_bytes(b"image") requests: list[ComfyUpscaleRequest] = [] def fake_run_upscale( request: ComfyUpscaleRequest, **kwargs: object, ) -> tuple[ComfyUpscaleResult, ...]: requests.append(request) return () monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale) _install_fake_comfy_model_client(monkeypatch) runner = CliRunner() result = invoke_cli( runner, [ "comfy", "upscale", str(image_path), "--style", "flat", "--lora", "character.safetensors", "--token", "char_token", "--lora-strength", "0.4", "--seed", "123", ], ) assert result.exit_code == 0, result.output assert requests[0].fast is False assert requests[0].style == "flat" assert requests[0].refiner is not None assert requests[0].refiner.unet == "z-image.safetensors" assert requests[0].lora is not None assert requests[0].lora.lora_name == "character.safetensors" assert requests[0].lora.activation_token == "char_token" assert requests[0].lora.strength == 0.4 assert requests[0].seed_root == 123 def test_comfy_upscale_quality_uses_configured_lora_path_chain( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import kneiff.infer.comfy.upscale as upscale_module from kneiff.infer.comfy.upscale import ( ComfyUpscaleRequest, ComfyUpscaleResult, ) models_dir = tmp_path / "models" lora_path = models_dir / "models" / "loras" / "5th-ZI-1" / "2400.safetensors" lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) monkeypatch.setenv(COMFY_LORAS_DIR_1_ENV_KEY, "5th-ZI-1") monkeypatch.setenv(COMFY_UPSCALE_LORA_ENV_KEY, "2400.safetensors") monkeypatch.setenv(COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, "env_token") monkeypatch.setenv(COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, "0.8") monkeypatch.setenv(COMFY_UPSCALE_DENOISE_BASE_ENV_KEY, "0.18") monkeypatch.setenv(COMFY_UPSCALE_MODEL_ENV_KEY, "4x-test.safetensors") monkeypatch.setenv(COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "z-image.safetensors") monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "qwen.safetensors") image_path = tmp_path / "image.png" image_path.write_bytes(b"image") requests: list[ComfyUpscaleRequest] = [] def fake_run_upscale( request: ComfyUpscaleRequest, **kwargs: object, ) -> tuple[ComfyUpscaleResult, ...]: requests.append(request) return () monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale) _install_fake_comfy_model_client(monkeypatch) runner = CliRunner() result = invoke_cli(runner, ["comfy", "upscale", str(image_path), "-d", "0.19"]) assert result.exit_code == 0, result.output assert requests[0].lora is not None assert requests[0].lora.lora_name == "5th-ZI-1/2400.safetensors" assert requests[0].lora.activation_token == "env_token" assert requests[0].lora.strength == 0.8 assert requests[0].denoise_base == 0.19 def test_comfy_upscale_quality_allows_configured_lora_without_token( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import kneiff.infer.comfy.upscale as upscale_module from kneiff.infer.comfy.upscale import ( ComfyUpscaleRequest, ComfyUpscaleResult, ) models_dir = tmp_path / "models" lora_path = models_dir / "models" / "loras" / "character.safetensors" lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) monkeypatch.setenv(COMFY_UPSCALE_LORA_ENV_KEY, "character.safetensors") monkeypatch.delenv(COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, raising=False) monkeypatch.setenv(COMFY_UPSCALE_MODEL_ENV_KEY, "4x-test.safetensors") monkeypatch.setenv(COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "z-image.safetensors") monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "qwen.safetensors") image_path = tmp_path / "image.png" image_path.write_bytes(b"image") requests: list[ComfyUpscaleRequest] = [] def fake_run_upscale( request: ComfyUpscaleRequest, **kwargs: object, ) -> tuple[ComfyUpscaleResult, ...]: requests.append(request) return () monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale) _install_fake_comfy_model_client(monkeypatch) runner = CliRunner() result = invoke_cli( runner, ["--skip-dotenv-layers", "comfy", "upscale", str(image_path)] ) assert result.exit_code == 0, result.output assert requests[0].lora is not None assert requests[0].lora.activation_token == "" def test_comfy_upscale_cli_requires_token_with_lora( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: models_dir = tmp_path / "models" lora_path = models_dir / "models" / "loras" / "character.safetensors" lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) monkeypatch.setenv(COMFY_UPSCALE_MODEL_ENV_KEY, "4x-test.safetensors") monkeypatch.setenv(COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "z-image.safetensors") monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "qwen.safetensors") image_path = tmp_path / "image.png" image_path.write_bytes(b"image") _install_fake_comfy_model_client(monkeypatch) runner = CliRunner() result = invoke_cli( runner, [ "comfy", "upscale", str(image_path), "--lora", "character.safetensors", ], ) assert result.exit_code != 0 assert "--token" in result.output def test_comfy_upscale_fast_rejects_explicit_quality_options( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: image_path = tmp_path / "image.png" image_path.write_bytes(b"image") _install_fake_comfy_model_client(monkeypatch) runner = CliRunner() result = invoke_cli( runner, [ "comfy", "upscale", str(image_path), "--fast", "--lora", "character.safetensors", ], ) assert result.exit_code != 0 assert "--fast" in result.output assert "LoRA options" in result.output def test_comfy_upscale_fast_rejects_z_image_workflow( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: image_path = tmp_path / "image.png" image_path.write_bytes(b"image") runner = CliRunner() result = invoke_cli( runner, [ "comfy", "upscale", str(image_path), "--fast", "-z", ], ) assert result.exit_code != 0 assert "--z-image" in result.output assert "--fast" in result.output def test_comfy_upscale_explicit_unavailable_model_is_strict( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: image_path = tmp_path / "image.png" image_path.write_bytes(b"image") _install_fake_comfy_model_client(monkeypatch) monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True) runner = CliRunner() result = invoke_cli( runner, [ "comfy", "upscale", str(image_path), "--model", "missing-upscaler.safetensors", "--fast", ], ) assert result.exit_code != 0 assert "missing-upscaler.safetensors" in result.output assert "Requested upscale model" in result.output assert "Selected upscale model" not in result.output def test_comfy_upscale_interactive_model_picker_is_current_run_only( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import kneiff.infer.comfy.upscale as upscale_module from kneiff.infer.comfy.upscale import ( ComfyUpscaleRequest, ComfyUpscaleResult, ) image_path = tmp_path / "image.png" image_path.write_bytes(b"image") monkeypatch.delenv(COMFY_UPSCALE_MODEL_ENV_KEY, raising=False) requests: list[ComfyUpscaleRequest] = [] options = dict(_DEFAULT_COMFY_INPUT_OPTIONS) options[("UpscaleModelLoader", "model_name")] = ( "first-upscaler.safetensors", "chosen-upscaler.safetensors", ) actions = iter(("down", "select")) def fake_run_upscale( request: ComfyUpscaleRequest, **kwargs: object, ) -> tuple[ComfyUpscaleResult, ...]: requests.append(request) return () monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale) _install_fake_comfy_model_client(monkeypatch, options=options) monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True) monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions)) runner = CliRunner() result = invoke_cli(runner, ["comfy", "upscale", str(image_path), "--fast"]) assert result.exit_code == 0, result.output assert requests[0].upscale_model == "chosen-upscaler.safetensors" assert KneiffConfig().comfy.upscale_model == DEFAULT_COMFY_UPSCALE_MODEL assert "Selected upscale model: chosen-upscaler.safetensors" in result.output def test_comfy_upscale_interactive_lora_token_prompt( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import kneiff.infer.comfy.upscale as upscale_module from kneiff.infer.comfy.upscale import ( ComfyUpscaleRequest, ComfyUpscaleResult, ) models_dir = tmp_path / "models" lora_path = models_dir / "models" / "loras" / "character.safetensors" lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) for env_key in ( COMFY_UPSCALE_MODEL_ENV_KEY, COMFY_UPSCALE_REFINER_UNET_ENV_KEY, COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY, COMFY_UPSCALE_REFINER_VAE_ENV_KEY, COMFY_UPSCALE_LORA_ENV_KEY, COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, COMFY_UPSCALE_DENOISE_BASE_ENV_KEY, ): monkeypatch.delenv(env_key, raising=False) image_path = tmp_path / "image.png" image_path.write_bytes(b"image") requests: list[ComfyUpscaleRequest] = [] def fake_run_upscale( request: ComfyUpscaleRequest, **kwargs: object, ) -> tuple[ComfyUpscaleResult, ...]: requests.append(request) return () monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale) _install_fake_comfy_model_client(monkeypatch) monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True) runner = CliRunner() result = invoke_cli( runner, [ "--skip-dotenv-layers", "comfy", "upscale", str(image_path), "--lora", "character.safetensors", ], input="char_token\n", ) assert result.exit_code == 0, result.output assert requests[0].lora is not None assert requests[0].lora.activation_token == "char_token" def test_comfy_showcase_cli_uses_preset_prompt_field( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import kneiff.infer.comfy.showcase as showcase_module from kneiff.infer.comfy.showcase import ( ShowcaseRunRequest, ShowcaseRunResult, ) prompts_path = tmp_path / "prompts.yaml" prompts_path.write_text( """ version: 1 defaults: seed: 123 prompts: - id: first uses: [showcase] captions: nlg: - prose prompt pony: - pony tags noob: - noob tags """, encoding="utf-8", ) requests: list[ShowcaseRunRequest] = [] def fake_run_showcase( request: ShowcaseRunRequest, **kwargs: object, ) -> ShowcaseRunResult: requests.append(request) return ShowcaseRunResult(results=()) monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase) runner = CliRunner() result = invoke_cli( runner, [ "comfy", "showcase", "-w", "pony", "-p", str(prompts_path), ], ) assert result.exit_code == 0, result.output assert requests[0].workflow_source.name == "showcase_pony.workflow.json" assert requests[0].prompt_field == "pony" assert requests[0].showcase_model == "pony" assert [prompt.id for prompt in requests[0].prompt_catalog.prompts] == ["first"] assert result.output.count("Generating 1 showcase image...") == 1 assert result.output.count("Showcase complete") == 1 result = invoke_cli( runner, [ "comfy", "showcase", "-w", "noob-willy", "-p", str(prompts_path), ], ) assert result.exit_code == 0, result.output assert requests[1].workflow_source.name == "showcase_noob-willy.workflow.json" assert requests[1].prompt_field == "noob" assert requests[1].showcase_model == "noob" def test_comfy_showcase_progress_stops_incomplete_after_generation_failure( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import kneiff.infer.comfy.showcase as showcase_module from kneiff.infer.comfy.showcase import ( ShowcasePromptResult, ShowcaseRunRequest, ) from kneiff.infer.comfy.showcase_runner import select_showcase_prompts prompts_path = tmp_path / "prompts.yaml" prompts_path.write_text( """ version: 1 defaults: seed: 123 prompts: - id: first uses: [showcase] captions: nlg: - prose prompt """, encoding="utf-8", ) reporters: list[_CapturedShowcaseProgress] = [] def fail_run_showcase( request: ShowcaseRunRequest, *, progress: Callable[[ShowcasePromptResult], None] | None = None, ) -> NoReturn: prompts, _skipped = select_showcase_prompts( request.prompt_catalog, request.prompt_field, ) prompt = prompts[0] if progress is not None: progress( ShowcasePromptResult( prompt=prompt, prompt_id="queued-first", filename_prefix="first", files=(), ) ) raise RuntimeError("generation broke") monkeypatch.setattr(showcase_module, "run_showcase", fail_run_showcase) monkeypatch.setattr( cli_comfy, "CliProgress", _capturing_showcase_progress_factory(reporters), ) result = invoke_cli( CliRunner(), ["comfy", "showcase", "--workflow", "anima", "-p", str(prompts_path)], ) assert result.exit_code == 1 assert reporters[0].total == 1 assert reporters[0].advances == ["first"] assert reporters[0].exit_error is RuntimeError assert "Showcase failed: generation broke" in result.output assert "Showcase complete" not in result.output def test_comfy_showcase_cli_infers_willy_noob_lora( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, isolated_cli_project: Path, ) -> None: import kneiff.infer.comfy.showcase as showcase_module from kneiff.infer.comfy.showcase import ( ShowcaseRunRequest, ShowcaseRunResult, ) models_dir = tmp_path / "comfyui-models" lora_path = ( models_dir / "models" / "loras" / "_ladybird" / "Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors" ) lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") _write_noob_showcase_prompts(isolated_cli_project / "prompts.knf.yaml") monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) requests: list[ShowcaseRunRequest] = [] def fake_run_showcase( request: ShowcaseRunRequest, **kwargs: object, ) -> ShowcaseRunResult: requests.append(request) return ShowcaseRunResult(results=()) monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase) runner = CliRunner() result = invoke_cli( runner, [ "comfy", "showcase", "--workflow", "auto", "--lora", "_ladybird/Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors", ], ) assert result.exit_code == 0, result.output assert requests[0].workflow_source.name == "showcase_noob-willy.workflow.json" catalog = requests[0].prompt_catalog prompt_ids = [prompt.id for prompt in catalog.prompts] assert prompt_ids[0] == "character_reference" assert "aurora_portrait" in prompt_ids assert catalog.overlay_prompt_count == 1 assert catalog.resolved_prompt_count == catalog.core_prompt_count + 1 assert requests[0].lora_name == ( "_ladybird/Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors" ) assert requests[0].prompt_field == "noob" assert requests[0].workflow_name == "noob-willy" assert "Prompt sources" in result.output assert "Workflow noob-willy" in result.output assert "Outputs" in result.output assert "Showcase complete" in result.output def test_comfy_showcase_cli_prompts_for_workflow_when_omitted( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import kneiff.infer.comfy.showcase as showcase_module from kneiff.infer.comfy.showcase import ( ShowcaseRunRequest, ShowcaseRunResult, ) models_dir = tmp_path / "comfyui-models" lora_path = ( models_dir / "models" / "loras" / "_ladybird" / "Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors" ) lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") prompts_path = tmp_path / "noob-prompts.yaml" _write_noob_showcase_prompts(prompts_path) monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True) actions = iter(("down", "select")) monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions)) requests: list[ShowcaseRunRequest] = [] def fake_run_showcase( request: ShowcaseRunRequest, **kwargs: object, ) -> ShowcaseRunResult: requests.append(request) return ShowcaseRunResult(results=()) monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase) runner = CliRunner() result = invoke_cli( runner, [ "comfy", "showcase", "--lora", "_ladybird/Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors", "--prompts", str(prompts_path), ], ) assert result.exit_code == 0, result.output assert requests[0].workflow_source.name == "showcase_noob-willy.workflow.json" assert requests[0].workflow_name == "noob-willy" assert requests[0].prompt_field == "noob" def test_comfy_showcase_cli_prompts_for_generic_noob_workflow( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import kneiff.infer.comfy.showcase as showcase_module from kneiff.infer.comfy.showcase import ( ShowcaseRunRequest, ShowcaseRunResult, ) models_dir = tmp_path / "comfyui-models" lora_path = ( models_dir / "models" / "loras" / "_ladybird" / "Rook_Kaefer-v1_0-NOOB.comfyui.safetensors" ) lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") prompts_path = tmp_path / "noob-prompts.yaml" _write_noob_showcase_prompts(prompts_path) monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True) actions = iter(("down", "select")) monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions)) requests: list[ShowcaseRunRequest] = [] def fake_run_showcase( request: ShowcaseRunRequest, **kwargs: object, ) -> ShowcaseRunResult: requests.append(request) return ShowcaseRunResult(results=()) monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase) runner = CliRunner() result = invoke_cli( runner, [ "comfy", "showcase", "--workflow", "auto", "--lora", "_ladybird/Rook_Kaefer-v1_0-NOOB.comfyui.safetensors", "--prompts", str(prompts_path), ], ) assert result.exit_code == 0, result.output assert requests[0].workflow_source.name == "showcase_noob-chenkin.workflow.json" assert requests[0].workflow_name == "noob-chenkin" assert "Selected workflow: noob-chenkin" in result.output def test_comfy_showcase_cli_rejects_generic_noob_auto_non_interactive( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: models_dir = tmp_path / "comfyui-models" lora_path = ( models_dir / "models" / "loras" / "_ladybird" / "Rook_Kaefer-v1_0-NOOB.comfyui.safetensors" ) lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) runner = CliRunner() result = invoke_cli( runner, [ "comfy", "showcase", "--workflow", "auto", "--lora", "_ladybird/Rook_Kaefer-v1_0-NOOB.comfyui.safetensors", "--prompts", str(SHOWCASE_PROMPTS_PATH), ], ) assert result.exit_code != 0 assert "LoRA matches multiple workflows" in result.output assert "noob-willy" in result.output assert "noob-chenkin" in result.output assert "noob-base" in result.output assert "noob-nova" in result.output assert "noob-scrimblosauce" in result.output assert "pass --workflow" in result.output def test_comfy_showcase_cli_prompts_when_packaged_workflow_is_missing( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import kneiff.infer.comfy.showcase as showcase_module from kneiff.infer.comfy.showcase import ( ShowcaseRunRequest, ShowcaseRunResult, ShowcaseWorkflowPreset, ) import kneiff.infer.comfy.workflow_presets as workflow_presets prompts_path = tmp_path / "prompts.yaml" prompts_path.write_text( """ version: 1 defaults: seed: 123 prompts: - id: first uses: [showcase] captions: nlg: - prose prompt """, encoding="utf-8", ) models_dir = tmp_path / "comfyui-models" lora_path = models_dir / "models" / "loras" / "_ladybird" / "rook.safetensors" lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) monkeypatch.setitem( workflow_presets.SHOWCASE_WORKFLOW_PRESETS, "pony", ShowcaseWorkflowPreset( name="pony", workflow_filename="missing_pony.workflow.json", prompt_field="pony", ), ) monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True) monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: "select") requests: list[ShowcaseRunRequest] = [] def fake_run_showcase( request: ShowcaseRunRequest, **kwargs: object, ) -> ShowcaseRunResult: requests.append(request) return ShowcaseRunResult(results=()) monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase) runner = CliRunner() result = invoke_cli( runner, [ "comfy", "showcase", "--workflow", "pony", "--lora", "_ladybird/rook.safetensors", "--prompts", str(prompts_path), ], ) assert result.exit_code == 0, result.output assert requests[0].workflow_source.name == "showcase_anima.workflow.json" assert requests[0].workflow_name == "anima" assert "Selected workflow: anima" in result.output def test_comfy_showcase_lora_picker_descends_into_directories( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: models_dir = tmp_path / "comfyui-models" lora_path = models_dir / "models" / "loras" / "nested" / "rook.safetensors" lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") actions = iter(("select", "down", "select")) output = io.StringIO() monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions)) selected = cli_comfy.select_showcase_lora_interactively( models_dir, console=Console(file=output, force_terminal=False), ) assert selected == "nested/rook.safetensors" assert "nested/" in output.getvalue() def test_comfy_showcase_lora_picker_starts_in_configured_directory( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: models_dir = tmp_path / "comfyui-models" start_dir = models_dir / "models" / "loras" / "_ladybird" / "v1_0" lora_path = start_dir / "rook.safetensors" lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") output = io.StringIO() monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: "select") selected = cli_comfy.select_showcase_lora_interactively( models_dir, console=Console(file=output, force_terminal=False), start_dir=start_dir, ) assert selected == "_ladybird/v1_0/rook.safetensors" assert "Directory: _ladybird/v1_0" in output.getvalue() def test_comfy_showcase_lora_picker_start_dir_env_rejects_missing_directory( tmp_path: Path, ) -> None: models_dir = tmp_path / "comfyui-models" missing_dir = models_dir / "models" / "loras" / "_ladybird" / "v1_0" config = ComfyConfig( models_dir=str(models_dir), loras_dir_1=str(missing_dir), ) with pytest.raises(ValueError, match="COMFY_LORAS_DIR_1 directory not found"): cli_comfy._resolve_comfy_loras_start_dir(models_dir, config=config) def test_comfy_showcase_lora_picker_start_dir_env_rejects_outside_lora_root( tmp_path: Path, ) -> None: models_dir = tmp_path / "comfyui-models" outside_dir = tmp_path / "other-loras" outside_dir.mkdir() config = ComfyConfig( models_dir=str(models_dir), loras_dir_1=str(outside_dir), ) with pytest.raises(ValueError, match="COMFY_LORAS_DIR_1 must be below"): cli_comfy._resolve_comfy_loras_start_dir(models_dir, config=config) def test_comfy_showcase_training_picker_uses_step_and_cleans_staged_lora( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, isolated_cli_project: Path, ) -> None: import kneiff.infer.comfy.showcase as showcase_module from kneiff.infer.comfy.showcase import ShowcaseRunRequest, ShowcaseRunResult output_root = ( isolated_cli_project / "TRAINING" / "ANIMA_1" / "_simpletuner-output" / "checkpoint-100" ) output_root.mkdir(parents=True) (output_root / "pytorch_lora_weights.safetensors").write_bytes(b"raw") preferred_path = output_root / "pytorch_lora_weights.comfyui.safetensors" preferred_path.write_bytes(b"comfyui") second_step = output_root.parent / "checkpoint-200" / "weights.safetensors" second_step.parent.mkdir() second_step.write_bytes(b"second-step") models_dir = tmp_path / "comfyui-models" (models_dir / "models" / "loras").mkdir(parents=True) monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True) actions = iter(("select", "down", "select")) monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions)) requests: list[ShowcaseRunRequest] = [] staged_paths: list[Path] = [] def fake_run_showcase( request: ShowcaseRunRequest, **kwargs: object, ) -> ShowcaseRunResult: requests.append(request) assert request.lora_name is not None staged_path = models_dir / "models" / "loras" / request.lora_name assert staged_path.read_bytes() == b"comfyui" staged_paths.append(staged_path) return ShowcaseRunResult(results=()) monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase) result = invoke_cli( CliRunner(), [ "comfy", "showcase", "-t", "100", "200", "100", "--workflow", "anima", ], ) assert result.exit_code == 0, result.output assert len(requests) == 1 assert requests[0].lora_name is not None assert requests[0].lora_name.endswith( "ANIMA_1/_simpletuner-output/checkpoint-100/" "pytorch_lora_weights.comfyui.safetensors" ) assert "Selected TRAINING LoRA:" in result.output assert not staged_paths[0].exists() assert not (models_dir / "models" / "loras" / ".kneiff-training").exists() def test_comfy_showcase_training_hides_root_exports_without_checkpoint_steps( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, isolated_cli_project: Path, ) -> None: import kneiff.infer.comfy.showcase as showcase_module from kneiff.infer.comfy.showcase import ShowcaseRunRequest, ShowcaseRunResult output_root = isolated_cli_project / "TRAINING" / "ANIMA_1" / "_simpletuner-output" output_root.mkdir(parents=True) root_lora = output_root / "root.comfyui.safetensors" root_lora.write_bytes(b"root") models_dir = tmp_path / "comfyui-models" (models_dir / "models" / "loras").mkdir(parents=True) monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True) requests: list[ShowcaseRunRequest] = [] def fake_run_showcase( request: ShowcaseRunRequest, **kwargs: object, ) -> ShowcaseRunResult: requests.append(request) return ShowcaseRunResult(results=()) monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase) result = invoke_cli( CliRunner(), ["comfy", "showcase", "-t", "--workflow", "anima"], ) assert result.exit_code != 0 assert "No checkpoint- LoRA outputs" in result.output assert not requests def test_comfy_showcase_training_runs_every_selected_lora_in_one_grid( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, isolated_cli_project: Path, ) -> None: import kneiff.infer.comfy.showcase as showcase_module import kneiff.infer.comfy.showcase_grid as showcase_grid_module from kneiff.infer.comfy.showcase import ( ShowcasePromptResult, ShowcaseRunRequest, ShowcaseRunResult, ) from kneiff.infer.comfy.showcase_grid import ShowcaseGridRow from kneiff.infer.comfy.showcase_runner import select_showcase_prompts output_root = isolated_cli_project / "TRAINING" / "ANIMA_1" / "_simpletuner-output" for step in (100, 200): lora_path = ( output_root / f"checkpoint-{step}" / "pytorch_lora_weights.comfyui.safetensors" ) lora_path.parent.mkdir(parents=True) lora_path.write_bytes(str(step).encode("ascii")) models_dir = tmp_path / "comfyui-models" (models_dir / "models" / "loras").mkdir(parents=True) monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True) actions = iter(("select", "down", "all", "select")) monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions)) requests: list[ShowcaseRunRequest] = [] staged_paths: list[Path] = [] captured_grid_rows: tuple[ShowcaseGridRow, ...] = () progress_reporters: list[_CapturedShowcaseProgress] = [] def fake_run_showcase( request: ShowcaseRunRequest, *, progress: Callable[[ShowcasePromptResult], None] | None = None, ) -> ShowcaseRunResult: requests.append(request) assert request.lora_name is not None staged_path = models_dir / "models" / "loras" / request.lora_name assert staged_path.is_file() staged_paths.append(staged_path) prompts, skipped_prompt_ids = select_showcase_prompts( request.prompt_catalog, request.prompt_field, ) results = tuple( ShowcasePromptResult( prompt=prompt, prompt_id=f"queued-{prompt.id}", filename_prefix=prompt.id, files=(), ) for prompt in prompts ) if progress is not None: for prompt_result in results: progress(prompt_result) return ShowcaseRunResult( results=results, skipped_prompt_ids=skipped_prompt_ids, ) def fake_write_showcase_grid( rows: tuple[ShowcaseGridRow, ...] | list[ShowcaseGridRow], *, output_subfolder: str, filename: str, **kwargs: object, ) -> ComfyUploadedImage: nonlocal captured_grid_rows captured_grid_rows = tuple(rows) assert all(path.is_file() for path in staged_paths) return ComfyUploadedImage( name=filename, subfolder=output_subfolder, type="output", ) monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase) monkeypatch.setattr( cli_comfy, "CliProgress", _capturing_showcase_progress_factory(progress_reporters), ) monkeypatch.setattr( showcase_grid_module, "write_showcase_grid", fake_write_showcase_grid, ) result = invoke_cli( CliRunner(), ["comfy", "showcase", "-t", "100", "200", "--workflow", "anima"], ) assert result.exit_code == 0, result.output assert len(requests) == 2 assert requests[0].generated_at == requests[1].generated_at assert all(request.showcase_model == "anima" for request in requests) assert [row.label for row in captured_grid_rows] == [ "ANIMA_1/_simpletuner-output/checkpoint-100/" "pytorch_lora_weights.comfyui.safetensors", "ANIMA_1/_simpletuner-output/checkpoint-200/" "pytorch_lora_weights.comfyui.safetensors", ] assert all(row.showcase_model == "anima" for row in captured_grid_rows) assert all(not path.exists() for path in staged_paths) assert not tuple( isolated_cli_project.joinpath("TRAINING").glob("*showcase-grid*.jpg") ) assert len(progress_reporters) == 1 assert progress_reporters[0].title == "Showcase" assert progress_reporters[0].total == sum( len(row.result.results) for row in captured_grid_rows ) assert progress_reporters[0].advances first_progress_text = progress_reporters[0].advances[0] final_progress_text = progress_reporters[0].advances[-1] assert first_progress_text is not None assert final_progress_text is not None assert first_progress_text.startswith("LoRA 1/2 · ") assert final_progress_text.startswith("LoRA 2/2 · ") assert "Selected TRAINING LoRAs:" in result.output assert "kneiff-showcase-grid.jpg" in result.output assert "LoRA: " not in result.output @pytest.mark.parametrize( ("args", "expected"), [ (["comfy", "showcase", "100"], "STEPS require --training"), ( ["comfy", "showcase", "-t", "--lora", "character.safetensors"], "Do not combine --training", ), ], ) def test_comfy_showcase_training_rejects_invalid_option_combinations( args: list[str], expected: str, ) -> None: result = invoke_cli(CliRunner(), args) assert result.exit_code != 0 assert expected in result.output def test_comfy_showcase_training_requires_interactive_terminal() -> None: result = invoke_cli(CliRunner(), ["comfy", "showcase", "-t"]) assert result.exit_code != 0 assert "requires an interactive" in result.output assert "terminal for LoRA selection" in result.output @pytest.mark.parametrize( ("step", "expected"), [ ("0", "positive"), ("200", "Available steps: 100"), ], ) def test_comfy_showcase_training_rejects_invalid_or_missing_steps( monkeypatch: pytest.MonkeyPatch, isolated_cli_project: Path, step: str, expected: str, ) -> None: checkpoint_root = ( isolated_cli_project / "TRAINING" / "ANIMA_1" / "_simpletuner-output" / "checkpoint-100" ) checkpoint_root.mkdir(parents=True) (checkpoint_root / "weights.safetensors").write_bytes(b"lora") monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True) result = invoke_cli(CliRunner(), ["comfy", "showcase", "-t", step]) assert result.exit_code != 0 assert expected in result.output def test_comfy_showcase_training_picker_opens_runs_before_checkpoints( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: from kneiff.training.lora.output_discovery import TrainingLoraCandidate first_path = tmp_path / "first.comfyui.safetensors" second_path = tmp_path / "second.safetensors" first_path.write_bytes(b"first") second_path.write_bytes(b"second") candidates = ( TrainingLoraCandidate( path=first_path, relative_path=Path( "first_1/_simpletuner-output/checkpoint-50/first.comfyui.safetensors" ), checkpoint_step=50, comfyui_native=True, ), TrainingLoraCandidate( path=second_path, relative_path=Path( "second_1/_simpletuner-output/checkpoint-100/second.safetensors" ), checkpoint_step=100, comfyui_native=False, ), ) actions = iter(("down", "select", "down", "select")) output = io.StringIO() monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions)) selected = cli_comfy.select_training_lora_interactively( candidates, console=Console(file=output, force_terminal=False), ) assert selected == (candidates[1],) assert "Select TRAINING run" in output.getvalue() assert "first_1/ (1 LoRA(s))" in output.getvalue() assert candidates[1].relative_path.as_posix() in output.getvalue() def test_comfy_showcase_training_picker_can_be_cancelled( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: from kneiff.training.lora.output_discovery import TrainingLoraCandidate lora_path = tmp_path / "weights.safetensors" lora_path.write_bytes(b"lora") candidate = TrainingLoraCandidate( path=lora_path, relative_path=Path( "ready_1/_simpletuner-output/checkpoint-100/weights.safetensors" ), checkpoint_step=100, comfyui_native=False, ) monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: "cancel") selected = cli_comfy.select_training_lora_interactively( (candidate,), console=Console(file=io.StringIO(), force_terminal=False), ) assert selected is None def test_comfy_showcase_training_picker_selects_multiple_loras( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: from kneiff.training.lora.output_discovery import TrainingLoraCandidate candidates = tuple( TrainingLoraCandidate( path=tmp_path / f"weights-{step}.safetensors", relative_path=Path( f"ready_1/_simpletuner-output/checkpoint-{step}/weights.safetensors" ), checkpoint_step=step, comfyui_native=False, ) for step in (100, 200) ) for candidate in candidates: candidate.path.write_bytes(b"lora") actions = iter(("select", "down", "toggle", "down", "toggle", "select")) output = io.StringIO() monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions)) selected = cli_comfy.select_training_lora_interactively( candidates, console=Console(file=output, force_terminal=False), ) assert selected == candidates assert "Selected TRAINING LoRAs:" in output.getvalue() assert "[x]" in output.getvalue() def test_comfy_showcase_training_infers_workflow_from_relative_path( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, isolated_cli_project: Path, ) -> None: import kneiff.infer.comfy.showcase as showcase_module from kneiff.infer.comfy.showcase import ShowcaseRunRequest, ShowcaseRunResult lora_path = ( isolated_cli_project / "TRAINING" / "ANIMA_1" / "_simpletuner-output" / "checkpoint-100" / "weights.comfyui.safetensors" ) lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") models_dir = tmp_path / "comfyui-models" (models_dir / "models" / "loras").mkdir(parents=True) monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True) actions = iter(("select", "down", "select")) monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions)) requests: list[ShowcaseRunRequest] = [] def fake_run_showcase( request: ShowcaseRunRequest, **kwargs: object, ) -> ShowcaseRunResult: requests.append(request) return ShowcaseRunResult(results=()) monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase) result = invoke_cli(CliRunner(), ["comfy", "showcase", "-t", "100"]) assert result.exit_code == 0, result.output assert requests[0].workflow_name == "anima" def test_comfy_showcase_training_falls_back_to_complete_workflow_picker( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, isolated_cli_project: Path, ) -> None: import kneiff.infer.comfy.showcase as showcase_module from kneiff.infer.comfy.showcase import ShowcaseRunRequest, ShowcaseRunResult lora_path = ( isolated_cli_project / "TRAINING" / "mystery_1" / "_simpletuner-output" / "checkpoint-100" / "weights.safetensors" ) lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") models_dir = tmp_path / "comfyui-models" (models_dir / "models" / "loras").mkdir(parents=True) monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True) actions = iter(("select", "down", "select", "select")) monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions)) requests: list[ShowcaseRunRequest] = [] def fake_run_showcase( request: ShowcaseRunRequest, **kwargs: object, ) -> ShowcaseRunResult: requests.append(request) return ShowcaseRunResult(results=()) monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase) result = invoke_cli(CliRunner(), ["comfy", "showcase", "-t", "100"]) assert result.exit_code == 0, result.output assert requests[0].workflow_name == "anima" assert "Selected workflow: anima" in result.output def test_comfy_showcase_training_explicit_auto_keeps_inference_failure( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, isolated_cli_project: Path, ) -> None: lora_path = ( isolated_cli_project / "TRAINING" / "mystery_1" / "_simpletuner-output" / "checkpoint-100" / "weights.safetensors" ) lora_path.parent.mkdir(parents=True) lora_path.write_bytes(b"lora") models_dir = tmp_path / "comfyui-models" lora_root = models_dir / "models" / "loras" lora_root.mkdir(parents=True) monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir)) monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True) actions = iter(("select", "down", "select")) monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions)) result = invoke_cli( CliRunner(), ["comfy", "showcase", "-t", "100", "--workflow", "auto"], ) assert result.exit_code != 0 assert "Could not infer a workflow" in result.output assert not (lora_root / ".kneiff-training").exists() def test_train_help_skips_storage_bootstrap(monkeypatch: pytest.MonkeyPatch) -> None: runner = CliRunner() monkeypatch.delenv("KNF_STORAGE", raising=False) help_result = invoke_cli(runner, ["train", "--help"]) train_result = invoke_cli(runner, ["train"]) assert help_result.exit_code == 0, help_result.output assert train_result.exit_code == 2 assert "Training run lifecycle workflows." in train_result.output assert "KNF_STORAGE" not in help_result.output assert "KNF_STORAGE" not in train_result.output def test_config_doctor_output_uses_only_knf_env_names() -> None: runner = CliRunner() forbidden = ("KNF_STORAGE_ROOT", "KNEIFF_APPRC_TOML", "KNEIFF_CONFIG_FILE") result = invoke_cli(runner, ["config", "doctor"]) assert "KNF_STORAGE" in result.output assert "KNF_APPRC_TOML" in result.output assert not any(name in result.output for name in forbidden) def test_cli_app_import_keeps_runtime_backends_lazy() -> None: script = ( "import sys\n" "import kneiff.cli.app\n" "heavy_modules = {\n" " 'huggingface_hub', 'numpy', 'openai', 'openpyxl', 'pandas',\n" " 'requests', 'safetensors', 'spandrel', 'torch', 'torchvision',\n" " 'transformers',\n" "}\n" "loaded = sorted(name for name in heavy_modules if name in sys.modules)\n" "print('\\n'.join(loaded))\n" "raise SystemExit(1 if loaded else 0)\n" ) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, check=False, ) assert result.returncode == 0, result.stdout + result.stderr def test_lora_command_group_is_removed() -> None: runner = CliRunner() result = invoke_cli(runner, ["lora", "--help"]) assert result.exit_code != 0 def test_old_diff_subcommand_is_removed() -> None: runner = CliRunner() result = invoke_cli(runner, ["diff", "--help"]) assert result.exit_code != 0 def test_img_tag_help_shows_selected_tag_output_options( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.delenv("KNF_STORAGE", raising=False) runner = CliRunner() result = invoke_cli(runner, ["img", "tag", "--help"]) assert result.exit_code == 0 assert "--txt" in result.output assert "--comma" in result.output assert "-c" in result.output assert "Write .txt sidecars next to images" in result.output assert "probability CSV output" in result.output assert "Legacy JTP-3 snapshot threshold" in result.output def test_img_tag_cli_passes_text_output_options( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.delenv("KNF_STORAGE", raising=False) runner = CliRunner() image_path = tmp_path / "image.png" image_path.write_bytes(b"image") captured: dict[str, tag_jtp3.Jtp3RunConfig] = {} def fake_run_jtp3( config: tag_jtp3.Jtp3RunConfig, snapshot_config: tag_jtp3.Jtp3SnapshotConfig | None = None, ) -> subprocess.CompletedProcess[str]: captured["config"] = config return subprocess.CompletedProcess([], 0) monkeypatch.setattr(tag_jtp3, "run_jtp3", fake_run_jtp3) result = invoke_cli(runner, ["img", "tag", str(image_path), "--txt", "--comma"]) assert result.exit_code == 0 assert captured["config"].write_txt is True assert captured["config"].comma_separated is True result = invoke_cli(runner, ["img", "tag", str(image_path), "-c"]) assert result.exit_code == 0 assert captured["config"].write_txt is False assert captured["config"].comma_separated is True def test_img_tag_rejects_csv_stdout_with_txt( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.delenv("KNF_STORAGE", raising=False) runner = CliRunner() image_path = tmp_path / "image.png" image_path.write_bytes(b"image") result = invoke_cli( runner, ["img", "tag", str(image_path), "--csv-stdout", "--txt"] ) assert result.exit_code != 0 assert "--csv-stdout cannot be combined with --txt or --comma" in result.output def test_package_import_smoke() -> None: for module_name in ( "kneiff", "kneiff.clients", "kneiff.datasets", "kneiff.datasets.manifest", "kneiff.datasets.export", "kneiff.training", "kneiff.training.lora", "kneiff.infer", "kneiff.cli", "kneiff.cli.app", "kneiff.main", "kneiff.app.workbench", "kneiff.utils", "kneiff.utils.image.caption", "kneiff_dev", ): assert importlib.import_module(module_name) def test_removed_import_paths_are_not_shimmed() -> None: for module_name in ( "kneiff.training.captions", "kneiff.training.dataset", "kneiff.training.manifest", "kneiff.training.augmentations", "kneiff.inference", "kneiff.cli_dataset", "kneiff.cli_dataset_report", "kneiff.cli_image", "kneiff.cli_lora", "kneiff.datasets.manifest_to_captions", "kneiff.datasets.captions", "kneiff.datasets.augment", "kneiff.datasets.manifest.core", ): with pytest.raises(ModuleNotFoundError): importlib.import_module(module_name)