141 lines
4.8 KiB
Python
141 lines
4.8 KiB
Python
"""Tests for the ensure_ollama orchestration (Ollama mocked via injected ops)."""
|
|
from types import SimpleNamespace
|
|
|
|
from tht.config import EmbeddingsConfig
|
|
from tht.cli.ollama_cmd import ensure_ollama
|
|
|
|
|
|
def _cfg(**kw):
|
|
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"
|