dschlueter
218c3fc791
Vision-capable GGUFs need a separate projector (mmproj) that maps image embeddings into the text model's space. Add `mmproj` and `mmproj_offload` as config keys and CLI overrides, and pass them through to llama-server. The projector path resolves under hf_home exactly like model_path, so it is covered by the existing read-only mount. validate_model_path() now also checks the projector, which means --change rejects a missing one *before* it removes the running container. --no-mmproj-offload is suppressed when no projector is configured, since llama.cpp rejects the flag on its own. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
246 lines
9.4 KiB
Python
246 lines
9.4 KiB
Python
import sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
|
|
|
import pytest # noqa: E402
|
|
|
|
from llamacppctl import actions, http_ops # noqa: E402
|
|
from llamacppctl.docker_ops import ContainerInfo # noqa: E402
|
|
from llamacppctl.http_ops import HttpError # noqa: E402
|
|
from llamacppctl.lock_ops import FileLock, LockError # noqa: E402
|
|
from llamacppctl.schema import ChatReply, CheckResult # noqa: E402
|
|
from tests.test_docker_ops import make_cfg # noqa: E402
|
|
|
|
|
|
def _cfg(tmp_path):
|
|
return make_cfg(lock_file=tmp_path / "test.lock")
|
|
|
|
|
|
# --- dry-run must be a pure preview (no side effects) ---------------------
|
|
|
|
|
|
def test_do_start_dry_run_does_not_start(tmp_path, monkeypatch, capsys):
|
|
cfg = _cfg(tmp_path)
|
|
called = {"start": 0}
|
|
monkeypatch.setattr(actions, "start_llama_container", lambda c: called.__setitem__("start", 1))
|
|
rc = actions.do_start(cfg, SimpleNamespace(), SimpleNamespace(dry_run=True))
|
|
assert rc == actions.EXIT_OK
|
|
assert called["start"] == 0
|
|
assert "docker run" in capsys.readouterr().out
|
|
|
|
|
|
def test_do_change_dry_run_does_not_remove(tmp_path, monkeypatch):
|
|
cfg = _cfg(tmp_path)
|
|
called = {"remove": 0}
|
|
monkeypatch.setattr(actions, "remove_container", lambda n: called.__setitem__("remove", 1))
|
|
rc = actions.do_change(cfg, SimpleNamespace(), SimpleNamespace(dry_run=True))
|
|
assert rc == actions.EXIT_OK
|
|
assert called["remove"] == 0
|
|
|
|
|
|
# --- --change must validate BEFORE removing the running container ----------
|
|
|
|
|
|
def test_do_change_validates_before_remove(tmp_path, monkeypatch):
|
|
cfg = _cfg(tmp_path)
|
|
called = {"remove": 0, "start": 0}
|
|
monkeypatch.setattr(actions, "remove_container", lambda n: called.__setitem__("remove", 1))
|
|
monkeypatch.setattr(actions, "start_llama_container", lambda c: called.__setitem__("start", 1))
|
|
|
|
def missing(_cfg):
|
|
raise FileNotFoundError("model file not found: /x")
|
|
|
|
monkeypatch.setattr(actions, "validate_model_path", missing)
|
|
rc = actions.do_change(cfg, SimpleNamespace(), SimpleNamespace(dry_run=False, force=False))
|
|
assert rc == actions.EXIT_MODEL_PATH_MISSING
|
|
assert called["remove"] == 0 # container was NOT touched
|
|
assert called["start"] == 0
|
|
|
|
|
|
# --- do_chat -------------------------------------------------------------
|
|
|
|
|
|
def test_do_chat_prints_content(tmp_path, monkeypatch, capsys):
|
|
cfg = _cfg(tmp_path)
|
|
monkeypatch.setattr(actions, "chat_completion_text", lambda c, p: ChatReply("Hallo", "stop"))
|
|
prompt = SimpleNamespace(user_prompt="hi", max_tokens=2048)
|
|
rc = actions.do_chat(cfg, prompt, SimpleNamespace(stream=False))
|
|
assert rc == actions.EXIT_OK
|
|
assert capsys.readouterr().out.strip() == "Hallo"
|
|
|
|
|
|
def test_do_chat_truncated_empty_warns_and_fails(tmp_path, monkeypatch, capsys):
|
|
cfg = _cfg(tmp_path)
|
|
monkeypatch.setattr(actions, "chat_completion_text", lambda c, p: ChatReply("", "length"))
|
|
prompt = SimpleNamespace(user_prompt="hi", max_tokens=128)
|
|
rc = actions.do_chat(cfg, prompt, SimpleNamespace(stream=False))
|
|
assert rc == actions.EXIT_GENERAL
|
|
err = capsys.readouterr().err
|
|
assert "max_tokens=128" in err and "abgeschnitten" in err
|
|
|
|
|
|
def test_do_chat_http_error_returns_not_ready(tmp_path, monkeypatch):
|
|
cfg = _cfg(tmp_path)
|
|
|
|
def boom(c, p):
|
|
raise HttpError("down")
|
|
|
|
monkeypatch.setattr(actions, "chat_completion_text", boom)
|
|
prompt = SimpleNamespace(user_prompt="hi", max_tokens=2048)
|
|
rc = actions.do_chat(cfg, prompt, SimpleNamespace(stream=False))
|
|
assert rc == actions.EXIT_HTTP_NOT_READY
|
|
|
|
|
|
def test_do_chat_stream(tmp_path, monkeypatch, capsys):
|
|
cfg = _cfg(tmp_path)
|
|
monkeypatch.setattr(
|
|
actions,
|
|
"stream_chat",
|
|
lambda c, p: iter([("content", "Hal"), ("content", "lo"), ("finish", "stop")]),
|
|
)
|
|
prompt = SimpleNamespace(user_prompt="hi", max_tokens=2048)
|
|
rc = actions.do_chat(cfg, prompt, SimpleNamespace(stream=True))
|
|
assert rc == actions.EXIT_OK
|
|
assert capsys.readouterr().out.strip() == "Hallo"
|
|
|
|
|
|
# --- exit-code mapping & force stop --------------------------------------
|
|
|
|
|
|
def test_check_exit_code():
|
|
ok = CheckResult(True, True, True, True, True, "u", "")
|
|
down = CheckResult(True, False, False, False, False, "u", "")
|
|
missing = CheckResult(False, False, False, False, False, "u", "")
|
|
assert actions.check_exit_code(ok) == actions.EXIT_OK
|
|
assert actions.check_exit_code(down) == actions.EXIT_HTTP_NOT_READY
|
|
assert actions.check_exit_code(missing) == actions.EXIT_HTTP_NOT_READY
|
|
|
|
|
|
def test_do_stop_passes_force(tmp_path, monkeypatch):
|
|
cfg = _cfg(tmp_path)
|
|
seen = {}
|
|
|
|
def fake_stop(name, remove=True, force=False):
|
|
seen["force"] = force
|
|
return True
|
|
|
|
monkeypatch.setattr(actions, "stop_container", fake_stop)
|
|
actions.do_stop(cfg, SimpleNamespace(force=True))
|
|
assert seen["force"] is True
|
|
|
|
|
|
# --- do_check ------------------------------------------------------------
|
|
|
|
|
|
def test_do_check_running_and_healthy(tmp_path, monkeypatch):
|
|
cfg = _cfg(tmp_path)
|
|
monkeypatch.setattr(
|
|
actions, "inspect_container", lambda n: ContainerInfo(n, "running", "healthy")
|
|
)
|
|
monkeypatch.setattr(http_ops, "check_health", lambda c: True)
|
|
monkeypatch.setattr(http_ops, "check_models", lambda c: True)
|
|
monkeypatch.setattr(actions, "chat_completion_text", lambda c, p: ChatReply("ok", "stop"))
|
|
result = actions.do_check(cfg, SimpleNamespace(user_prompt="ping", max_tokens=8), SimpleNamespace())
|
|
assert result.container_running is True
|
|
assert result.http_ok is True
|
|
assert result.chat_ok is True
|
|
assert actions.check_exit_code(result) == actions.EXIT_OK
|
|
|
|
|
|
def test_do_check_missing_container(tmp_path, monkeypatch):
|
|
cfg = _cfg(tmp_path)
|
|
monkeypatch.setattr(
|
|
actions, "inspect_container", lambda n: ContainerInfo(n, "missing", "none")
|
|
)
|
|
result = actions.do_check(cfg, SimpleNamespace(user_prompt=None, max_tokens=8), SimpleNamespace())
|
|
assert result.container_exists is False
|
|
assert result.container_running is False
|
|
assert result.http_ok is False
|
|
assert actions.check_exit_code(result) == actions.EXIT_HTTP_NOT_READY
|
|
|
|
|
|
# --- do_start (non-dry-run) ----------------------------------------------
|
|
|
|
|
|
def _wire_start(monkeypatch, ready: bool):
|
|
monkeypatch.setattr(actions, "validate_model_path", lambda c: None)
|
|
monkeypatch.setattr(actions, "start_llama_container", lambda c: "abc123def456")
|
|
monkeypatch.setattr(actions, "wait_until_ready", lambda c, p: ready)
|
|
monkeypatch.setattr(actions, "container_logs", lambda n, t=100: "log tail")
|
|
|
|
|
|
def test_do_start_happy_path(tmp_path, monkeypatch, capsys):
|
|
cfg = _cfg(tmp_path)
|
|
_wire_start(monkeypatch, ready=True)
|
|
monkeypatch.setattr(actions, "chat_completion_text", lambda c, p: ChatReply("Servus", "stop"))
|
|
prompt = SimpleNamespace(user_prompt="hi", max_tokens=2048)
|
|
rc = actions.do_start(cfg, prompt, SimpleNamespace(dry_run=False, force=False, logs=False))
|
|
out = capsys.readouterr().out
|
|
assert rc == actions.EXIT_OK
|
|
assert "Model ready." in out
|
|
assert "Servus" in out
|
|
|
|
|
|
def test_do_start_readiness_failure(tmp_path, monkeypatch, capsys):
|
|
cfg = _cfg(tmp_path)
|
|
_wire_start(monkeypatch, ready=False)
|
|
prompt = SimpleNamespace(user_prompt=None, max_tokens=2048)
|
|
rc = actions.do_start(cfg, prompt, SimpleNamespace(dry_run=False, force=False, logs=True, log_lines=50))
|
|
err = capsys.readouterr().err
|
|
assert rc == actions.EXIT_HTTP_NOT_READY
|
|
assert "did not become ready" in err
|
|
assert "log tail" in err # --logs printed the tail
|
|
|
|
|
|
# --- _container_lock force bypass ----------------------------------------
|
|
|
|
|
|
def test_container_lock_busy_raises_without_force(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
with FileLock(cfg.lock_file): # hold the lock
|
|
with pytest.raises(LockError):
|
|
with actions._container_lock(cfg, SimpleNamespace(force=False)):
|
|
pass
|
|
|
|
|
|
def test_container_lock_force_bypasses_busy(tmp_path, capsys):
|
|
cfg = _cfg(tmp_path)
|
|
entered = False
|
|
with FileLock(cfg.lock_file): # hold the lock
|
|
with actions._container_lock(cfg, SimpleNamespace(force=True)):
|
|
entered = True
|
|
assert entered is True
|
|
assert "busy lock" in capsys.readouterr().err
|
|
|
|
|
|
# --- model / mmproj path validation ---------------------------------------
|
|
|
|
|
|
def test_validate_model_path_accepts_existing_model(tmp_path):
|
|
(tmp_path / "m.gguf").write_bytes(b"x")
|
|
cfg = make_cfg(hf_home=tmp_path, model_path="m.gguf")
|
|
actions.validate_model_path(cfg) # must not raise
|
|
|
|
|
|
def test_validate_model_path_rejects_missing_model(tmp_path):
|
|
cfg = make_cfg(hf_home=tmp_path, model_path="absent.gguf")
|
|
with pytest.raises(FileNotFoundError, match="model file not found"):
|
|
actions.validate_model_path(cfg)
|
|
|
|
|
|
def test_validate_model_path_rejects_missing_mmproj(tmp_path):
|
|
# The model exists but the configured projector does not: --change must fail
|
|
# here, before the running container is removed.
|
|
(tmp_path / "m.gguf").write_bytes(b"x")
|
|
cfg = make_cfg(hf_home=tmp_path, model_path="m.gguf", mmproj="absent-proj.gguf")
|
|
with pytest.raises(FileNotFoundError, match="mmproj file not found"):
|
|
actions.validate_model_path(cfg)
|
|
|
|
|
|
def test_validate_model_path_accepts_existing_mmproj(tmp_path):
|
|
(tmp_path / "m.gguf").write_bytes(b"x")
|
|
(tmp_path / "proj.gguf").write_bytes(b"x")
|
|
cfg = make_cfg(hf_home=tmp_path, model_path="m.gguf", mmproj="proj.gguf")
|
|
actions.validate_model_path(cfg) # must not raise
|