110 lines
3.3 KiB
Python
110 lines
3.3 KiB
Python
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).parent.parent.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from tools import lea_micro_preflight as preflight
|
|
|
|
|
|
FREE_OUTPUT = """\
|
|
total used free shared buff/cache available
|
|
Mem: 64202 15500 32000 123 16702 47000
|
|
Swap: 8192 1024 7168
|
|
"""
|
|
|
|
|
|
def test_parse_free_m_extracts_ram_and_swap():
|
|
parsed = preflight.parse_free_m(FREE_OUTPUT)
|
|
|
|
assert parsed["mem"]["total"] == 64202
|
|
assert parsed["mem"]["available"] == 47000
|
|
assert parsed["swap"] == {"total": 8192, "used": 1024, "free": 7168}
|
|
|
|
|
|
def test_parse_free_m_accepts_french_locale_labels():
|
|
output = """\
|
|
total utilisé libre partagé tamp/cache disponible
|
|
Mem: 126365 60425 2919 12847 77071 65939
|
|
Échange: 8191 3397 4794
|
|
"""
|
|
|
|
parsed = preflight.parse_free_m(output)
|
|
|
|
assert parsed["mem"]["used"] == 60425
|
|
assert parsed["mem"]["available"] == 65939
|
|
assert parsed["swap"] == {"total": 8191, "used": 3397, "free": 4794}
|
|
|
|
|
|
def test_parse_nvidia_smi_memory_multiple_gpus():
|
|
parsed = preflight.parse_nvidia_smi_memory("8123, 24576\n3999 MiB, 12288 MiB\n")
|
|
|
|
assert parsed == [
|
|
{"free_mib": 8123, "total_mib": 24576},
|
|
{"free_mib": 3999, "total_mib": 12288},
|
|
]
|
|
|
|
|
|
def test_extract_ollama_tags_accepts_name_and_model_keys():
|
|
tags = preflight.extract_ollama_tags(
|
|
{
|
|
"models": [
|
|
{"name": "qwen2.5vl:7b-rpa"},
|
|
{"model": "qwen2.5:7b"},
|
|
{"name": ""},
|
|
"ignored",
|
|
]
|
|
}
|
|
)
|
|
|
|
assert tags == {"qwen2.5vl:7b-rpa", "qwen2.5:7b"}
|
|
|
|
|
|
def _install_fakes(monkeypatch, *, resident=True, tags_ok=True, swap_used=1024):
|
|
free_output = FREE_OUTPUT.replace("1024", str(swap_used), 1)
|
|
|
|
def fake_run_command(args, timeout=5.0):
|
|
if args[0] == "nvidia-smi":
|
|
return 0, "8123, 24576", ""
|
|
if args[0] == "free":
|
|
return 0, free_output, ""
|
|
raise AssertionError(f"unexpected command: {args!r}")
|
|
|
|
def fake_http_json(url, timeout=2.0):
|
|
if url.endswith("/api/tags"):
|
|
models = [{"name": "qwen2.5vl:7b-rpa"}]
|
|
if tags_ok:
|
|
models.append({"name": "qwen2.5:7b"})
|
|
return True, {"models": models}, ""
|
|
if url.endswith("/api/ps"):
|
|
models = [{"name": "qwen2.5vl:7b-rpa"}] if resident else []
|
|
return True, {"models": models}, ""
|
|
raise AssertionError(f"unexpected url: {url!r}")
|
|
|
|
monkeypatch.setattr(preflight, "run_command", fake_run_command)
|
|
monkeypatch.setattr(preflight, "http_json", fake_http_json)
|
|
|
|
|
|
def test_main_returns_zero_when_all_checks_ok(monkeypatch, capsys):
|
|
_install_fakes(monkeypatch)
|
|
|
|
assert preflight.main(["--json"]) == 0
|
|
report = json.loads(capsys.readouterr().out)
|
|
assert report["overall"] == "ok"
|
|
assert report["warmup"] == "disabled"
|
|
|
|
|
|
def test_main_warns_when_vlm_not_resident_and_strict_exits_one(monkeypatch):
|
|
_install_fakes(monkeypatch, resident=False)
|
|
|
|
assert preflight.main([]) == 0
|
|
assert preflight.main(["--strict"]) == 1
|
|
|
|
|
|
def test_main_fails_when_required_model_missing(monkeypatch):
|
|
_install_fakes(monkeypatch, tags_ok=False)
|
|
|
|
assert preflight.main([]) == 2
|