Coverage for src/lilbee/providers/fleet/engine_diagnostics.py: 100%

1"""Inspect the engine binary's linked runtimes and its device probe's output.""" 

2 

3from __future__ import annotations 

4 

5import shutil 

6import subprocess 

7from pathlib import Path 

8 

9_LDD_TIMEOUT_S = 10 

10# Substrings that mark a device-init failure in the probe output. The HIP 

11# backend is compiled from ggml-cuda, so its error lines also say "cuda". 

12_ERROR_MARKERS = ("error", "fail", "no cuda") 

13# How much of the probe output to quote when no specific error line is found. 

14_DIAGNOSTIC_TAIL_CHARS = 300 

15 

16 

17def ldd_output(binary: Path, env: dict[str, str]) -> str | None: 

18 """``ldd`` stdout for *binary* under *env*; None when ldd can't run on it.""" 

19 ldd = shutil.which("ldd") 

20 if ldd is None: 

21 return None 

22 try: 

23 proc = subprocess.run( # noqa: S603 - ldd path and the resolved binary 

24 [ldd, str(binary)], 

25 capture_output=True, 

26 text=True, 

27 encoding="utf-8", 

28 errors="replace", 

29 timeout=_LDD_TIMEOUT_S, 

30 env=env, 

31 check=False, 

32 ) 

33 except (OSError, subprocess.SubprocessError): 

34 # Not an ELF, a static binary, or a timeout: nothing to inspect. 

35 return None 

36 return proc.stdout 

37 

38 

39def links_any(binary: Path, env: dict[str, str], sonames: tuple[str, ...]) -> bool: 

40 """True when *binary* lists any of *sonames*, resolved or not.""" 

41 out = ldd_output(binary, env) 

42 if out is None: 

43 return False 

44 return any(soname in out for soname in sonames) 

45 

46 

47def device_probe_diagnostic(probe_output: str) -> str: 

48 """The probe's device-init error line, or a short tail of its output.""" 

49 out = probe_output.strip() 

50 for line in out.splitlines(): 

51 lowered = line.lower() 

52 if "cuda" in lowered and any(marker in lowered for marker in _ERROR_MARKERS): 

53 return line.strip() 

54 return out[-_DIAGNOSTIC_TAIL_CHARS:] if out else "(the engine's device probe printed nothing)"