llamacppctl/tests/test_docker_ops.py

import sys
from pathlib import Path
from unittest.mock import patch, MagicMock

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from llamacppctl.docker_ops import (  # noqa: E402
    build_run_command,
    container_exists,
    container_running,
    docker_available,
    format_command_for_display,
    inspect_container,
)
from llamacppctl.schema import ServerConfig  # noqa: E402


def make_cfg(**overrides) -> ServerConfig:
    base = dict(
        image="ghcr.io/ggml-org/llama.cpp:server-cuda",
        hf_home=Path("/srv/models"),
        model_path="qwen3/default.gguf",
        container_name="test_llm",
        host_port=8001,
        container_port=8000,
        model_alias="default_llm",
        gpu_device="0",
        restart_policy="unless-stopped",
        ctx_size=262144,
        n_predict=16384,
        temp=0.65,
        top_p=0.80,
        top_k=20,
        min_p=0.01,
        repeat_penalty=1.05,
        main_gpu=0,
        ngl=999,
        batch_size=1024,
        ubatch_size=512,
        parallel=1,
        cache_type_k="q4_0",
        cache_type_v="q4_0",
        reasoning="on",
        jinja=True,
        fa=True,
        kv_unified=True,
        cont_batching=True,
        no_context_shift=True,
        host="0.0.0.0",
        health_endpoint="/health",
        models_endpoint="/v1/models",
        chat_endpoint="/v1/chat/completions",
        timeout=300,
        poll_interval=2.0,
        lock_file=Path("/tmp/llamacppctl.test_llm.lock"),
    )
    base.update(overrides)
    return ServerConfig(**base)


def test_docker_available_true():
    with patch("subprocess.run") as mock_run:
        mock_run.return_value = MagicMock(returncode=0)
        assert docker_available() is True


def test_docker_available_false_when_missing():
    with patch("subprocess.run", side_effect=FileNotFoundError()):
        assert docker_available() is False


def test_container_exists_true():
    with patch("llamacppctl.docker_ops._run") as mock_run:
        mock_run.return_value = MagicMock(stdout="test_llm\nother\n")
        assert container_exists("test_llm") is True


def test_container_exists_false():
    with patch("llamacppctl.docker_ops._run") as mock_run:
        mock_run.return_value = MagicMock(stdout="other\n")
        assert container_exists("test_llm") is False


def test_container_running_true():
    with patch("llamacppctl.docker_ops._run") as mock_run:
        mock_run.return_value = MagicMock(stdout="test_llm\n")
        assert container_running("test_llm") is True


def test_inspect_container_missing():
    with patch("llamacppctl.docker_ops.container_exists", return_value=False):
        info = inspect_container("ghost")
        assert info.status == "missing"
        assert info.health == "none"


def test_inspect_container_running():
    fake_json = (
        '[{"State": {"Status": "running", "Health": {"Status": "healthy"}}}]'
    )
    with patch("llamacppctl.docker_ops.container_exists", return_value=True), patch(
        "llamacppctl.docker_ops._run"
    ) as mock_run:
        mock_run.return_value = MagicMock(stdout=fake_json)
        info = inspect_container("test_llm")
        assert info.status == "running"
        assert info.health == "healthy"


def test_build_run_command_contains_key_flags():
    cfg = make_cfg()
    cmd = build_run_command(cfg)
    assert cmd[0:2] == ["docker", "run"]
    assert "--gpus" in cmd
    assert "device=0" in cmd
    assert "--name" in cmd
    assert "test_llm" in cmd
    assert "-p" in cmd
    # loopback-only by default (expose=False)
    assert "127.0.0.1:8001:8000" in cmd
    assert "--jinja" in cmd
    assert "--reasoning" in cmd
    assert "on" in cmd
    assert "--no-context-shift" in cmd
    assert "--kv-unified" in cmd
    assert "--cont-batching" in cmd


def test_build_run_command_overrides_healthcheck_to_configured_port():
    cfg = make_cfg()
    cmd = build_run_command(cfg)
    assert "--health-cmd" in cmd
    # healthcheck must target the configured container_port (8000), not the
    # image default (8080), and hit the configured health endpoint.
    assert "curl -f http://localhost:8000/health || exit 1" in cmd
    assert "--health-start-period" in cmd


def test_build_run_command_expose_binds_all_interfaces():
    cfg = make_cfg(expose=True)
    cmd = build_run_command(cfg)
    assert "8001:8000" in cmd
    assert "127.0.0.1:8001:8000" not in cmd


def test_build_run_command_api_key_passed_to_server():
    cfg = make_cfg(api_key="secret123")
    cmd = build_run_command(cfg)
    assert "--api-key" in cmd
    assert "secret123" in cmd


def test_build_run_command_no_api_key_by_default():
    cfg = make_cfg()
    assert "--api-key" not in build_run_command(cfg)


def test_build_run_command_respects_disabled_flags():
    cfg = make_cfg(jinja=False, kv_unified=False, cont_batching=False, no_context_shift=False, fa=False)
    cmd = build_run_command(cfg)
    assert "--jinja" not in cmd
    assert "--kv-unified" not in cmd
    assert "--cont-batching" not in cmd
    assert "--no-context-shift" not in cmd


def test_build_run_command_absolute_model_path():
    cfg = make_cfg(model_path="/abs/path/model.gguf")
    cmd = build_run_command(cfg)
    idx = cmd.index("-m")
    assert cmd[idx + 1] == "/abs/path/model.gguf"


def test_build_run_command_relative_model_path():
    cfg = make_cfg(model_path="qwen3/default.gguf")
    cmd = build_run_command(cfg)
    idx = cmd.index("-m")
    assert cmd[idx + 1] == "/hf_home/qwen3/default.gguf"


def test_format_command_for_display_quotes_properly():
    cmd = ["docker", "run", "--name", "has space"]
    out = format_command_for_display(cmd)
    assert "'has space'" in out