path: root/tests/test_config.py

from __future__ import annotations

from pathlib import Path

import pytest

from mcp_sd15.config import ConfigError, load_config


def make_models(root: Path) -> None:
    for folder, names in {
        "checkpoints": ["base.safetensors"],
        "vae": ["vae.safetensors"],
        "lora": ["a.safetensors", "b.safetensors"],
        "controlnet": ["depth.safetensors", "pose.safetensors", "inpaint.safetensors"],
    }.items():
        destination = root / folder
        destination.mkdir(parents=True)
        for name in names:
            (destination / name).touch()


def valid_toml(root: Path) -> str:
    return f'''[model]
root = "{root.as_posix()}"
checkpoint = "base.safetensors"
vae = "vae.safetensors"
[[model.lora]]
file = "a.safetensors"
weight = 0.8
[[model.lora]]
file = "b.safetensors"
weight = -0.2
[generation]
steps = 20
cfg_scale = 6.5
scheduler = "euler_a"
default_size = "portrait_512"
[controlnet.depth]
enabled = true
model = "depth.safetensors"
preprocessor = "midas"
[controlnet.openpose]
enabled = true
model = "pose.safetensors"
[controlnet.inpaint]
enabled = true
model = "inpaint.safetensors"
[output]
enabled = true
directory = "generated"
'''


def test_loads_and_resolves_all_model_paths(tmp_path: Path) -> None:
    root = tmp_path / "models"
    make_models(root)
    path = tmp_path / "config.toml"
    path.write_text(valid_toml(root), encoding="utf-8")

    config = load_config(path)

    assert config.model.checkpoint_path() == root / "checkpoints" / "base.safetensors"
    assert config.model.vae_path() == root / "vae" / "vae.safetensors"
    assert config.model.lora_paths() == (
        root / "lora" / "a.safetensors",
        root / "lora" / "b.safetensors",
    )
    assert config.generation.default_size == "portrait_512"
    assert config.output.directory == tmp_path / "generated"
    assert config.controlnet_path("openpose_full") == root / "controlnet" / "pose.safetensors"


@pytest.mark.parametrize(
    "line, message",
    [
        ('checkpoint = "../base.safetensors"', "file name"),
        ("steps = 0", "steps"),
        ('scheduler = "dpmpp_2m"', "scheduler"),
        ('scheduler = "euler"', "scheduler"),
        ('scheduler = "unknown"', "scheduler"),
        ('default_size = "1024"', "default_size"),
        ('preprocessor = "zoe"', "preprocessor"),
    ],
)
def test_rejects_invalid_values(tmp_path: Path, line: str, message: str) -> None:
    root = tmp_path / "models"
    make_models(root)
    text = valid_toml(root)
    if line.startswith("checkpoint"):
        text = text.replace('checkpoint = "base.safetensors"', line)
    elif line.startswith("steps"):
        text = text.replace("steps = 20", line)
    elif line.startswith("scheduler"):
        text = text.replace('scheduler = "euler_a"', line)
    elif line.startswith("default_size"):
        text = text.replace('default_size = "portrait_512"', line)
    else:
        text = text.replace('preprocessor = "midas"', line)
    path = tmp_path / "bad.toml"
    path.write_text(text, encoding="utf-8")

    with pytest.raises(ConfigError, match=message):
        load_config(path)


def test_missing_model_fails_startup_validation(tmp_path: Path) -> None:
    root = tmp_path / "models"
    make_models(root)
    (root / "controlnet" / "depth.safetensors").unlink()
    path = tmp_path / "config.toml"
    path.write_text(valid_toml(root), encoding="utf-8")

    with pytest.raises(ConfigError, match=r"depth\.safetensors"):
        load_config(path)


def test_can_skip_path_checks_for_schema_only_use(tmp_path: Path) -> None:
    path = tmp_path / "config.toml"
    path.write_text(
        '[model]\nroot = "missing"\ncheckpoint = "base.safetensors"\n', encoding="utf-8"
    )
    config = load_config(path, validate_paths=False)
    assert config.model.root == (tmp_path / "missing").resolve()
    assert config.generation.scheduler == "dpmpp_2m_karras"


