test(tasks): add preflight fallback coverage to scheduler tests
Adds test_detect_vram_preflight_fallback to cover the spec path where cf-orch is unreachable but scripts.preflight.get_gpus() succeeds, verifying detect_available_vram_gb() returns the summed total VRAM. Uses sys.modules injection to simulate the preflight module being present.
This commit is contained in:
parent
5801928f8e
commit
09a5087c72
1 changed files with 23 additions and 0 deletions
|
|
@ -5,6 +5,7 @@ import sqlite3
|
||||||
import threading
|
import threading
|
||||||
import time
|
import time
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from types import ModuleType
|
||||||
from typing import List
|
from typing import List
|
||||||
from unittest.mock import MagicMock, patch
|
from unittest.mock import MagicMock, patch
|
||||||
|
|
||||||
|
|
@ -88,6 +89,28 @@ def test_detect_vram_cforch_empty_nodes_falls_back():
|
||||||
assert result == 999.0
|
assert result == 999.0
|
||||||
|
|
||||||
|
|
||||||
|
def test_detect_vram_preflight_fallback():
|
||||||
|
"""Falls back to preflight total VRAM when cf-orch is unreachable."""
|
||||||
|
# Build a fake scripts.preflight module with get_gpus returning two GPUs.
|
||||||
|
fake_scripts = ModuleType("scripts")
|
||||||
|
fake_preflight = ModuleType("scripts.preflight")
|
||||||
|
fake_preflight.get_gpus = lambda: [ # type: ignore[attr-defined]
|
||||||
|
{"vram_total_gb": 8.0},
|
||||||
|
{"vram_total_gb": 4.0},
|
||||||
|
]
|
||||||
|
fake_scripts.preflight = fake_preflight # type: ignore[attr-defined]
|
||||||
|
|
||||||
|
with patch("circuitforge_core.tasks.scheduler.httpx") as mock_httpx, \
|
||||||
|
patch.dict(
|
||||||
|
__import__("sys").modules,
|
||||||
|
{"scripts": fake_scripts, "scripts.preflight": fake_preflight},
|
||||||
|
):
|
||||||
|
mock_httpx.get.side_effect = ConnectionRefusedError()
|
||||||
|
result = detect_available_vram_gb()
|
||||||
|
|
||||||
|
assert result == pytest.approx(12.0) # 8.0 + 4.0 GB
|
||||||
|
|
||||||
|
|
||||||
# ── TaskScheduler basic behaviour ─────────────────────────────────────────────
|
# ── TaskScheduler basic behaviour ─────────────────────────────────────────────
|
||||||
|
|
||||||
def test_enqueue_returns_true_on_success(tmp_db: Path):
|
def test_enqueue_returns_true_on_success(tmp_db: Path):
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue