llamacppctl/tests/test_actions.py

import sys
from pathlib import Path
from types import SimpleNamespace

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

from llamacppctl import actions  # noqa: E402
from llamacppctl.http_ops import HttpError  # 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