def test_server_defaults_to_stdio_when_omitted(tmp_path: Path) -> None:
    path = tmp_path / "config.toml"
    path.write_text(
        f'[model]\nroot = "{tmp_path.as_posix()}"\ncheckpoint = "base.safetensors"\n',
        encoding="utf-8",
    )
    config = load_config(path, validate_paths=False)
    assert config.server.transport == "stdio"
    assert config.server.host == "127.0.0.1"
    assert config.server.port == 8000


def test_server_reads_streamable_http(tmp_path: Path) -> None:
    path = tmp_path / "config.toml"
    path.write_text(
        f'[model]\nroot = "{tmp_path.as_posix()}"\ncheckpoint = "base.safetensors"\n'
        '[server]\ntransport = "streamable-http"\nhost = "0.0.0.0"\nport = 9000\n',
        encoding="utf-8",
    )
    config = load_config(path, validate_paths=False)
    assert config.server.transport == "streamable-http"
    assert config.server.host == "0.0.0.0"
    assert config.server.port == 9000


@pytest.mark.parametrize(
    "transport",
    ["sse", "http", "ws", "unknown"],
)
def test_rejects_invalid_transport(tmp_path: Path, transport: str) -> None:
    path = tmp_path / "config.toml"
    path.write_text(
        f'[model]\nroot = "{tmp_path.as_posix()}"\ncheckpoint = "base.safetensors"\n'
        f'[server]\ntransport = "{transport}"\n',
        encoding="utf-8",
    )
    with pytest.raises(ConfigError, match="transport"):
        load_config(path, validate_paths=False)


@pytest.mark.parametrize(
    "port, message",
    [
        (0, "port"),
        (65536, "port"),
        (-1, "port"),
        ("8000", "port"),
    ],
)
def test_rejects_invalid_port(tmp_path: Path, port: str | int, message: str) -> None:
    path = tmp_path / "config.toml"
    port_literal = f'"{port}"' if isinstance(port, str) else str(port)
    path.write_text(
        f'[model]\nroot = "{tmp_path.as_posix()}"\ncheckpoint = "base.safetensors"\n'
        f'[server]\ntransport = "streamable-http"\nport = {port_literal}\n',
        encoding="utf-8",
    )
    with pytest.raises(ConfigError, match=message):
        load_config(path, validate_paths=False)


def test_rejects_empty_host(tmp_path: Path) -> None:
    path = tmp_path / "config.toml"
    path.write_text(
        f'[model]\nroot = "{tmp_path.as_posix()}"\ncheckpoint = "base.safetensors"\n'
        '[server]\ntransport = "streamable-http"\nhost = ""\n',
        encoding="utf-8",
    )
    with pytest.raises(ConfigError, match="host"):
        load_config(path, validate_paths=False)


def test_prompt_defaults_to_empty_when_omitted(tmp_path: Path) -> None:
    path = tmp_path / "config.toml"
    path.write_text(
        f'[model]\nroot = "{tmp_path.as_posix()}"\ncheckpoint = "base.safetensors"\n',
        encoding="utf-8",
    )
    config = load_config(path, validate_paths=False)
    assert config.prompt.positive == ""
    assert config.prompt.negative == ""


def test_prompt_reads_files_from_root_prompts(tmp_path: Path) -> None:
    root = tmp_path / "models"
    (root / "checkpoints").mkdir(parents=True)
    (root / "checkpoints" / "base.safetensors").touch()
    (root / "prompts").mkdir()
    (root / "prompts" / "positive.txt").write_text("masterpiece, best quality", encoding="utf-8")
    (root / "prompts" / "negative.txt").write_text("low quality, blurry", encoding="utf-8")
    path = tmp_path / "config.toml"
    path.write_text(
        f'[model]\nroot = "{root.as_posix()}"\ncheckpoint = "base.safetensors"\n'
        '[prompt]\npositive_file = "positive.txt"\nnegative_file = "negative.txt"\n',
        encoding="utf-8",
    )
    config = load_config(path)
    assert config.prompt.positive == "masterpiece, best quality"
    assert config.prompt.negative == "low quality, blurry"


