86 lines
3.0 KiB
Python
86 lines
3.0 KiB
Python
from app.modules.system import service
|
|
from app.modules.system.service import get_runtime_readiness, parse_nvidia_smi_csv, parse_probe_stdout
|
|
|
|
|
|
def test_parse_nvidia_smi_csv():
|
|
output = "0, NVIDIA GeForce RTX 4090, 24564, 15, 24069, 0, 37\n"
|
|
gpus = parse_nvidia_smi_csv(output)
|
|
assert gpus == [
|
|
{
|
|
"index": 0,
|
|
"name": "NVIDIA GeForce RTX 4090",
|
|
"memory_total_mb": 24564,
|
|
"memory_used_mb": 15,
|
|
"memory_free_mb": 24069,
|
|
"utilization_gpu_percent": 0,
|
|
"temperature_c": 37,
|
|
}
|
|
]
|
|
|
|
|
|
def test_parse_probe_stdout_uses_last_json_line():
|
|
parsed = parse_probe_stdout(
|
|
"warning before json\n"
|
|
'{"checks":[{"module":"torch","passed":true,"version":"2.6.0"}],"extra":{"python":"3.11"}}\n'
|
|
)
|
|
|
|
assert parsed["checks"][0]["module"] == "torch"
|
|
assert parsed["extra"]["python"] == "3.11"
|
|
|
|
|
|
def test_runtime_readiness_marks_missing_env(monkeypatch):
|
|
monkeypatch.setattr(service, "_readiness_cache", None)
|
|
monkeypatch.setattr(service, "get_conda_envs", lambda: {"available": True, "envs": []})
|
|
|
|
readiness = get_runtime_readiness(force=True)
|
|
|
|
assert readiness["passed"] is False
|
|
assert all(not item["exists"] for item in readiness["envs"])
|
|
|
|
|
|
def test_runtime_readiness_aggregates_probe_results(monkeypatch):
|
|
monkeypatch.setattr(service, "_readiness_cache", None)
|
|
specs = [
|
|
{"role": "task", "name": "seg_smp", "label": "task", "env_file": "envs/seg_smp.yml", "required_imports": []},
|
|
{"role": "mmseg", "name": "seg_mmcv", "label": "mmseg", "env_file": "envs/seg_mmcv.yml", "required_imports": []},
|
|
]
|
|
monkeypatch.setattr(service, "runtime_environment_specs", lambda: specs)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"get_conda_envs",
|
|
lambda: {
|
|
"available": True,
|
|
"envs": [
|
|
{"name": "seg_smp", "path": "/envs/seg_smp"},
|
|
{"name": "seg_mmcv", "path": "/envs/seg_mmcv"},
|
|
],
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"inspect_conda_env",
|
|
lambda name, imports: {
|
|
"passed": True,
|
|
"checks": [{"module": "torch", "passed": True}],
|
|
"extra": {"python": "3.11"},
|
|
},
|
|
)
|
|
|
|
readiness = get_runtime_readiness(force=True)
|
|
|
|
assert readiness["passed"] is True
|
|
assert readiness["envs"][0]["path"] == "/envs/seg_smp"
|
|
assert readiness["specs"]["env_files"] == ["envs/seg_smp.yml", "envs/seg_mmcv.yml"]
|
|
|
|
|
|
def test_runtime_readiness_refreshes_stale_failure_cache(monkeypatch):
|
|
monkeypatch.setattr(service, "_readiness_cache", (100.0, {"passed": False, "envs": [], "available": True}))
|
|
monkeypatch.setattr(service.time, "time", lambda: 100.0 + service.READINESS_FAILURE_CACHE_SECONDS + 1)
|
|
monkeypatch.setattr(service, "runtime_environment_specs", lambda: [])
|
|
monkeypatch.setattr(service, "get_conda_envs", lambda: {"available": True, "envs": []})
|
|
|
|
readiness = get_runtime_readiness(force=False)
|
|
|
|
assert readiness["cached"] is False
|
|
assert readiness["passed"] is True
|