kneifftools / tests /test_cli_smoke.py
kneiff's picture
feat(manifest)!: migrate captioning to Kneifftags
09df1fe
Raw
History Blame Contribute Delete
105 kB
from __future__ import annotations
from collections.abc import Callable
import io
import importlib
import os
import subprocess
import sys
from pathlib import Path
from types import SimpleNamespace
from typing import NoReturn, cast
import apprc as rc
import pytest
from rich.console import Console
from typer.testing import CliRunner
import kneiff.cli.comfy as cli_comfy
import kneiff.cli.dataset as cli_dataset
from kneiff.config import (
COMFY_I2I_LORA_1_ENV_KEY,
COMFY_I2I_LORA_2_ENV_KEY,
COMFY_I2I_LORA_ENV_KEY,
COMFY_I2I_LORA_STRENGTH_1_ENV_KEY,
COMFY_I2I_LORA_STRENGTH_2_ENV_KEY,
COMFY_I2I_LORA_STRENGTH_ENV_KEY,
COMFY_I2I_MODEL_DUO_ENV_KEY,
COMFY_I2I_MODEL_SOLO_ENV_KEY,
COMFY_LORAS_DIR_1_ENV_KEY,
COMFY_LORAS_DIR_2_ENV_KEY,
COMFY_MODELS_DIR_ENV_KEY,
COMFY_OUTPUT_DIR_ENV_KEY,
COMFY_POLL_INTERVAL_SECONDS_ENV_KEY,
COMFY_PROMPT_TIMEOUT_SECONDS_ENV_KEY,
COMFY_UPSCALE_MODEL_ENV_KEY,
COMFY_UPSCALE_DENOISE_BASE_ENV_KEY,
COMFY_UPSCALE_LORA_ENV_KEY,
COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY,
COMFY_UPSCALE_LORA_TOKEN_ENV_KEY,
COMFY_UPSCALE_REFINER_CLIP_ENV_KEY,
COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY,
COMFY_UPSCALE_REFINER_UNET_ENV_KEY,
COMFY_UPSCALE_REFINER_VAE_ENV_KEY,
COMFY_URL_ENV_KEY,
COMFY_T2I_LORA_DUO_1_ENV_KEY,
COMFY_T2I_LORA_DUO_2_ENV_KEY,
COMFY_T2I_LORA_SOLO_ENV_KEY,
COMFY_T2I_LORA_STRENGTH_DUO_1_ENV_KEY,
COMFY_T2I_LORA_STRENGTH_DUO_2_ENV_KEY,
COMFY_T2I_LORA_STRENGTH_SOLO_ENV_KEY,
COMFY_T2I_MODEL_DUO_ENV_KEY,
COMFY_T2I_MODEL_SOLO_ENV_KEY,
DEFAULT_COMFY_UPSCALE_MODEL,
DEFAULT_COMFY_UPSCALE_DENOISE_BASE,
DEFAULT_COMFY_UPSCALE_LORA,
DEFAULT_COMFY_UPSCALE_LORA_STRENGTH,
DEFAULT_COMFY_UPSCALE_LORA_TOKEN,
DEFAULT_COMFY_UPSCALE_REFINER_CLIP,
DEFAULT_COMFY_UPSCALE_REFINER_CLIP_TYPE,
DEFAULT_COMFY_UPSCALE_REFINER_UNET,
DEFAULT_COMFY_UPSCALE_REFINER_VAE,
KNF_APPRC_TOML_ENV_KEY,
KNEIFF_RC,
ComfyConfig,
KneiffConfig,
LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP,
LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP_TYPE,
LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_UNET,
LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_VAE,
)
from kneiff.infer.comfy.client import ComfyUploadedImage
from kneiff.progress import ProgressUpdate
import kneiff.utils.image.tag_jtp3 as tag_jtp3
from tests._cli_helpers import invoke_cli
pytestmark = pytest.mark.usefixtures("isolated_cli_project")
SHOWCASE_PROMPTS_PATH = (
Path(__file__).parent / "fixtures" / "projects" / "aurora" / "prompts.knf.yaml"
)
def _write_noob_showcase_prompts(path: Path) -> None:
"""Write one standalone catalog usable by NOOB CLI routing tests."""
path.write_text(
"""
version: 1
defaults:
seed: 123
prompts:
- id: aurora_portrait
uses: [showcase, training_validation]
captions:
nlg:
- Aurora_Fox portrait
pony:
- score_9, Aurora_Fox, portrait
noob:
- Aurora_Fox, portrait
""",
encoding="utf-8",
)
_DEFAULT_COMFY_INPUT_OPTIONS = {
("UpscaleModelLoader", "model_name"): (
DEFAULT_COMFY_UPSCALE_MODEL,
"local-upscaler.safetensors",
"cli-upscaler.safetensors",
"4x-test.safetensors",
),
("UNETLoader", "unet_name"): (
DEFAULT_COMFY_UPSCALE_REFINER_UNET,
LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_UNET,
"z-image.safetensors",
),
("CLIPLoader", "clip_name"): (
DEFAULT_COMFY_UPSCALE_REFINER_CLIP,
LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP,
"qwen.safetensors",
),
("CLIPLoader", "type"): (
DEFAULT_COMFY_UPSCALE_REFINER_CLIP_TYPE,
LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP_TYPE,
),
("VAELoader", "vae_name"): (
DEFAULT_COMFY_UPSCALE_REFINER_VAE,
LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_VAE,
),
}
class _FakeComfyModelClient:
def __init__(
self,
server_url: str,
options: dict[tuple[str, str], tuple[str, ...]],
) -> None:
self.server_url = server_url.rstrip("/")
self.options = options
self.input_option_requests: list[tuple[str, str]] = []
def input_options(self, node_class: str, input_name: str) -> tuple[str, ...]:
key = (node_class, input_name)
self.input_option_requests.append(key)
return self.options.get(key, ())
class _CapturedShowcaseProgress:
"""Record aggregate CLI progress without rendering a terminal bar."""
def __init__(
self,
title: str,
total: int,
console: Console | None,
) -> None:
self.title = title
self.total = total
self.console = console
self.advances: list[str | None] = []
self.exit_error: type[BaseException] | None = None
def __enter__(self) -> _CapturedShowcaseProgress:
return self
def __exit__(self, *exc_info: object) -> None:
error_type = exc_info[0]
self.exit_error = (
error_type
if isinstance(error_type, type) and issubclass(error_type, BaseException)
else None
)
def advance(self, text: str | None = None, *, amount: int = 1) -> None:
self.advances.extend([text] * amount)
def step(self, text: str, *, amount: int = 1) -> None:
self.total += amount
self.advance(text, amount=amount)
def apply(self, update: ProgressUpdate) -> None:
"""Record one shared progress update."""
self.total += update.additional_total
self.advance(update.description, amount=update.advance)
def _capturing_showcase_progress_factory(
reporters: list[_CapturedShowcaseProgress],
) -> Callable[..., _CapturedShowcaseProgress]:
def create(
title: str,
total: int = 0,
console: Console | None = None,
) -> _CapturedShowcaseProgress:
reporter = _CapturedShowcaseProgress(title, total, console)
reporters.append(reporter)
return reporter
return create
def _install_fake_comfy_model_client(
monkeypatch: pytest.MonkeyPatch,
*,
options: dict[tuple[str, str], tuple[str, ...]] | None = None,
) -> _FakeComfyModelClient:
client = _FakeComfyModelClient(
"http://127.0.0.1:8188",
options or dict(_DEFAULT_COMFY_INPUT_OPTIONS),
)
monkeypatch.setattr(cli_comfy, "ComfyUiClient", lambda server_url: client)
return client
@pytest.fixture(autouse=True)
def _isolate_comfy_env(monkeypatch: pytest.MonkeyPatch) -> None:
import kneiff.infer.comfy.showcase_grid as showcase_grid_module
owner = rc.schema.owner_for(ComfyConfig)
for field in owner.fields:
monkeypatch.delenv(owner.env_key(field.name), raising=False)
monkeypatch.setattr(
showcase_grid_module,
"write_showcase_grid",
lambda rows, output_subfolder, filename, **kwargs: ComfyUploadedImage(
name=filename,
subfolder=output_subfolder,
type="output",
),
)
def test_cli_help_smoke() -> None:
runner = CliRunner()
for args in (
["--help"],
["comfy", "--help"],
["comfy", "showcase", "--help"],
["comfy", "upscale", "--help"],
["comfy", "outpaint", "--help"],
["comfy", "t2i", "--help"],
["comfy", "t2i", "solo", "--help"],
["comfy", "t2i", "duo", "--help"],
["config", "--help"],
["config", "paths", "--help"],
["config", "show", "--help"],
["config", "doctor", "--help"],
["config", "setup", "--help"],
["config", "set", "--help"],
["config", "edit", "--help"],
["config", "app", "--help"],
["config", "app", "init", "--help"],
["config", "storage", "--help"],
["config", "storage", "add", "--help"],
["config", "storage", "list", "--help"],
["config", "storage", "remove", "--help"],
["img", "--help"],
["img", "tag", "--help"],
["llm", "--help"],
["llm", "prompt", "--help"],
["dataset", "--help"],
["dataset", "aspects", "--help"],
["dataset", "describe", "--help"],
["model", "--help"],
["model", "convert", "--help"],
["train", "--help"],
["train", "start", "--help"],
["train", "prepare", "--help"],
["train", "runs", "--help"],
["train", "review", "--help"],
["train", "grid", "--help"],
):
result = invoke_cli(runner, args)
assert result.exit_code == 0
def test_comfy_t2i_help_exposes_preset_specific_options() -> None:
runner = CliRunner()
solo = invoke_cli(runner, ["comfy", "t2i", "solo", "--help"])
duo = invoke_cli(runner, ["comfy", "t2i", "duo", "--help"])
assert solo.exit_code == 0
assert "--pipeline" in solo.output
assert "--activation-token-1" in solo.output
assert "--t2i-lora-strength-1" in solo.output
assert "--num-cleanups" in solo.output
assert "--i2i-model" in solo.output
assert "--i2i-lora-strength" in solo.output
assert duo.exit_code == 0
assert "--activation-token-2" in duo.output
assert "--num-cleanups" in duo.output
assert "--i2i-lora-strength-2" in duo.output
def test_comfy_t2i_solo_requires_pipeline_outside_an_interactive_terminal() -> None:
runner = CliRunner()
result = invoke_cli(runner, ["comfy", "t2i", "solo", "studio portrait"])
assert result.exit_code == 2
assert "--pipeline" in result.output
def test_showcase_workflow_picker_keeps_project_order_without_duplicates() -> None:
picker_names = cli_comfy._workflow_picker_preset_names(
["pony", "project-only", "anima", "pony", "noob", "project-only"]
)
assert picker_names == ["pony", "project-only", "anima", "noob"]
def test_cli_root_help_shows_canonical_image_group_only() -> None:
runner = CliRunner()
result = invoke_cli(runner, ["--help"])
assert result.exit_code == 0
assert "│ img" in result.output
assert "│ image" not in result.output
assert "infer" not in result.output
def test_image_group_alias_is_removed() -> None:
runner = CliRunner()
result = invoke_cli(runner, ["image", "--help"])
assert result.exit_code != 0
assert "No such command 'image'" in result.output
def test_infer_group_is_removed() -> None:
runner = CliRunner()
result = invoke_cli(runner, ["infer", "--help"])
assert result.exit_code != 0
assert "No such command 'infer'" in result.output
def test_config_uses_native_apprc_index_path(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.delenv(KNF_APPRC_TOML_ENV_KEY, raising=False)
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config-home"))
expected_path = tmp_path / "config-home" / "knf" / "knf.apprc.toml"
assert KNEIFF_RC.spec.index_env_key == KNF_APPRC_TOML_ENV_KEY
assert KNEIFF_RC.spec.default_index_path() == expected_path
assert KNEIFF_RC.spec.index_path() == expected_path
def test_cli_help_shows_batch_config_and_resume_run_usage() -> None:
runner = CliRunner()
start_result = invoke_cli(runner, ["train", "start", "--help"])
prepare_result = invoke_cli(runner, ["train", "prepare", "--help"])
dataset_result = invoke_cli(runner, ["dataset", "plan", "--help"])
assert start_result.exit_code == 0
assert "[CONFIG]..." in start_result.output
assert "knf train start CONFIG..." in start_result.output
assert "knf train start CONFIG... --resume RUN" in start_result.output
assert "Pass RUN after --resume" in start_result.output
assert prepare_result.exit_code == 0
assert "[CONFIG]..." in prepare_result.output
assert dataset_result.exit_code == 0
assert "[CONFIG]..." in dataset_result.output
def test_comfy_showcase_help_shows_prompt_field() -> None:
runner = CliRunner()
output_option = "--" + "output"
result = invoke_cli(runner, ["comfy", "showcase", "--help"])
assert result.exit_code == 0
assert "--lora" in result.output
assert "--training" in result.output
assert "[STEPS]..." in result.output
assert "--prompts" in result.output
assert "--prompt-field" in result.output
assert output_option not in result.output
assert "auto" in result.output
assert "packaged preset" in result.output
assert "noob-nova" in result.output
assert "showcase_<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
@pytest.mark.parametrize(
"inference_option",
[
("--workflow", "willy"),
("--url", "http://comfy.test"),
("--seed", "42"),
],
)
def test_comfy_outpaint_safe_border_rejects_inference_only_options(
isolated_cli_project: Path,
inference_option: tuple[str, str],
) -> None:
from PIL import Image
source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
source_path.parent.mkdir(parents=True, exist_ok=True)
Image.new("RGB", (640, 960), color="white").save(source_path)
result = invoke_cli(
CliRunner(),
[
"comfy",
"outpaint",
str(source_path),
"--safe-border",
*inference_option,
],
input="",
)
assert result.exit_code != 0
normalized_output = " ".join(result.output.replace("│", " ").split())
assert "--safe-border cannot be combined" in normalized_output
def test_comfy_outpaint_safe_border_requires_configured_local_output_root(
isolated_cli_project: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from PIL import Image
source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
source_path.parent.mkdir(parents=True, exist_ok=True)
Image.new("RGB", (640, 960), color="white").save(source_path)
monkeypatch.setenv(COMFY_OUTPUT_DIR_ENV_KEY, "")
result = invoke_cli(
CliRunner(),
["comfy", "outpaint", str(source_path), "--safe-border"],
input="",
)
assert result.exit_code != 0
assert f"Set {COMFY_OUTPUT_DIR_ENV_KEY} to use --safe-border" in result.output
def test_comfy_outpaint_interrupt_exits_after_requesting_cancellation(
isolated_cli_project: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from PIL import Image
import kneiff.infer.comfy.outpaint as outpaint_module
source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
source_path.parent.mkdir(parents=True, exist_ok=True)
Image.new("RGB", (640, 960), color="white").save(source_path)
def interrupted_run_outpaint(_request: object, **_kwargs: object) -> None:
raise KeyboardInterrupt
monkeypatch.setattr(outpaint_module, "run_outpaint", interrupted_run_outpaint)
result = invoke_cli(
CliRunner(),
["comfy", "outpaint", str(source_path), "-w", "willy"],
input="",
)
assert result.exit_code == 130, result.output
assert "requested cancellation of its ComfyUI jobs" in result.stderr
def test_comfy_outpaint_from_pruned_aspects_resolves_and_outpaints_sources(
isolated_cli_project: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from PIL import Image
import kneiff.datasets.export.aspect_analysis as aspect_analysis
import kneiff.infer.comfy.outpaint as outpaint_module
import kneiff.training.lora.simpletuner_resolution_report as aspect_report
source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
source_path.parent.mkdir(parents=True, exist_ok=True)
Image.new("RGB", (640, 960), color="white").save(source_path)
config_path = isolated_cli_project / "configs" / "ANIMA.knf.yaml"
config_path.parent.mkdir(parents=True, exist_ok=True)
config_path.write_text(
'mappings:\n fullbody:\n - "0-FULLBODY"\n',
encoding="utf-8",
)
requests: list[outpaint_module.ComfyOutpaintRequest] = []
def fake_build_aspect_analysis(
plan: object,
*,
process_count: int,
progress_callback: object,
) -> object:
assert getattr(plan, "dataset_name") == "ANIMA"
assert process_count == 1
assert callable(progress_callback)
return object()
def fake_run_outpaint(
request: outpaint_module.ComfyOutpaintRequest,
**_kwargs: object,
) -> tuple[outpaint_module.ComfyOutpaintResult, ...]:
requests.append(request)
return ()
monkeypatch.setattr(
aspect_analysis,
"build_dataset_aspect_analysis",
fake_build_aspect_analysis,
)
monkeypatch.setattr(
aspect_report,
"pruned_aspect_bucket_source_paths",
lambda _analysis: ("SOURCE/0-FULLBODY/scene.png",),
)
monkeypatch.setattr(outpaint_module, "run_outpaint", fake_run_outpaint)
result = invoke_cli(
CliRunner(),
[
"comfy",
"outpaint",
"-a",
"--aspects-config",
"ANIMA",
"-w",
"willy",
"--square-fraction",
"0.5",
],
input="",
)
assert result.exit_code == 0, result.output
assert "ANIMA: found 1 physically pruned source image(s)." in result.stderr
assert requests[0].inputs == (source_path.resolve(),)
assert requests[0].square_fraction == 0.5
def test_comfy_outpaint_from_pruned_aspects_empty_result_is_a_noop(
isolated_cli_project: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
import kneiff.datasets.export.aspect_analysis as aspect_analysis
import kneiff.training.lora.simpletuner_resolution_report as aspect_report
config_path = isolated_cli_project / "configs" / "ANIMA.knf.yaml"
config_path.parent.mkdir(parents=True, exist_ok=True)
config_path.write_text(
'mappings:\n fullbody:\n - "0-FULLBODY"\n',
encoding="utf-8",
)
monkeypatch.setattr(
aspect_analysis,
"build_dataset_aspect_analysis",
lambda *_args, **_kwargs: object(),
)
monkeypatch.setattr(
aspect_report,
"pruned_aspect_bucket_source_paths",
lambda _analysis: (),
)
result = invoke_cli(
CliRunner(),
["comfy", "outpaint", "-a"],
input="",
)
assert result.exit_code == 0, result.output
assert "No source images are physically pruned" in result.stderr
def test_comfy_outpaint_rejects_pruned_aspects_with_other_input_modes(
isolated_cli_project: Path,
) -> None:
from PIL import Image
source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
source_path.parent.mkdir(parents=True, exist_ok=True)
Image.new("RGB", (640, 960), color="white").save(source_path)
config_path = isolated_cli_project / "configs" / "ANIMA.knf.yaml"
config_path.parent.mkdir(parents=True, exist_ok=True)
config_path.write_text(
'mappings:\n fullbody:\n - "0-FULLBODY"\n',
encoding="utf-8",
)
result = invoke_cli(
CliRunner(),
["comfy", "outpaint", "-a", str(source_path), "-w", "willy"],
input="",
)
assert result.exit_code != 0
normalized_output = " ".join(result.output.replace("│", " ").split())
assert "cannot be combined with positional image inputs" in normalized_output
piped_result = invoke_cli(
CliRunner(),
["comfy", "outpaint", "-a", "-w", "willy"],
input="SOURCE/0-FULLBODY/scene.png\n",
)
assert piped_result.exit_code != 0
normalized_piped_output = " ".join(piped_result.output.replace("│", " ").split())
assert "cannot be combined with positional image inputs" in normalized_piped_output
def test_comfy_outpaint_empty_source_stream_is_a_successful_noop(
isolated_cli_project: Path,
) -> None:
del isolated_cli_project
result = invoke_cli(
CliRunner(),
["comfy", "outpaint"],
input="",
)
assert result.exit_code == 0, result.output
assert "No outpaint source images received on stdin" in result.stderr
def test_comfy_outpaint_rejects_positional_images_and_source_stream(
isolated_cli_project: Path,
) -> None:
from PIL import Image
source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
source_path.parent.mkdir(parents=True, exist_ok=True)
Image.new("RGB", (640, 960), color="white").save(source_path)
result = invoke_cli(
CliRunner(),
["comfy", "outpaint", str(source_path), "-w", "willy"],
input="SOURCE/0-FULLBODY/scene.png\n",
)
assert result.exit_code != 0
normalized_output = " ".join(result.output.replace("│", " ").split())
assert (
"either positional image files or piped SOURCE/... paths" in normalized_output
)
def test_comfy_upscale_requires_storage(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.delenv("KNF_STORAGE", raising=False)
monkeypatch.delenv(KNF_APPRC_TOML_ENV_KEY, raising=False)
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config-home"))
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
runner = CliRunner()
result = invoke_cli(
runner,
[
"comfy",
"upscale",
str(image_path),
"--model",
"cli-upscaler.safetensors",
"--fast",
],
)
assert result.exit_code != 0
assert "KNF_STORAGE is required" in result.output
def test_comfy_upscale_quality_uses_brain_off_defaults(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
import kneiff.infer.comfy.upscale as upscale_module
from kneiff.infer.comfy.upscale import (
ComfyUpscaleRequest,
ComfyUpscaleResult,
)
for env_key in (
COMFY_UPSCALE_MODEL_ENV_KEY,
COMFY_UPSCALE_REFINER_UNET_ENV_KEY,
COMFY_UPSCALE_REFINER_CLIP_ENV_KEY,
COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY,
COMFY_UPSCALE_REFINER_VAE_ENV_KEY,
COMFY_UPSCALE_LORA_ENV_KEY,
COMFY_UPSCALE_LORA_TOKEN_ENV_KEY,
COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY,
COMFY_UPSCALE_DENOISE_BASE_ENV_KEY,
):
monkeypatch.delenv(env_key, raising=False)
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
requests: list[ComfyUpscaleRequest] = []
def fake_run_upscale(
request: ComfyUpscaleRequest,
**kwargs: object,
) -> tuple[ComfyUpscaleResult, ...]:
requests.append(request)
return ()
monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
_install_fake_comfy_model_client(monkeypatch)
runner = CliRunner()
result = invoke_cli(
runner, ["--skip-dotenv-layers", "comfy", "upscale", str(image_path)]
)
assert result.exit_code == 0, result.output
assert requests[0].upscale_model == DEFAULT_COMFY_UPSCALE_MODEL
assert requests[0].quality_workflow == "krea2"
assert requests[0].refiner is not None
assert requests[0].refiner.unet == DEFAULT_COMFY_UPSCALE_REFINER_UNET
assert requests[0].refiner.clip == DEFAULT_COMFY_UPSCALE_REFINER_CLIP
def test_comfy_upscale_z_image_uses_legacy_defaults(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
import kneiff.infer.comfy.upscale as upscale_module
from kneiff.infer.comfy.upscale import (
ComfyUpscaleRequest,
ComfyUpscaleResult,
)
for env_key in (
COMFY_UPSCALE_MODEL_ENV_KEY,
COMFY_UPSCALE_REFINER_UNET_ENV_KEY,
COMFY_UPSCALE_REFINER_CLIP_ENV_KEY,
COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY,
COMFY_UPSCALE_REFINER_VAE_ENV_KEY,
COMFY_UPSCALE_LORA_ENV_KEY,
COMFY_UPSCALE_LORA_TOKEN_ENV_KEY,
COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY,
COMFY_UPSCALE_DENOISE_BASE_ENV_KEY,
):
monkeypatch.delenv(env_key, raising=False)
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
requests: list[ComfyUpscaleRequest] = []
def fake_run_upscale(
request: ComfyUpscaleRequest,
**kwargs: object,
) -> tuple[ComfyUpscaleResult, ...]:
requests.append(request)
return ()
monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
_install_fake_comfy_model_client(monkeypatch)
runner = CliRunner()
result = invoke_cli(
runner,
["--skip-dotenv-layers", "comfy", "upscale", str(image_path), "-z"],
)
assert result.exit_code == 0, result.output
assert requests[0].quality_workflow == "z-image"
assert requests[0].refiner is not None
assert requests[0].refiner.unet == LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_UNET
assert requests[0].refiner.clip == LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP
assert requests[0].refiner.clip_type == (
LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP_TYPE
)
assert requests[0].refiner.vae == LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_VAE
def test_comfy_upscale_quality_rejects_unavailable_default_non_interactive(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
for env_key in (
COMFY_UPSCALE_MODEL_ENV_KEY,
COMFY_UPSCALE_REFINER_UNET_ENV_KEY,
COMFY_UPSCALE_REFINER_CLIP_ENV_KEY,
COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY,
COMFY_UPSCALE_REFINER_VAE_ENV_KEY,
COMFY_UPSCALE_LORA_ENV_KEY,
COMFY_UPSCALE_LORA_TOKEN_ENV_KEY,
COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY,
COMFY_UPSCALE_DENOISE_BASE_ENV_KEY,
):
monkeypatch.delenv(env_key, raising=False)
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
options = dict(_DEFAULT_COMFY_INPUT_OPTIONS)
options[("UNETLoader", "unet_name")] = ("other-unet.safetensors",)
_install_fake_comfy_model_client(monkeypatch, options=options)
runner = CliRunner()
result = invoke_cli(
runner, ["--skip-dotenv-layers", "comfy", "upscale", str(image_path)]
)
assert result.exit_code != 0
assert COMFY_UPSCALE_REFINER_UNET_ENV_KEY in result.output
assert DEFAULT_COMFY_UPSCALE_REFINER_UNET in result.output
assert "--fast" in result.output
def test_comfy_upscale_quality_cli_passes_refiner_and_lora(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
import kneiff.infer.comfy.upscale as upscale_module
from kneiff.infer.comfy.upscale import (
ComfyUpscaleRequest,
ComfyUpscaleResult,
)
models_dir = tmp_path / "models"
lora_path = models_dir / "models" / "loras" / "character.safetensors"
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
env_lora_path = models_dir / "models" / "loras" / "env-character.safetensors"
env_lora_path.write_bytes(b"lora")
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
monkeypatch.setenv(COMFY_UPSCALE_MODEL_ENV_KEY, "4x-test.safetensors")
monkeypatch.setenv(COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "z-image.safetensors")
monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "qwen.safetensors")
monkeypatch.setenv(COMFY_UPSCALE_LORA_ENV_KEY, "env-character.safetensors")
monkeypatch.setenv(COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, "env_token")
monkeypatch.setenv(COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, "0.9")
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
requests: list[ComfyUpscaleRequest] = []
def fake_run_upscale(
request: ComfyUpscaleRequest,
**kwargs: object,
) -> tuple[ComfyUpscaleResult, ...]:
requests.append(request)
return ()
monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
_install_fake_comfy_model_client(monkeypatch)
runner = CliRunner()
result = invoke_cli(
runner,
[
"comfy",
"upscale",
str(image_path),
"--style",
"flat",
"--lora",
"character.safetensors",
"--token",
"char_token",
"--lora-strength",
"0.4",
"--seed",
"123",
],
)
assert result.exit_code == 0, result.output
assert requests[0].fast is False
assert requests[0].style == "flat"
assert requests[0].refiner is not None
assert requests[0].refiner.unet == "z-image.safetensors"
assert requests[0].lora is not None
assert requests[0].lora.lora_name == "character.safetensors"
assert requests[0].lora.activation_token == "char_token"
assert requests[0].lora.strength == 0.4
assert requests[0].seed_root == 123
def test_comfy_upscale_quality_uses_configured_lora_path_chain(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
import kneiff.infer.comfy.upscale as upscale_module
from kneiff.infer.comfy.upscale import (
ComfyUpscaleRequest,
ComfyUpscaleResult,
)
models_dir = tmp_path / "models"
lora_path = models_dir / "models" / "loras" / "5th-ZI-1" / "2400.safetensors"
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
monkeypatch.setenv(COMFY_LORAS_DIR_1_ENV_KEY, "5th-ZI-1")
monkeypatch.setenv(COMFY_UPSCALE_LORA_ENV_KEY, "2400.safetensors")
monkeypatch.setenv(COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, "env_token")
monkeypatch.setenv(COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, "0.8")
monkeypatch.setenv(COMFY_UPSCALE_DENOISE_BASE_ENV_KEY, "0.18")
monkeypatch.setenv(COMFY_UPSCALE_MODEL_ENV_KEY, "4x-test.safetensors")
monkeypatch.setenv(COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "z-image.safetensors")
monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "qwen.safetensors")
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
requests: list[ComfyUpscaleRequest] = []
def fake_run_upscale(
request: ComfyUpscaleRequest,
**kwargs: object,
) -> tuple[ComfyUpscaleResult, ...]:
requests.append(request)
return ()
monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
_install_fake_comfy_model_client(monkeypatch)
runner = CliRunner()
result = invoke_cli(runner, ["comfy", "upscale", str(image_path), "-d", "0.19"])
assert result.exit_code == 0, result.output
assert requests[0].lora is not None
assert requests[0].lora.lora_name == "5th-ZI-1/2400.safetensors"
assert requests[0].lora.activation_token == "env_token"
assert requests[0].lora.strength == 0.8
assert requests[0].denoise_base == 0.19
def test_comfy_upscale_quality_allows_configured_lora_without_token(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
import kneiff.infer.comfy.upscale as upscale_module
from kneiff.infer.comfy.upscale import (
ComfyUpscaleRequest,
ComfyUpscaleResult,
)
models_dir = tmp_path / "models"
lora_path = models_dir / "models" / "loras" / "character.safetensors"
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
monkeypatch.setenv(COMFY_UPSCALE_LORA_ENV_KEY, "character.safetensors")
monkeypatch.delenv(COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, raising=False)
monkeypatch.setenv(COMFY_UPSCALE_MODEL_ENV_KEY, "4x-test.safetensors")
monkeypatch.setenv(COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "z-image.safetensors")
monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "qwen.safetensors")
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
requests: list[ComfyUpscaleRequest] = []
def fake_run_upscale(
request: ComfyUpscaleRequest,
**kwargs: object,
) -> tuple[ComfyUpscaleResult, ...]:
requests.append(request)
return ()
monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
_install_fake_comfy_model_client(monkeypatch)
runner = CliRunner()
result = invoke_cli(
runner, ["--skip-dotenv-layers", "comfy", "upscale", str(image_path)]
)
assert result.exit_code == 0, result.output
assert requests[0].lora is not None
assert requests[0].lora.activation_token == ""
def test_comfy_upscale_cli_requires_token_with_lora(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
models_dir = tmp_path / "models"
lora_path = models_dir / "models" / "loras" / "character.safetensors"
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
monkeypatch.setenv(COMFY_UPSCALE_MODEL_ENV_KEY, "4x-test.safetensors")
monkeypatch.setenv(COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "z-image.safetensors")
monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "qwen.safetensors")
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
_install_fake_comfy_model_client(monkeypatch)
runner = CliRunner()
result = invoke_cli(
runner,
[
"comfy",
"upscale",
str(image_path),
"--lora",
"character.safetensors",
],
)
assert result.exit_code != 0
assert "--token" in result.output
def test_comfy_upscale_fast_rejects_explicit_quality_options(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
_install_fake_comfy_model_client(monkeypatch)
runner = CliRunner()
result = invoke_cli(
runner,
[
"comfy",
"upscale",
str(image_path),
"--fast",
"--lora",
"character.safetensors",
],
)
assert result.exit_code != 0
assert "--fast" in result.output
assert "LoRA options" in result.output
def test_comfy_upscale_fast_rejects_z_image_workflow(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
runner = CliRunner()
result = invoke_cli(
runner,
[
"comfy",
"upscale",
str(image_path),
"--fast",
"-z",
],
)
assert result.exit_code != 0
assert "--z-image" in result.output
assert "--fast" in result.output
def test_comfy_upscale_explicit_unavailable_model_is_strict(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
_install_fake_comfy_model_client(monkeypatch)
monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
runner = CliRunner()
result = invoke_cli(
runner,
[
"comfy",
"upscale",
str(image_path),
"--model",
"missing-upscaler.safetensors",
"--fast",
],
)
assert result.exit_code != 0
assert "missing-upscaler.safetensors" in result.output
assert "Requested upscale model" in result.output
assert "Selected upscale model" not in result.output
def test_comfy_upscale_interactive_model_picker_is_current_run_only(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
import kneiff.infer.comfy.upscale as upscale_module
from kneiff.infer.comfy.upscale import (
ComfyUpscaleRequest,
ComfyUpscaleResult,
)
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
monkeypatch.delenv(COMFY_UPSCALE_MODEL_ENV_KEY, raising=False)
requests: list[ComfyUpscaleRequest] = []
options = dict(_DEFAULT_COMFY_INPUT_OPTIONS)
options[("UpscaleModelLoader", "model_name")] = (
"first-upscaler.safetensors",
"chosen-upscaler.safetensors",
)
actions = iter(("down", "select"))
def fake_run_upscale(
request: ComfyUpscaleRequest,
**kwargs: object,
) -> tuple[ComfyUpscaleResult, ...]:
requests.append(request)
return ()
monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
_install_fake_comfy_model_client(monkeypatch, options=options)
monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
runner = CliRunner()
result = invoke_cli(runner, ["comfy", "upscale", str(image_path), "--fast"])
assert result.exit_code == 0, result.output
assert requests[0].upscale_model == "chosen-upscaler.safetensors"
assert KneiffConfig().comfy.upscale_model == DEFAULT_COMFY_UPSCALE_MODEL
assert "Selected upscale model: chosen-upscaler.safetensors" in result.output
def test_comfy_upscale_interactive_lora_token_prompt(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
import kneiff.infer.comfy.upscale as upscale_module
from kneiff.infer.comfy.upscale import (
ComfyUpscaleRequest,
ComfyUpscaleResult,
)
models_dir = tmp_path / "models"
lora_path = models_dir / "models" / "loras" / "character.safetensors"
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
for env_key in (
COMFY_UPSCALE_MODEL_ENV_KEY,
COMFY_UPSCALE_REFINER_UNET_ENV_KEY,
COMFY_UPSCALE_REFINER_CLIP_ENV_KEY,
COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY,
COMFY_UPSCALE_REFINER_VAE_ENV_KEY,
COMFY_UPSCALE_LORA_ENV_KEY,
COMFY_UPSCALE_LORA_TOKEN_ENV_KEY,
COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY,
COMFY_UPSCALE_DENOISE_BASE_ENV_KEY,
):
monkeypatch.delenv(env_key, raising=False)
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
requests: list[ComfyUpscaleRequest] = []
def fake_run_upscale(
request: ComfyUpscaleRequest,
**kwargs: object,
) -> tuple[ComfyUpscaleResult, ...]:
requests.append(request)
return ()
monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
_install_fake_comfy_model_client(monkeypatch)
monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
runner = CliRunner()
result = invoke_cli(
runner,
[
"--skip-dotenv-layers",
"comfy",
"upscale",
str(image_path),
"--lora",
"character.safetensors",
],
input="char_token\n",
)
assert result.exit_code == 0, result.output
assert requests[0].lora is not None
assert requests[0].lora.activation_token == "char_token"
def test_comfy_showcase_cli_uses_preset_prompt_field(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
import kneiff.infer.comfy.showcase as showcase_module
from kneiff.infer.comfy.showcase import (
ShowcaseRunRequest,
ShowcaseRunResult,
)
prompts_path = tmp_path / "prompts.yaml"
prompts_path.write_text(
"""
version: 1
defaults:
seed: 123
prompts:
- id: first
uses: [showcase]
captions:
nlg:
- prose prompt
pony:
- pony tags
noob:
- noob tags
""",
encoding="utf-8",
)
requests: list[ShowcaseRunRequest] = []
def fake_run_showcase(
request: ShowcaseRunRequest,
**kwargs: object,
) -> ShowcaseRunResult:
requests.append(request)
return ShowcaseRunResult(results=())
monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)
runner = CliRunner()
result = invoke_cli(
runner,
[
"comfy",
"showcase",
"-w",
"pony",
"-p",
str(prompts_path),
],
)
assert result.exit_code == 0, result.output
assert requests[0].workflow_source.name == "showcase_pony.workflow.json"
assert requests[0].prompt_field == "pony"
assert requests[0].showcase_model == "pony"
assert [prompt.id for prompt in requests[0].prompt_catalog.prompts] == ["first"]
assert result.output.count("Generating 1 showcase image...") == 1
assert result.output.count("Showcase complete") == 1
result = invoke_cli(
runner,
[
"comfy",
"showcase",
"-w",
"noob-willy",
"-p",
str(prompts_path),
],
)
assert result.exit_code == 0, result.output
assert requests[1].workflow_source.name == "showcase_noob-willy.workflow.json"
assert requests[1].prompt_field == "noob"
assert requests[1].showcase_model == "noob"
def test_comfy_showcase_progress_stops_incomplete_after_generation_failure(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
import kneiff.infer.comfy.showcase as showcase_module
from kneiff.infer.comfy.showcase import (
ShowcasePromptResult,
ShowcaseRunRequest,
)
from kneiff.infer.comfy.showcase_runner import select_showcase_prompts
prompts_path = tmp_path / "prompts.yaml"
prompts_path.write_text(
"""
version: 1
defaults:
seed: 123
prompts:
- id: first
uses: [showcase]
captions:
nlg:
- prose prompt
""",
encoding="utf-8",
)
reporters: list[_CapturedShowcaseProgress] = []
def fail_run_showcase(
request: ShowcaseRunRequest,
*,
progress: Callable[[ShowcasePromptResult], None] | None = None,
) -> NoReturn:
prompts, _skipped = select_showcase_prompts(
request.prompt_catalog,
request.prompt_field,
)
prompt = prompts[0]
if progress is not None:
progress(
ShowcasePromptResult(
prompt=prompt,
prompt_id="queued-first",
filename_prefix="first",
files=(),
)
)
raise RuntimeError("generation broke")
monkeypatch.setattr(showcase_module, "run_showcase", fail_run_showcase)
monkeypatch.setattr(
cli_comfy,
"CliProgress",
_capturing_showcase_progress_factory(reporters),
)
result = invoke_cli(
CliRunner(),
["comfy", "showcase", "--workflow", "anima", "-p", str(prompts_path)],
)
assert result.exit_code == 1
assert reporters[0].total == 1
assert reporters[0].advances == ["first"]
assert reporters[0].exit_error is RuntimeError
assert "Showcase failed: generation broke" in result.output
assert "Showcase complete" not in result.output
def test_comfy_showcase_cli_infers_willy_noob_lora(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
isolated_cli_project: Path,
) -> None:
import kneiff.infer.comfy.showcase as showcase_module
from kneiff.infer.comfy.showcase import (
ShowcaseRunRequest,
ShowcaseRunResult,
)
models_dir = tmp_path / "comfyui-models"
lora_path = (
models_dir
/ "models"
/ "loras"
/ "_ladybird"
/ "Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors"
)
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
_write_noob_showcase_prompts(isolated_cli_project / "prompts.knf.yaml")
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
requests: list[ShowcaseRunRequest] = []
def fake_run_showcase(
request: ShowcaseRunRequest,
**kwargs: object,
) -> ShowcaseRunResult:
requests.append(request)
return ShowcaseRunResult(results=())
monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)
runner = CliRunner()
result = invoke_cli(
runner,
[
"comfy",
"showcase",
"--workflow",
"auto",
"--lora",
"_ladybird/Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors",
],
)
assert result.exit_code == 0, result.output
assert requests[0].workflow_source.name == "showcase_noob-willy.workflow.json"
catalog = requests[0].prompt_catalog
prompt_ids = [prompt.id for prompt in catalog.prompts]
assert prompt_ids[0] == "character_reference"
assert "aurora_portrait" in prompt_ids
assert catalog.overlay_prompt_count == 1
assert catalog.resolved_prompt_count == catalog.core_prompt_count + 1
assert requests[0].lora_name == (
"_ladybird/Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors"
)
assert requests[0].prompt_field == "noob"
assert requests[0].workflow_name == "noob-willy"
assert "Prompt sources" in result.output
assert "Workflow noob-willy" in result.output
assert "Outputs" in result.output
assert "Showcase complete" in result.output
def test_comfy_showcase_cli_prompts_for_workflow_when_omitted(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
import kneiff.infer.comfy.showcase as showcase_module
from kneiff.infer.comfy.showcase import (
ShowcaseRunRequest,
ShowcaseRunResult,
)
models_dir = tmp_path / "comfyui-models"
lora_path = (
models_dir
/ "models"
/ "loras"
/ "_ladybird"
/ "Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors"
)
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
prompts_path = tmp_path / "noob-prompts.yaml"
_write_noob_showcase_prompts(prompts_path)
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
actions = iter(("down", "select"))
monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
requests: list[ShowcaseRunRequest] = []
def fake_run_showcase(
request: ShowcaseRunRequest,
**kwargs: object,
) -> ShowcaseRunResult:
requests.append(request)
return ShowcaseRunResult(results=())
monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)
runner = CliRunner()
result = invoke_cli(
runner,
[
"comfy",
"showcase",
"--lora",
"_ladybird/Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors",
"--prompts",
str(prompts_path),
],
)
assert result.exit_code == 0, result.output
assert requests[0].workflow_source.name == "showcase_noob-willy.workflow.json"
assert requests[0].workflow_name == "noob-willy"
assert requests[0].prompt_field == "noob"
def test_comfy_showcase_cli_prompts_for_generic_noob_workflow(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
import kneiff.infer.comfy.showcase as showcase_module
from kneiff.infer.comfy.showcase import (
ShowcaseRunRequest,
ShowcaseRunResult,
)
models_dir = tmp_path / "comfyui-models"
lora_path = (
models_dir
/ "models"
/ "loras"
/ "_ladybird"
/ "Rook_Kaefer-v1_0-NOOB.comfyui.safetensors"
)
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
prompts_path = tmp_path / "noob-prompts.yaml"
_write_noob_showcase_prompts(prompts_path)
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
actions = iter(("down", "select"))
monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
requests: list[ShowcaseRunRequest] = []
def fake_run_showcase(
request: ShowcaseRunRequest,
**kwargs: object,
) -> ShowcaseRunResult:
requests.append(request)
return ShowcaseRunResult(results=())
monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)
runner = CliRunner()
result = invoke_cli(
runner,
[
"comfy",
"showcase",
"--workflow",
"auto",
"--lora",
"_ladybird/Rook_Kaefer-v1_0-NOOB.comfyui.safetensors",
"--prompts",
str(prompts_path),
],
)
assert result.exit_code == 0, result.output
assert requests[0].workflow_source.name == "showcase_noob-chenkin.workflow.json"
assert requests[0].workflow_name == "noob-chenkin"
assert "Selected workflow: noob-chenkin" in result.output
def test_comfy_showcase_cli_rejects_generic_noob_auto_non_interactive(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
models_dir = tmp_path / "comfyui-models"
lora_path = (
models_dir
/ "models"
/ "loras"
/ "_ladybird"
/ "Rook_Kaefer-v1_0-NOOB.comfyui.safetensors"
)
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
runner = CliRunner()
result = invoke_cli(
runner,
[
"comfy",
"showcase",
"--workflow",
"auto",
"--lora",
"_ladybird/Rook_Kaefer-v1_0-NOOB.comfyui.safetensors",
"--prompts",
str(SHOWCASE_PROMPTS_PATH),
],
)
assert result.exit_code != 0
assert "LoRA matches multiple workflows" in result.output
assert "noob-willy" in result.output
assert "noob-chenkin" in result.output
assert "noob-base" in result.output
assert "noob-nova" in result.output
assert "noob-scrimblosauce" in result.output
assert "pass --workflow" in result.output
def test_comfy_showcase_cli_prompts_when_packaged_workflow_is_missing(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
import kneiff.infer.comfy.showcase as showcase_module
from kneiff.infer.comfy.showcase import (
ShowcaseRunRequest,
ShowcaseRunResult,
ShowcaseWorkflowPreset,
)
import kneiff.infer.comfy.workflow_presets as workflow_presets
prompts_path = tmp_path / "prompts.yaml"
prompts_path.write_text(
"""
version: 1
defaults:
seed: 123
prompts:
- id: first
uses: [showcase]
captions:
nlg:
- prose prompt
""",
encoding="utf-8",
)
models_dir = tmp_path / "comfyui-models"
lora_path = models_dir / "models" / "loras" / "_ladybird" / "rook.safetensors"
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
monkeypatch.setitem(
workflow_presets.SHOWCASE_WORKFLOW_PRESETS,
"pony",
ShowcaseWorkflowPreset(
name="pony",
workflow_filename="missing_pony.workflow.json",
prompt_field="pony",
),
)
monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: "select")
requests: list[ShowcaseRunRequest] = []
def fake_run_showcase(
request: ShowcaseRunRequest,
**kwargs: object,
) -> ShowcaseRunResult:
requests.append(request)
return ShowcaseRunResult(results=())
monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)
runner = CliRunner()
result = invoke_cli(
runner,
[
"comfy",
"showcase",
"--workflow",
"pony",
"--lora",
"_ladybird/rook.safetensors",
"--prompts",
str(prompts_path),
],
)
assert result.exit_code == 0, result.output
assert requests[0].workflow_source.name == "showcase_anima.workflow.json"
assert requests[0].workflow_name == "anima"
assert "Selected workflow: anima" in result.output
def test_comfy_showcase_lora_picker_descends_into_directories(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
models_dir = tmp_path / "comfyui-models"
lora_path = models_dir / "models" / "loras" / "nested" / "rook.safetensors"
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
actions = iter(("select", "down", "select"))
output = io.StringIO()
monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
selected = cli_comfy.select_showcase_lora_interactively(
models_dir,
console=Console(file=output, force_terminal=False),
)
assert selected == "nested/rook.safetensors"
assert "nested/" in output.getvalue()
def test_comfy_showcase_lora_picker_starts_in_configured_directory(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
models_dir = tmp_path / "comfyui-models"
start_dir = models_dir / "models" / "loras" / "_ladybird" / "v1_0"
lora_path = start_dir / "rook.safetensors"
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
output = io.StringIO()
monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: "select")
selected = cli_comfy.select_showcase_lora_interactively(
models_dir,
console=Console(file=output, force_terminal=False),
start_dir=start_dir,
)
assert selected == "_ladybird/v1_0/rook.safetensors"
assert "Directory: _ladybird/v1_0" in output.getvalue()
def test_comfy_showcase_lora_picker_start_dir_env_rejects_missing_directory(
tmp_path: Path,
) -> None:
models_dir = tmp_path / "comfyui-models"
missing_dir = models_dir / "models" / "loras" / "_ladybird" / "v1_0"
config = ComfyConfig(
models_dir=str(models_dir),
loras_dir_1=str(missing_dir),
)
with pytest.raises(ValueError, match="COMFY_LORAS_DIR_1 directory not found"):
cli_comfy._resolve_comfy_loras_start_dir(models_dir, config=config)
def test_comfy_showcase_lora_picker_start_dir_env_rejects_outside_lora_root(
tmp_path: Path,
) -> None:
models_dir = tmp_path / "comfyui-models"
outside_dir = tmp_path / "other-loras"
outside_dir.mkdir()
config = ComfyConfig(
models_dir=str(models_dir),
loras_dir_1=str(outside_dir),
)
with pytest.raises(ValueError, match="COMFY_LORAS_DIR_1 must be below"):
cli_comfy._resolve_comfy_loras_start_dir(models_dir, config=config)
def test_comfy_showcase_training_picker_uses_step_and_cleans_staged_lora(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
isolated_cli_project: Path,
) -> None:
import kneiff.infer.comfy.showcase as showcase_module
from kneiff.infer.comfy.showcase import ShowcaseRunRequest, ShowcaseRunResult
output_root = (
isolated_cli_project
/ "TRAINING"
/ "ANIMA_1"
/ "_simpletuner-output"
/ "checkpoint-100"
)
output_root.mkdir(parents=True)
(output_root / "pytorch_lora_weights.safetensors").write_bytes(b"raw")
preferred_path = output_root / "pytorch_lora_weights.comfyui.safetensors"
preferred_path.write_bytes(b"comfyui")
second_step = output_root.parent / "checkpoint-200" / "weights.safetensors"
second_step.parent.mkdir()
second_step.write_bytes(b"second-step")
models_dir = tmp_path / "comfyui-models"
(models_dir / "models" / "loras").mkdir(parents=True)
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
actions = iter(("select", "down", "select"))
monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
requests: list[ShowcaseRunRequest] = []
staged_paths: list[Path] = []
def fake_run_showcase(
request: ShowcaseRunRequest,
**kwargs: object,
) -> ShowcaseRunResult:
requests.append(request)
assert request.lora_name is not None
staged_path = models_dir / "models" / "loras" / request.lora_name
assert staged_path.read_bytes() == b"comfyui"
staged_paths.append(staged_path)
return ShowcaseRunResult(results=())
monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)
result = invoke_cli(
CliRunner(),
[
"comfy",
"showcase",
"-t",
"100",
"200",
"100",
"--workflow",
"anima",
],
)
assert result.exit_code == 0, result.output
assert len(requests) == 1
assert requests[0].lora_name is not None
assert requests[0].lora_name.endswith(
"ANIMA_1/_simpletuner-output/checkpoint-100/"
"pytorch_lora_weights.comfyui.safetensors"
)
assert "Selected TRAINING LoRA:" in result.output
assert not staged_paths[0].exists()
assert not (models_dir / "models" / "loras" / ".kneiff-training").exists()
def test_comfy_showcase_training_hides_root_exports_without_checkpoint_steps(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
isolated_cli_project: Path,
) -> None:
import kneiff.infer.comfy.showcase as showcase_module
from kneiff.infer.comfy.showcase import ShowcaseRunRequest, ShowcaseRunResult
output_root = isolated_cli_project / "TRAINING" / "ANIMA_1" / "_simpletuner-output"
output_root.mkdir(parents=True)
root_lora = output_root / "root.comfyui.safetensors"
root_lora.write_bytes(b"root")
models_dir = tmp_path / "comfyui-models"
(models_dir / "models" / "loras").mkdir(parents=True)
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
requests: list[ShowcaseRunRequest] = []
def fake_run_showcase(
request: ShowcaseRunRequest,
**kwargs: object,
) -> ShowcaseRunResult:
requests.append(request)
return ShowcaseRunResult(results=())
monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)
result = invoke_cli(
CliRunner(),
["comfy", "showcase", "-t", "--workflow", "anima"],
)
assert result.exit_code != 0
assert "No checkpoint-<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
@pytest.mark.parametrize(
("args", "expected"),
[
(["comfy", "showcase", "100"], "STEPS require --training"),
(
["comfy", "showcase", "-t", "--lora", "character.safetensors"],
"Do not combine --training",
),
],
)
def test_comfy_showcase_training_rejects_invalid_option_combinations(
args: list[str],
expected: str,
) -> None:
result = invoke_cli(CliRunner(), args)
assert result.exit_code != 0
assert expected in result.output
def test_comfy_showcase_training_requires_interactive_terminal() -> None:
result = invoke_cli(CliRunner(), ["comfy", "showcase", "-t"])
assert result.exit_code != 0
assert "requires an interactive" in result.output
assert "terminal for LoRA selection" in result.output
@pytest.mark.parametrize(
("step", "expected"),
[
("0", "positive"),
("200", "Available steps: 100"),
],
)
def test_comfy_showcase_training_rejects_invalid_or_missing_steps(
monkeypatch: pytest.MonkeyPatch,
isolated_cli_project: Path,
step: str,
expected: str,
) -> None:
checkpoint_root = (
isolated_cli_project
/ "TRAINING"
/ "ANIMA_1"
/ "_simpletuner-output"
/ "checkpoint-100"
)
checkpoint_root.mkdir(parents=True)
(checkpoint_root / "weights.safetensors").write_bytes(b"lora")
monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
result = invoke_cli(CliRunner(), ["comfy", "showcase", "-t", step])
assert result.exit_code != 0
assert expected in result.output
def test_comfy_showcase_training_picker_opens_runs_before_checkpoints(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
from kneiff.training.lora.output_discovery import TrainingLoraCandidate
first_path = tmp_path / "first.comfyui.safetensors"
second_path = tmp_path / "second.safetensors"
first_path.write_bytes(b"first")
second_path.write_bytes(b"second")
candidates = (
TrainingLoraCandidate(
path=first_path,
relative_path=Path(
"first_1/_simpletuner-output/checkpoint-50/first.comfyui.safetensors"
),
checkpoint_step=50,
comfyui_native=True,
),
TrainingLoraCandidate(
path=second_path,
relative_path=Path(
"second_1/_simpletuner-output/checkpoint-100/second.safetensors"
),
checkpoint_step=100,
comfyui_native=False,
),
)
actions = iter(("down", "select", "down", "select"))
output = io.StringIO()
monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
selected = cli_comfy.select_training_lora_interactively(
candidates,
console=Console(file=output, force_terminal=False),
)
assert selected == (candidates[1],)
assert "Select TRAINING run" in output.getvalue()
assert "first_1/ (1 LoRA(s))" in output.getvalue()
assert candidates[1].relative_path.as_posix() in output.getvalue()
def test_comfy_showcase_training_picker_can_be_cancelled(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
from kneiff.training.lora.output_discovery import TrainingLoraCandidate
lora_path = tmp_path / "weights.safetensors"
lora_path.write_bytes(b"lora")
candidate = TrainingLoraCandidate(
path=lora_path,
relative_path=Path(
"ready_1/_simpletuner-output/checkpoint-100/weights.safetensors"
),
checkpoint_step=100,
comfyui_native=False,
)
monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: "cancel")
selected = cli_comfy.select_training_lora_interactively(
(candidate,),
console=Console(file=io.StringIO(), force_terminal=False),
)
assert selected is None
def test_comfy_showcase_training_picker_selects_multiple_loras(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
from kneiff.training.lora.output_discovery import TrainingLoraCandidate
candidates = tuple(
TrainingLoraCandidate(
path=tmp_path / f"weights-{step}.safetensors",
relative_path=Path(
f"ready_1/_simpletuner-output/checkpoint-{step}/weights.safetensors"
),
checkpoint_step=step,
comfyui_native=False,
)
for step in (100, 200)
)
for candidate in candidates:
candidate.path.write_bytes(b"lora")
actions = iter(("select", "down", "toggle", "down", "toggle", "select"))
output = io.StringIO()
monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
selected = cli_comfy.select_training_lora_interactively(
candidates,
console=Console(file=output, force_terminal=False),
)
assert selected == candidates
assert "Selected TRAINING LoRAs:" in output.getvalue()
assert "[x]" in output.getvalue()
def test_comfy_showcase_training_infers_workflow_from_relative_path(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
isolated_cli_project: Path,
) -> None:
import kneiff.infer.comfy.showcase as showcase_module
from kneiff.infer.comfy.showcase import ShowcaseRunRequest, ShowcaseRunResult
lora_path = (
isolated_cli_project
/ "TRAINING"
/ "ANIMA_1"
/ "_simpletuner-output"
/ "checkpoint-100"
/ "weights.comfyui.safetensors"
)
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
models_dir = tmp_path / "comfyui-models"
(models_dir / "models" / "loras").mkdir(parents=True)
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
actions = iter(("select", "down", "select"))
monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
requests: list[ShowcaseRunRequest] = []
def fake_run_showcase(
request: ShowcaseRunRequest,
**kwargs: object,
) -> ShowcaseRunResult:
requests.append(request)
return ShowcaseRunResult(results=())
monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)
result = invoke_cli(CliRunner(), ["comfy", "showcase", "-t", "100"])
assert result.exit_code == 0, result.output
assert requests[0].workflow_name == "anima"
def test_comfy_showcase_training_falls_back_to_complete_workflow_picker(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
isolated_cli_project: Path,
) -> None:
import kneiff.infer.comfy.showcase as showcase_module
from kneiff.infer.comfy.showcase import ShowcaseRunRequest, ShowcaseRunResult
lora_path = (
isolated_cli_project
/ "TRAINING"
/ "mystery_1"
/ "_simpletuner-output"
/ "checkpoint-100"
/ "weights.safetensors"
)
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
models_dir = tmp_path / "comfyui-models"
(models_dir / "models" / "loras").mkdir(parents=True)
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
actions = iter(("select", "down", "select", "select"))
monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
requests: list[ShowcaseRunRequest] = []
def fake_run_showcase(
request: ShowcaseRunRequest,
**kwargs: object,
) -> ShowcaseRunResult:
requests.append(request)
return ShowcaseRunResult(results=())
monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)
result = invoke_cli(CliRunner(), ["comfy", "showcase", "-t", "100"])
assert result.exit_code == 0, result.output
assert requests[0].workflow_name == "anima"
assert "Selected workflow: anima" in result.output
def test_comfy_showcase_training_explicit_auto_keeps_inference_failure(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
isolated_cli_project: Path,
) -> None:
lora_path = (
isolated_cli_project
/ "TRAINING"
/ "mystery_1"
/ "_simpletuner-output"
/ "checkpoint-100"
/ "weights.safetensors"
)
lora_path.parent.mkdir(parents=True)
lora_path.write_bytes(b"lora")
models_dir = tmp_path / "comfyui-models"
lora_root = models_dir / "models" / "loras"
lora_root.mkdir(parents=True)
monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
actions = iter(("select", "down", "select"))
monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
result = invoke_cli(
CliRunner(),
["comfy", "showcase", "-t", "100", "--workflow", "auto"],
)
assert result.exit_code != 0
assert "Could not infer a workflow" in result.output
assert not (lora_root / ".kneiff-training").exists()
def test_train_help_skips_storage_bootstrap(monkeypatch: pytest.MonkeyPatch) -> None:
runner = CliRunner()
monkeypatch.delenv("KNF_STORAGE", raising=False)
help_result = invoke_cli(runner, ["train", "--help"])
train_result = invoke_cli(runner, ["train"])
assert help_result.exit_code == 0, help_result.output
assert train_result.exit_code == 2
assert "Training run lifecycle workflows." in train_result.output
assert "KNF_STORAGE" not in help_result.output
assert "KNF_STORAGE" not in train_result.output
def test_config_doctor_output_uses_only_knf_env_names() -> None:
runner = CliRunner()
forbidden = ("KNF_STORAGE_ROOT", "KNEIFF_APPRC_TOML", "KNEIFF_CONFIG_FILE")
result = invoke_cli(runner, ["config", "doctor"])
assert "KNF_STORAGE" in result.output
assert "KNF_APPRC_TOML" in result.output
assert not any(name in result.output for name in forbidden)
def test_cli_app_import_keeps_runtime_backends_lazy() -> None:
script = (
"import sys\n"
"import kneiff.cli.app\n"
"heavy_modules = {\n"
" 'huggingface_hub', 'numpy', 'openai', 'openpyxl', 'pandas',\n"
" 'requests', 'safetensors', 'spandrel', 'torch', 'torchvision',\n"
" 'transformers',\n"
"}\n"
"loaded = sorted(name for name in heavy_modules if name in sys.modules)\n"
"print('\\n'.join(loaded))\n"
"raise SystemExit(1 if loaded else 0)\n"
)
result = subprocess.run(
[sys.executable, "-c", script],
capture_output=True,
text=True,
check=False,
)
assert result.returncode == 0, result.stdout + result.stderr
def test_lora_command_group_is_removed() -> None:
runner = CliRunner()
result = invoke_cli(runner, ["lora", "--help"])
assert result.exit_code != 0
def test_old_diff_subcommand_is_removed() -> None:
runner = CliRunner()
result = invoke_cli(runner, ["diff", "--help"])
assert result.exit_code != 0
def test_img_tag_help_shows_selected_tag_output_options(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("KNF_STORAGE", raising=False)
runner = CliRunner()
result = invoke_cli(runner, ["img", "tag", "--help"])
assert result.exit_code == 0
assert "--txt" in result.output
assert "--comma" in result.output
assert "-c" in result.output
assert "Write .txt sidecars next to images" in result.output
assert "probability CSV output" in result.output
assert "Legacy JTP-3 snapshot threshold" in result.output
def test_img_tag_cli_passes_text_output_options(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("KNF_STORAGE", raising=False)
runner = CliRunner()
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
captured: dict[str, tag_jtp3.Jtp3RunConfig] = {}
def fake_run_jtp3(
config: tag_jtp3.Jtp3RunConfig,
snapshot_config: tag_jtp3.Jtp3SnapshotConfig | None = None,
) -> subprocess.CompletedProcess[str]:
captured["config"] = config
return subprocess.CompletedProcess([], 0)
monkeypatch.setattr(tag_jtp3, "run_jtp3", fake_run_jtp3)
result = invoke_cli(runner, ["img", "tag", str(image_path), "--txt", "--comma"])
assert result.exit_code == 0
assert captured["config"].write_txt is True
assert captured["config"].comma_separated is True
result = invoke_cli(runner, ["img", "tag", str(image_path), "-c"])
assert result.exit_code == 0
assert captured["config"].write_txt is False
assert captured["config"].comma_separated is True
def test_img_tag_rejects_csv_stdout_with_txt(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("KNF_STORAGE", raising=False)
runner = CliRunner()
image_path = tmp_path / "image.png"
image_path.write_bytes(b"image")
result = invoke_cli(
runner, ["img", "tag", str(image_path), "--csv-stdout", "--txt"]
)
assert result.exit_code != 0
assert "--csv-stdout cannot be combined with --txt or --comma" in result.output
def test_package_import_smoke() -> None:
for module_name in (
"kneiff",
"kneiff.clients",
"kneiff.datasets",
"kneiff.datasets.manifest",
"kneiff.datasets.export",
"kneiff.training",
"kneiff.training.lora",
"kneiff.infer",
"kneiff.cli",
"kneiff.cli.app",
"kneiff.main",
"kneiff.app.workbench",
"kneiff.utils",
"kneiff.utils.image.caption",
"kneiff_dev",
):
assert importlib.import_module(module_name)
def test_removed_import_paths_are_not_shimmed() -> None:
for module_name in (
"kneiff.training.captions",
"kneiff.training.dataset",
"kneiff.training.manifest",
"kneiff.training.augmentations",
"kneiff.inference",
"kneiff.cli_dataset",
"kneiff.cli_dataset_report",
"kneiff.cli_image",
"kneiff.cli_lora",
"kneiff.datasets.manifest_to_captions",
"kneiff.datasets.captions",
"kneiff.datasets.augment",
"kneiff.datasets.manifest.core",
):
with pytest.raises(ModuleNotFoundError):
importlib.import_module(module_name)