fix(installer): survive a release without a checksum manifest, and stop inventing GPUs (#736)
* installer: leave model downloads to first run when the release pins no checksums * installer: refuse an unpinned VOSK model before spending the download * installer: require a real device before reporting a Vulkan GPU * build: replace the deprecated license table and complete the packages list * installer: treat software Vulkan renderers as no GPU, not as one * installer: keep diagnostics off stdout so captured detection output stays clean
This commit is contained in:
@@ -0,0 +1,326 @@
|
||||
"""Behavioural tests for the installer's Vulkan detection.
|
||||
|
||||
`detect_vulkan` used to accept any non-empty `vulkaninfo --summary` as a GPU, and
|
||||
read only the first 20 lines while `check_vulkan_gpu_compatibility` read all of
|
||||
them -- so the wizard could report a GPU in Step 1 and none in Step 3.
|
||||
"""
|
||||
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
INSTALL_SH = Path(__file__).resolve().parents[1] / "install.sh"
|
||||
SOURCE = INSTALL_SH.read_text()
|
||||
|
||||
FUNCTIONS = (
|
||||
"is_software_renderer",
|
||||
"vulkan_device_names",
|
||||
"detect_vulkan",
|
||||
"check_vulkan_gpu_compatibility",
|
||||
)
|
||||
|
||||
PRELUDE = """
|
||||
set -uo pipefail
|
||||
HAS_VULKAN="no"
|
||||
VULKAN_DEVICE=""
|
||||
VULKAN_SOFTWARE_DEVICE=""
|
||||
print_info() { echo "INFO: $*"; }
|
||||
print_warning() { echo "WARNING: $*"; }
|
||||
"""
|
||||
|
||||
# The one list both detectors read; lifted from install.sh so it cannot drift.
|
||||
PATTERNS = re.search(r"^VULKAN_SOFTWARE_PATTERNS=\([^)]*\)", SOURCE, re.M)
|
||||
assert PATTERNS, "VULKAN_SOFTWARE_PATTERNS missing from install.sh"
|
||||
|
||||
# A loader that resolved no driver: header, error, no device.
|
||||
NO_DEVICES = """\
|
||||
ERROR: [Loader Message] Code 0 : setup_loader_term_phys_devs: Failed to detect any valid GPUs
|
||||
==========
|
||||
VULKANINFO
|
||||
==========
|
||||
|
||||
Vulkan Instance Version: 1.3.239
|
||||
|
||||
Instance Extensions: count = 19
|
||||
Layers: count = 0
|
||||
|
||||
Devices:
|
||||
========
|
||||
"""
|
||||
|
||||
# A real adapter listed past the old `head -20` cutoff.
|
||||
DEVICE_PAST_LINE_20 = (
|
||||
"".join(f"instance preamble line {n}\n" for n in range(1, 25))
|
||||
+ "\tdeviceName = AMD Radeon RX 7900 XTX (RADV NAVI31)\n"
|
||||
)
|
||||
|
||||
SOFTWARE_ONLY = """\
|
||||
==========
|
||||
VULKANINFO
|
||||
==========
|
||||
|
||||
Devices:
|
||||
========
|
||||
GPU0:
|
||||
\tdeviceName = llvmpipe (LLVM 15.0.6, 256 bits)
|
||||
"""
|
||||
|
||||
# A real adapter the loader happens to list after the software one.
|
||||
SOFTWARE_THEN_HARDWARE = (
|
||||
SOFTWARE_ONLY + "\tdeviceName = AMD Radeon RX 7900 XTX (RADV NAVI31)\n"
|
||||
)
|
||||
|
||||
DEVICE_EARLY = """\
|
||||
==========
|
||||
VULKANINFO
|
||||
==========
|
||||
|
||||
Devices:
|
||||
========
|
||||
GPU0:
|
||||
\tdeviceName = Intel(R) Iris(R) Xe Graphics (TGL GT2)
|
||||
"""
|
||||
|
||||
|
||||
def _functions() -> str:
|
||||
chunks = [PATTERNS.group(0)]
|
||||
for name in FUNCTIONS:
|
||||
match = re.search(rf"^{name}\(\) \{{.*?^\}}", SOURCE, re.M | re.S)
|
||||
assert match, f"{name} not found in install.sh"
|
||||
chunks.append(match.group(0))
|
||||
return "\n".join(chunks)
|
||||
|
||||
|
||||
# PATH is restricted to these so a real vulkaninfo cannot answer for a stub.
|
||||
NEEDED_TOOLS = ("awk", "head", "grep", "cut", "tr", "xargs", "sed", "cat")
|
||||
|
||||
|
||||
def _run(tmp_path: Path, summary: str | None, script: str) -> subprocess.CompletedProcess:
|
||||
"""Run `script` with a stubbed vulkaninfo, or with none at all."""
|
||||
binary = tmp_path / "bin"
|
||||
binary.mkdir(exist_ok=True)
|
||||
for tool in NEEDED_TOOLS:
|
||||
found = shutil.which(tool)
|
||||
if found and not (binary / tool).exists():
|
||||
(binary / tool).symlink_to(found)
|
||||
if summary is not None:
|
||||
stub = binary / "vulkaninfo"
|
||||
stub.write_text(textwrap.dedent("""\
|
||||
#!/bin/bash
|
||||
case "$1" in
|
||||
--summary) cat <<'SUMMARY'
|
||||
{summary}
|
||||
SUMMARY
|
||||
;;
|
||||
--features) echo "" ;;
|
||||
esac
|
||||
""").format(summary=summary.rstrip("\n")))
|
||||
stub.chmod(0o755)
|
||||
env_path = f'export PATH="{binary}"\n'
|
||||
return subprocess.run(
|
||||
["bash", "-c", PRELUDE + env_path + _functions() + "\n" + script],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
|
||||
class TestDetectVulkan:
|
||||
def test_a_loader_without_devices_is_not_a_gpu(self, tmp_path):
|
||||
result = _run(
|
||||
tmp_path, NO_DEVICES, 'detect_vulkan; echo "RC=$? HAS=$HAS_VULKAN DEV=[$VULKAN_DEVICE]"'
|
||||
)
|
||||
assert "RC=1 HAS=no DEV=[]" in result.stdout, result.stdout + result.stderr
|
||||
|
||||
def test_it_never_reports_the_fallback_string_as_a_device(self, tmp_path):
|
||||
"""The old code printed this literal as though it were an adapter name."""
|
||||
result = _run(tmp_path, NO_DEVICES, 'detect_vulkan; echo "DEV=[$VULKAN_DEVICE]"')
|
||||
assert "Vulkan-compatible GPU" not in result.stdout
|
||||
|
||||
def test_a_device_listed_past_line_20_is_still_found(self, tmp_path):
|
||||
result = _run(
|
||||
tmp_path,
|
||||
DEVICE_PAST_LINE_20,
|
||||
'detect_vulkan; echo "HAS=$HAS_VULKAN DEV=[$VULKAN_DEVICE]"',
|
||||
)
|
||||
assert "HAS=yes" in result.stdout, result.stdout + result.stderr
|
||||
assert "DEV=[AMD Radeon RX 7900 XTX (RADV NAVI31)]" in result.stdout
|
||||
|
||||
def test_an_ordinary_device_is_reported_by_name(self, tmp_path):
|
||||
result = _run(
|
||||
tmp_path, DEVICE_EARLY, 'detect_vulkan; echo "HAS=$HAS_VULKAN DEV=[$VULKAN_DEVICE]"'
|
||||
)
|
||||
assert "HAS=yes" in result.stdout
|
||||
assert "DEV=[Intel(R) Iris(R) Xe Graphics (TGL GT2)]" in result.stdout
|
||||
|
||||
def test_absent_vulkaninfo_is_no(self, tmp_path):
|
||||
result = _run(tmp_path, None, 'detect_vulkan; echo "RC=$? HAS=$HAS_VULKAN"')
|
||||
assert "RC=1 HAS=no" in result.stdout, result.stdout + result.stderr
|
||||
|
||||
|
||||
class TestTheTwoDetectorsAgree:
|
||||
"""Both readers must see the same devices, or the wizard contradicts itself."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"summary", [NO_DEVICES, DEVICE_PAST_LINE_20, DEVICE_EARLY], ids=["none", "late", "early"]
|
||||
)
|
||||
def test_a_claimed_gpu_is_a_gpu_the_compat_check_can_see(self, tmp_path, summary):
|
||||
result = _run(
|
||||
tmp_path,
|
||||
summary,
|
||||
'detect_vulkan || true; echo "HAS=$HAS_VULKAN"; '
|
||||
"check_vulkan_gpu_compatibility || true",
|
||||
)
|
||||
claimed = "HAS=yes" in result.stdout
|
||||
saw_no_devices = "No hardware GPU found" in result.stdout
|
||||
assert not (claimed and saw_no_devices), (
|
||||
"detect_vulkan announced a GPU that check_vulkan_gpu_compatibility "
|
||||
f"cannot find:\n{result.stdout}"
|
||||
)
|
||||
|
||||
|
||||
class TestSoftwareRenderers:
|
||||
"""whisper.cpp on a CPU implementation of Vulkan is slower than its own CPU
|
||||
backend, so a software renderer must not be reported as a GPU."""
|
||||
|
||||
def test_a_software_renderer_is_not_a_gpu(self, tmp_path):
|
||||
result = _run(
|
||||
tmp_path,
|
||||
SOFTWARE_ONLY,
|
||||
'detect_vulkan; echo "RC=$? HAS=$HAS_VULKAN DEV=[$VULKAN_DEVICE]"',
|
||||
)
|
||||
assert "RC=1 HAS=no DEV=[]" in result.stdout, result.stdout + result.stderr
|
||||
|
||||
def test_it_is_remembered_for_the_message(self, tmp_path):
|
||||
"""Step 1 says "software Vulkan only", which beats "no GPU detected"."""
|
||||
result = _run(
|
||||
tmp_path, SOFTWARE_ONLY, 'detect_vulkan || true; echo "SOFT=[$VULKAN_SOFTWARE_DEVICE]"'
|
||||
)
|
||||
assert "SOFT=[llvmpipe (LLVM 15.0.6, 256 bits)]" in result.stdout
|
||||
|
||||
def test_hardware_still_wins_when_listed_after_software(self, tmp_path):
|
||||
result = _run(
|
||||
tmp_path,
|
||||
SOFTWARE_THEN_HARDWARE,
|
||||
'detect_vulkan; echo "HAS=$HAS_VULKAN DEV=[$VULKAN_DEVICE]"',
|
||||
)
|
||||
assert "HAS=yes" in result.stdout, result.stdout + result.stderr
|
||||
assert "DEV=[AMD Radeon RX 7900 XTX (RADV NAVI31)]" in result.stdout
|
||||
|
||||
def test_both_detectors_agree_that_it_is_not_hardware(self, tmp_path):
|
||||
result = _run(
|
||||
tmp_path,
|
||||
SOFTWARE_ONLY,
|
||||
'detect_vulkan || true; echo "HAS=$HAS_VULKAN"; check_vulkan_gpu_compatibility || true',
|
||||
)
|
||||
assert "HAS=no" in result.stdout
|
||||
assert "No hardware GPU found" in result.stdout
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"name", ["llvmpipe", "swiftshader", "lavapipe", "zink", "virtio", "venus"]
|
||||
)
|
||||
def test_every_pattern_is_recognised(self, tmp_path, name):
|
||||
result = _run(
|
||||
tmp_path, NO_DEVICES, f'is_software_renderer "{name} (something)"; echo "RC=$?"'
|
||||
)
|
||||
assert "RC=0" in result.stdout, result.stdout + result.stderr
|
||||
|
||||
|
||||
class TestDiagnosticsDoNotPolluteCapturedOutput:
|
||||
"""install.sh runs under `set -Eeuo pipefail` with an ERR trap that calls
|
||||
print_error. While print_error wrote to stdout, that message landed *inside*
|
||||
`$(detect_whispercpp_backends)` -- so a machine with a working Vulkan GPU
|
||||
parsed garbage, fell back to CPU, and never logged the error, because the
|
||||
substitution had swallowed it. `vulkaninfo --features` exits non-zero on real
|
||||
drivers, which is what fired the trap.
|
||||
"""
|
||||
|
||||
# A working hardware GPU, and --features exiting non-zero as it does in practice.
|
||||
STUB = """#!/bin/bash
|
||||
case "$1" in
|
||||
--summary) echo "\tdeviceName = Intel(R) Graphics (LNL)" ;;
|
||||
--features) echo "VK_KHR_16bit_storage"; exit 1 ;;
|
||||
esac
|
||||
"""
|
||||
|
||||
FUNCTIONS = (
|
||||
"is_software_renderer",
|
||||
"vulkan_device_names",
|
||||
"detect_nvidia_gpu",
|
||||
"detect_vulkan",
|
||||
"check_vulkan_gpu_compatibility",
|
||||
"cuda_toolkit_root_has_runtime_library",
|
||||
"find_valid_cuda_toolkit_root",
|
||||
"detect_whispercpp_backends",
|
||||
# the real ones, so the test fails if they go back to stdout
|
||||
"print_error",
|
||||
"print_warning",
|
||||
)
|
||||
|
||||
def _script(self, tmp_path: Path, body: str) -> subprocess.CompletedProcess:
|
||||
binary = tmp_path / "bin"
|
||||
binary.mkdir(exist_ok=True)
|
||||
for tool in NEEDED_TOOLS + ("find", "ls", "wc", "dirname", "basename", "readlink"):
|
||||
found = shutil.which(tool)
|
||||
if found and not (binary / tool).exists():
|
||||
(binary / tool).symlink_to(found)
|
||||
stub = binary / "vulkaninfo"
|
||||
stub.write_text(self.STUB)
|
||||
stub.chmod(0o755)
|
||||
|
||||
chunks = [PATTERNS.group(0)]
|
||||
for name in self.FUNCTIONS:
|
||||
match = re.search(rf"^{name}\(\) \{{.*?^\}}", SOURCE, re.M | re.S)
|
||||
assert match, f"{name} not found in install.sh"
|
||||
chunks.append(match.group(0))
|
||||
|
||||
prelude = (
|
||||
"set -Eeuo pipefail\n"
|
||||
f'export PATH="{binary}"\n'
|
||||
'HAS_NVIDIA_GPU="no"\nHAS_VULKAN="no"\nVULKAN_DEVICE=""\n'
|
||||
'VULKAN_SOFTWARE_DEVICE=""\nGPU_NAME=""\nGPU_MEMORY=""\n'
|
||||
'INSTALL_LOG_FILE="/dev/null"\n'
|
||||
'print_info() { echo "INFO: $*"; }\n'
|
||||
'print_success() { echo "SUCCESS: $*"; }\n'
|
||||
)
|
||||
trap = (
|
||||
"trap 'print_error \"Unexpected error near line $LINENO "
|
||||
"(exit code $?); see $INSTALL_LOG_FILE\"' ERR\n"
|
||||
)
|
||||
return subprocess.run(
|
||||
["bash", "-c", prelude + "\n".join(chunks) + "\n" + trap + body],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
def test_print_error_stays_off_stdout(self, tmp_path):
|
||||
result = self._script(tmp_path, 'print_error "boom"')
|
||||
assert "boom" not in result.stdout
|
||||
assert "boom" in result.stderr
|
||||
|
||||
def test_print_warning_stays_off_stdout(self, tmp_path):
|
||||
result = self._script(tmp_path, 'print_warning "careful"')
|
||||
assert "careful" not in result.stdout
|
||||
assert "careful" in result.stderr
|
||||
|
||||
def test_backend_info_is_a_single_clean_line(self, tmp_path):
|
||||
result = self._script(
|
||||
tmp_path, 'B=$(detect_whispercpp_backends); printf "%s" "$B" | wc -l; echo "[$B]"'
|
||||
)
|
||||
assert "ERROR" not in result.stdout, f"trap output captured:\n{result.stdout}"
|
||||
lines = result.stdout.splitlines()
|
||||
assert lines, result.stdout + result.stderr
|
||||
assert lines[0] == "0", f"BACKEND_INFO spans several lines:\n{result.stdout}"
|
||||
|
||||
def test_a_working_gpu_is_still_offered(self, tmp_path):
|
||||
"""The user-visible symptom: Step 3 refused GPU on hardware that supports it."""
|
||||
result = self._script(
|
||||
tmp_path,
|
||||
'B=$(detect_whispercpp_backends); CAN=$(echo "$B" | cut -d: -f3); echo "CAN=[$CAN]"',
|
||||
)
|
||||
assert "CAN=[true]" in result.stdout, result.stdout + result.stderr
|
||||
@@ -571,5 +571,98 @@ download_model_file() {
|
||||
self.assertEqual(leftovers, [])
|
||||
|
||||
|
||||
class TestReleaseWithoutAManifest(unittest.TestCase):
|
||||
"""install.sh comes from main; the tree it clones is the last release tag.
|
||||
|
||||
An absent manifest means the release predates pinning, so models are left to
|
||||
first run. A manifest that omits a model still fails closed.
|
||||
"""
|
||||
|
||||
SOURCE = INSTALL_SH.read_text()
|
||||
|
||||
PRELUDE = """
|
||||
set -uo pipefail
|
||||
print_info() { echo "INFO: $*"; }
|
||||
print_warning() { echo "WARNING: $*"; }
|
||||
print_error() { echo "ERROR: $*"; }
|
||||
print_success() { echo "SUCCESS: $*"; }
|
||||
command_exists() { command -v "$1" >/dev/null 2>&1; }
|
||||
check_connectivity() { return 0; }
|
||||
download_model_file() { echo "DOWNLOAD ATTEMPTED"; return 1; }
|
||||
"""
|
||||
|
||||
def _function_body(self, name: str) -> str:
|
||||
start = self.SOURCE.index(f"\n{name}() {{")
|
||||
end = self.SOURCE.index("\n}\n", start) + len("\n}\n")
|
||||
return self.SOURCE[start:end]
|
||||
|
||||
def _run(self, script: str, functions=()) -> subprocess.CompletedProcess:
|
||||
body = self.PRELUDE + "\n".join(self._function_body(n) for n in functions)
|
||||
return subprocess.run(["bash", "-c", body + "\n" + script], capture_output=True, text=True)
|
||||
|
||||
def test_available_when_the_manifest_ships(self):
|
||||
result = self._run(
|
||||
f'MODEL_CHECKSUMS_FILE="{REPO_ROOT}/src/vocalinux/utils/model_checksums.txt"; '
|
||||
"model_verification_available",
|
||||
functions=("model_verification_available",),
|
||||
)
|
||||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
|
||||
def test_unavailable_when_the_release_predates_pinning(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
result = self._run(
|
||||
f'MODEL_CHECKSUMS_FILE="{tmp}/nope.txt"; model_verification_available',
|
||||
functions=("model_verification_available",),
|
||||
)
|
||||
self.assertEqual(result.returncode, 1)
|
||||
|
||||
def test_vosk_refuses_before_spending_the_download(self):
|
||||
"""The 40MB fetch must not happen when nothing can verify it."""
|
||||
with TemporaryDirectory() as tmp:
|
||||
result = self._run(
|
||||
f'DATA_DIR="{tmp}/data"; VOCALINUX_TMP_DIR="{tmp}/tmp"; '
|
||||
f'mkdir -p "$DATA_DIR" "$VOCALINUX_TMP_DIR"; '
|
||||
f'MODEL_CHECKSUMS_FILE="{tmp}/nope.txt"; '
|
||||
"install_vosk_models; echo RC=$?",
|
||||
functions=("pinned_digest_for", "install_vosk_models"),
|
||||
)
|
||||
out = result.stdout + result.stderr
|
||||
self.assertIn("RC=1", out)
|
||||
self.assertIn("No checksum is pinned", out)
|
||||
self.assertNotIn("DOWNLOAD ATTEMPTED", out, "refused only after downloading 40MB")
|
||||
|
||||
def _model_install_block(self) -> str:
|
||||
"""The top-level section deciding which models are pre-downloaded."""
|
||||
start = self.SOURCE.index("# Install models based on selected engine")
|
||||
end = self.SOURCE.index("# config.json survives reinstalls", start)
|
||||
return self.SOURCE[start:end]
|
||||
|
||||
def test_the_manifest_dependent_downloads_are_gated(self):
|
||||
"""Both engines whose digests live in the manifest sit behind the gate."""
|
||||
block = self._model_install_block()
|
||||
for call in ("install_whispercpp_model ||", "install_vosk_models ||"):
|
||||
gate = block.rindex("if model_verification_available; then", 0, block.index(call))
|
||||
self.assertLess(gate, block.index(call), f"{call} is not behind the gate")
|
||||
|
||||
def test_openai_whisper_stays_ungated(self):
|
||||
"""It verifies against the sha256 in its own URL, so needs no manifest."""
|
||||
block = self._model_install_block()
|
||||
call = block.index("install_whisper_model ||")
|
||||
preceding = block[:call]
|
||||
self.assertGreater(
|
||||
preceding.rindex("fi"),
|
||||
preceding.rindex("if model_verification_available; then"),
|
||||
"the OpenAI Whisper download must not be behind the manifest gate",
|
||||
)
|
||||
|
||||
def test_the_skew_is_explained_once(self):
|
||||
block = self._model_install_block()
|
||||
self.assertEqual(
|
||||
block.count("if ! model_verification_available; then"),
|
||||
1,
|
||||
"the reason models are skipped should be stated once, not per engine",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user