def test_prompt_rejects_path_traversal(tmp_path: Path) -> None:
    root = tmp_path / "models"
    (root / "checkpoints").mkdir(parents=True)
    (root / "checkpoints" / "base.safetensors").touch()
    path = tmp_path / "config.toml"
    path.write_text(
        f'[model]\nroot = "{root.as_posix()}"\ncheckpoint = "base.safetensors"\n'
        '[prompt]\npositive_file = "../escape.txt"\n',
        encoding="utf-8",
    )
    with pytest.raises(ConfigError, match="file name"):
        load_config(path, validate_paths=False)


def test_prompt_rejects_missing_file(tmp_path: Path) -> None:
    root = tmp_path / "models"
    (root / "checkpoints").mkdir(parents=True)
    (root / "checkpoints" / "base.safetensors").touch()
    path = tmp_path / "config.toml"
    path.write_text(
        f'[model]\nroot = "{root.as_posix()}"\ncheckpoint = "base.safetensors"\n'
        '[prompt]\npositive_file = "absent.txt"\n',
        encoding="utf-8",
    )
    with pytest.raises(ConfigError, match="prompt file"):
        load_config(path, validate_paths=False)


def test_prompt_accepts_only_positive(tmp_path: Path) -> None:
    root = tmp_path / "models"
    (root / "checkpoints").mkdir(parents=True)
    (root / "checkpoints" / "base.safetensors").touch()
    (root / "prompts").mkdir()
    (root / "prompts" / "positive.txt").write_text("masterpiece", encoding="utf-8")
    path = tmp_path / "config.toml"
    path.write_text(
        f'[model]\nroot = "{root.as_posix()}"\ncheckpoint = "base.safetensors"\n'
        '[prompt]\npositive_file = "positive.txt"\n',
        encoding="utf-8",
    )
    config = load_config(path)
    assert config.prompt.positive == "masterpiece"
    assert config.prompt.negative == ""


def test_generation_defaults_clip_skip_and_pad_last_block(tmp_path: Path) -> None:
    path = tmp_path / "config.toml"
    path.write_text(
        f'[model]\nroot = "{tmp_path.as_posix()}"\ncheckpoint = "base.safetensors"\n',
        encoding="utf-8",
    )
    config = load_config(path, validate_paths=False)
    assert config.generation.clip_skip == 0
    assert config.generation.pad_last_block is False


def test_generation_reads_clip_skip_and_pad_last_block(tmp_path: Path) -> None:
    path = tmp_path / "config.toml"
    path.write_text(
        f'[model]\nroot = "{tmp_path.as_posix()}"\ncheckpoint = "base.safetensors"\n'
        "[generation]\nclip_skip = 2\npad_last_block = true\n",
        encoding="utf-8",
    )
    config = load_config(path, validate_paths=False)
    assert config.generation.clip_skip == 2
    assert config.generation.pad_last_block is True


@pytest.mark.parametrize(
    "clip_skip, message",
    [
        (-1, "clip_skip"),
        (13, "clip_skip"),
        (2.5, "clip_skip"),
        (True, "clip_skip"),
    ],
)
def test_rejects_invalid_clip_skip(
    tmp_path: Path, clip_skip: float | int | bool, message: str
) -> None:
    path = tmp_path / "config.toml"
    literal = "true" if isinstance(clip_skip, bool) else str(clip_skip)
    path.write_text(
        f'[model]\nroot = "{tmp_path.as_posix()}"\ncheckpoint = "base.safetensors"\n'
        f"[generation]\nclip_skip = {literal}\n",
        encoding="utf-8",
    )
    with pytest.raises(ConfigError, match=message):
        load_config(path, validate_paths=False)