189 lines
6.7 KiB
Python
189 lines
6.7 KiB
Python
"""Tests for the ensure_ollama orchestration (Ollama mocked via injected ops)."""
|
|
import json as _json
|
|
from types import SimpleNamespace
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli import ollama_cmd
|
|
from tht.cli.ollama_cmd import ensure_ollama, ollama_app
|
|
from tht.config import EmbeddingsConfig
|
|
|
|
|
|
def _cfg(**kw):
|
|
kw.setdefault("model", "nomic-embed-text-v2-moe")
|
|
kw.setdefault("dim", 768)
|
|
emb = EmbeddingsConfig(base_url="http://localhost:11434", **kw)
|
|
return SimpleNamespace(embeddings=emb)
|
|
|
|
|
|
def test_no_embeddings_config_is_hard_error():
|
|
r = ensure_ollama(SimpleNamespace(embeddings=None), timeout=5, no_start=False)
|
|
assert r["ok"] is False and r["stage"] == "config"
|
|
|
|
|
|
def test_server_up_model_present_warms_ok():
|
|
warmed = []
|
|
r = ensure_ollama(
|
|
_cfg(model="nomic-embed-text-v2-moe"), timeout=5, no_start=False,
|
|
probe=lambda url: True,
|
|
installed_models=lambda url: {"nomic-embed-text-v2-moe:latest"},
|
|
start=lambda cmd: (_ for _ in ()).throw(AssertionError("must not start")),
|
|
warm=lambda cfg: warmed.append(True),
|
|
)
|
|
assert r == {"ok": True, "server": "up", "model": "warmed", "model_name": "nomic-embed-text-v2-moe"}
|
|
assert warmed == [True]
|
|
|
|
|
|
def test_server_down_then_started_after_poll():
|
|
started = []
|
|
probes = iter([False, True]) # down, then up after start
|
|
r = ensure_ollama(
|
|
_cfg(), timeout=5, no_start=False,
|
|
probe=lambda url: next(probes),
|
|
installed_models=lambda url: {"nomic-embed-text-v2-moe"},
|
|
start=lambda cmd: started.append(cmd),
|
|
warm=lambda cfg: None,
|
|
sleep=lambda s: None,
|
|
)
|
|
assert r["ok"] is True and r["server"] == "started"
|
|
assert started and started[0] == ["ollama", "serve"]
|
|
|
|
|
|
def test_server_unreachable_after_timeout_is_error():
|
|
clk = iter([0.0, 1.0, 2.0, 99.0]) # monotonic crosses the deadline
|
|
r = ensure_ollama(
|
|
_cfg(), timeout=5, no_start=False,
|
|
probe=lambda url: False, # never comes up
|
|
installed_models=lambda url: set(),
|
|
start=lambda cmd: None,
|
|
warm=lambda cfg: None,
|
|
sleep=lambda s: None,
|
|
clock=lambda: next(clk),
|
|
)
|
|
assert r["ok"] is False and r["stage"] == "server"
|
|
|
|
|
|
def test_no_start_and_down_is_error_without_starting():
|
|
r = ensure_ollama(
|
|
_cfg(), timeout=5, no_start=True,
|
|
probe=lambda url: False,
|
|
installed_models=lambda url: set(),
|
|
start=lambda cmd: (_ for _ in ()).throw(AssertionError("must not start")),
|
|
warm=lambda cfg: None,
|
|
)
|
|
assert r["ok"] is False and r["stage"] == "server"
|
|
|
|
|
|
def test_empty_start_cmd_disables_autostart():
|
|
r = ensure_ollama(
|
|
_cfg(start_cmd=[]), timeout=5, no_start=False,
|
|
probe=lambda url: False,
|
|
installed_models=lambda url: set(),
|
|
start=lambda cmd: (_ for _ in ()).throw(AssertionError("must not start")),
|
|
warm=lambda cfg: None,
|
|
)
|
|
assert r["ok"] is False and r["stage"] == "server"
|
|
|
|
|
|
def test_model_absent_is_error_with_pull_guidance():
|
|
r = ensure_ollama(
|
|
_cfg(model="missing-model"), timeout=5, no_start=False,
|
|
probe=lambda url: True,
|
|
installed_models=lambda url: {"nomic-embed-text-v2-moe"},
|
|
start=lambda cmd: None,
|
|
warm=lambda cfg: None,
|
|
)
|
|
assert r["ok"] is False and r["stage"] == "model"
|
|
assert "ollama pull missing-model" in r["error"]
|
|
|
|
|
|
def test_warm_failure_is_error():
|
|
r = ensure_ollama(
|
|
_cfg(), timeout=5, no_start=False,
|
|
probe=lambda url: True,
|
|
installed_models=lambda url: {"nomic-embed-text-v2-moe"},
|
|
start=lambda cmd: None,
|
|
warm=lambda cfg: (_ for _ in ()).throw(RuntimeError("boom")),
|
|
)
|
|
assert r["ok"] is False and r["stage"] == "warm"
|
|
|
|
|
|
def test_custom_start_cmd_used():
|
|
started = []
|
|
probes = iter([False, True])
|
|
ensure_ollama(
|
|
_cfg(bin="ollama", start_cmd=["docker", "start", "ollama"]), timeout=5, no_start=False,
|
|
probe=lambda url: next(probes),
|
|
installed_models=lambda url: {"nomic-embed-text-v2-moe"},
|
|
start=lambda cmd: started.append(cmd),
|
|
warm=lambda cfg: None,
|
|
sleep=lambda s: None,
|
|
)
|
|
assert started[0] == ["docker", "start", "ollama"]
|
|
|
|
|
|
def test_start_failure_is_server_error():
|
|
r = ensure_ollama(
|
|
_cfg(), timeout=5, no_start=False,
|
|
probe=lambda url: False,
|
|
installed_models=lambda url: set(),
|
|
start=lambda cmd: (_ for _ in ()).throw(FileNotFoundError("ollama: not found")),
|
|
warm=lambda cfg: None,
|
|
sleep=lambda s: None,
|
|
)
|
|
assert r["ok"] is False and r["stage"] == "server"
|
|
assert "avvio" in r["error"].lower() or "not found" in r["error"].lower()
|
|
|
|
|
|
def test_initial_probe_exception_is_server_error():
|
|
r = ensure_ollama(
|
|
_cfg(), timeout=5, no_start=False,
|
|
probe=lambda url: (_ for _ in ()).throw(RuntimeError("boom")),
|
|
installed_models=lambda url: set(),
|
|
start=lambda cmd: None,
|
|
warm=lambda cfg: None,
|
|
)
|
|
assert r["ok"] is False and r["stage"] == "server"
|
|
|
|
|
|
def _patch(monkeypatch, result):
|
|
monkeypatch.setattr(ollama_cmd, "_load_config_or_exit", lambda _c: SimpleNamespace(embeddings=object()))
|
|
monkeypatch.setattr(ollama_cmd, "ensure_ollama", lambda cfg, **kw: result)
|
|
|
|
|
|
def test_cli_ok_exit_zero_and_json_pristine(monkeypatch):
|
|
_patch(monkeypatch, {"ok": True, "server": "up", "model": "warmed", "model_name": "m"})
|
|
res = CliRunner().invoke(ollama_app, ["--json"])
|
|
assert res.exit_code == 0, res.output
|
|
assert _json.loads(res.output) == {"ok": True, "server": "up", "model": "warmed", "model_name": "m"}
|
|
|
|
|
|
def test_cli_error_exit_one_and_json_on_stdout(monkeypatch):
|
|
_patch(monkeypatch, {"ok": False, "stage": "model", "error": "missing"})
|
|
res = CliRunner().invoke(ollama_app, ["--json"])
|
|
assert res.exit_code == 1
|
|
assert _json.loads(res.output) == {"ok": False, "stage": "model", "error": "missing"}
|
|
|
|
|
|
def test_cli_error_human_mode_exit_one(monkeypatch):
|
|
_patch(monkeypatch, {"ok": False, "stage": "server", "error": "down"})
|
|
res = CliRunner().invoke(ollama_app, [])
|
|
assert res.exit_code == 1
|
|
|
|
|
|
def test_cli_registered_on_root_app():
|
|
from tht.cli import app # the root Typer app
|
|
runner = CliRunner()
|
|
res = runner.invoke(app, ["ollama", "--help"])
|
|
assert res.exit_code == 0
|
|
assert "ensure" in res.output
|
|
|
|
|
|
def test_cli_root_app_ensure_end_to_end(monkeypatch):
|
|
from tht.cli import app # root tht app
|
|
|
|
_patch(monkeypatch, {"ok": True, "server": "up", "model": "warmed", "model_name": "m"})
|
|
res = CliRunner().invoke(app, ["ollama", "ensure", "--json"])
|
|
assert res.exit_code == 0, res.output
|
|
assert _json.loads(res.output) == {"ok": True, "server": "up", "model": "warmed", "model_name": "m"}
|