Spaces:
Runtime error
Runtime error
| 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 | |
| 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_<preset>.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 | |
| 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-<step> 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 | |
| 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 | |
| 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) | |