from pathlib import Path import pytest from gpu_rent.llm_runtime import ( already_have_ollama_tag, decide_runtime, normalize_runtime, parse_ollama_models, preferred_ollama_model, write_ollama_models_preset, ) def test_normalize_runtime(): assert normalize_runtime(None) == "none" assert normalize_runtime("OLLAMA") == "ollama" with pytest.raises(ValueError): normalize_runtime("llamacpp") with pytest.raises(ValueError): normalize_runtime("foo") def test_decide_runtime_flags_win(): assert ( decide_runtime(flag=None, ollama_flag=True, from_config="none") == "ollama" ) assert ( decide_runtime(flag="ollama", ollama_flag=False, from_config="none") == "ollama" ) assert decide_runtime(flag=None, ollama_flag=False, from_config="ollama") == "ollama" def test_parse_ollama_models(tmp_path: Path): path = tmp_path / "m.yaml" path.write_text( "models:\n - name: huihui_ai/qwen2.5-abliterate:7b\n default: true\n - qwen2.5:3b\n", encoding="utf-8", ) entries = parse_ollama_models(path) assert [e.name for e in entries] == [ "huihui_ai/qwen2.5-abliterate:7b", "qwen2.5:3b", ] assert entries[0].default is True def test_parse_empty_manifest(tmp_path: Path): path = tmp_path / "empty.yaml" path.write_text("models: []\n", encoding="utf-8") assert parse_ollama_models(path) == [] assert parse_ollama_models(tmp_path / "missing.yaml") == [] def test_write_preset(tmp_path: Path): path = tmp_path / "out.yaml" write_ollama_models_preset(path, "recommended") entries = parse_ollama_models(path) assert entries[0].name == "huihui_ai/qwen3-vl-abliterated:8b-instruct" assert entries[0].default is True assert entries[1].name == "huihui_ai/qwen2.5-vl-abliterated:7b" write_ollama_models_preset(path, "big") assert parse_ollama_models(path)[0].name == "huihui_ai/qwen2.5-vl-abliterated:32b" write_ollama_models_preset(path, "text") assert parse_ollama_models(path)[0].name == "huihui_ai/qwen2.5-abliterate:7b" def test_already_have_ollama_tag_exact_only(): have = {"qwen2.5:7b", "foo:latest"} assert already_have_ollama_tag(have, "qwen2.5:7b") assert not already_have_ollama_tag(have, "qwen2.5:3b") assert already_have_ollama_tag(have, "foo") assert already_have_ollama_tag(have, "foo:latest") def test_preferred_ollama_model(tmp_path: Path): path = tmp_path / "m.yaml" path.write_text( "models:\n - name: a:3b\n - name: b:7b\n default: true\n", encoding="utf-8", ) assert preferred_ollama_model(path) == "b:7b" path.write_text('models:\n - "only:7b"\n', encoding="utf-8") assert preferred_ollama_model(path) == "only:7b" assert preferred_ollama_model(tmp_path / "missing.yaml") is None def test_warmup_ollama_http_skips_loaded(monkeypatch): from gpu_rent.llm_runtime import warmup_ollama_http class Resp: def raise_for_status(self) -> None: return None def json(self): return {"models": [{"name": "foo:7b"}]} class Client: def __init__(self, *a, **k): pass def __enter__(self): return self def __exit__(self, *a): return False def get(self, url, timeout=None): assert url.endswith("/api/ps") return Resp() def post(self, *_a, **_k): raise AssertionError("must not chat when already loaded") monkeypatch.setattr("gpu_rent.llm_runtime.httpx.Client", Client) msg = warmup_ollama_http("http://127.0.0.1:17811", "foo:7b") assert "skip" in msg def test_maybe_warmup_skips_when_runtime_none(): from gpu_rent.llm_runtime import maybe_warmup_ollama_local logs: list[str] = [] class Cfg: llm_runtime = "none" ollama_models_manifest = Path("missing.yaml") ollama_local_port = 17811 maybe_warmup_ollama_local(Cfg(), logs.append) assert logs == []