Coverage for core / venv_paths.py: 76.7%

43 statements  

« prev     ^ index     » next       coverage.py v7.14.0, created at 2026-05-12 04:49 +0000

1""" 

2core/venv_paths.py — single source of truth for per-backend venv paths. 

3 

4Each TTS / VLM / STT backend that imposes a conflicting dep cage gets its 

5own venv under ``<data_dir>/venvs/<backend>/`` so its transitive deps 

6stay isolated from the bundled python-embed. Two consumers depend on 

7this resolution: 

8 

9 1. INSTALL path — Nunba's ``tts.backend_venv`` creates the venv, 

10 pip-installs into it, runs verification. 

11 2. SPAWN path — HARTOS's ``integrations.service_tools.gpu_worker`` 

12 spawns the worker subprocess from the SAME venv's ``python.exe``. 

13 

14If those two paths drift (e.g. install writes to ``A/venvs/`` but spawn 

15reads ``B/venvs/``), the worker fails at startup with ``ModuleNotFoundError`` 

16because the dep was installed to a venv the spawn path never looks at. 

17That parallel-paths bug surfaced 2026-05-03 for chatterbox_turbo — install 

18went into the venv, spawn went into python-embed → ``died during startup 

19(exit=1)``. Consolidating the resolution here is the close-out: both 

20``tts.backend_venv`` and ``gpu_worker`` import from this module so the 

21path is computed in exactly one place. 

22 

23Public API 

24---------- 

25 venv_root() -> str 

26 venv_path(backend) -> str 

27 venv_python(backend) -> str 

28 venv_python_if_exists(backend) -> Optional[str] 

29""" 

30from __future__ import annotations 

31 

32import os 

33import sys 

34from typing import Optional 

35 

36 

37_VENV_ROOT_CACHE: Optional[str] = None 

38 

39 

40def _reset_cache_for_tests() -> None: 

41 """Reset the cached venv root. Test hook only — do not call from 

42 production code (the cache makes hot-path lookups O(1)).""" 

43 global _VENV_ROOT_CACHE 

44 _VENV_ROOT_CACHE = None 

45 

46 

47def venv_root() -> str: 

48 """Return the directory that holds every per-backend venv. 

49 

50 Resolution order (highest priority first): 

51 1. ``NUNBA_VENV_ROOT_OVERRIDE`` env var (tests / custom deploys). 

52 2. ``core.platform_paths.get_data_dir() / "data" / "venvs"`` 

53 (the canonical answer in any normal install). 

54 3. OS-aware fallback when ``core.platform_paths`` is unimportable 

55 (pure-Nunba lint runs that have not yet activated HARTOS). 

56 """ 

57 override = os.environ.get("NUNBA_VENV_ROOT_OVERRIDE", "").strip() 

58 if override: 

59 os.makedirs(override, exist_ok=True) 

60 return override 

61 

62 global _VENV_ROOT_CACHE 

63 if _VENV_ROOT_CACHE is not None: 

64 return _VENV_ROOT_CACHE 

65 

66 try: 

67 from core.platform_paths import get_data_dir # type: ignore 

68 base = os.path.join(str(get_data_dir()), "data", "venvs") 

69 except Exception: 

70 # platform_paths unimportable — replicate its decision tree. 

71 home = os.path.expanduser("~") 

72 if sys.platform == "win32": 

73 base = os.path.join(home, "Documents", "Nunba", "data", "venvs") 

74 elif sys.platform == "darwin": 

75 base = os.path.join(home, "Library", "Application Support", 

76 "Nunba", "data", "venvs") 

77 else: 

78 base = os.path.join(home, ".config", "nunba", "data", "venvs") 

79 

80 os.makedirs(base, exist_ok=True) 

81 _VENV_ROOT_CACHE = base 

82 return base 

83 

84 

85def _validate_backend_name(backend: str) -> None: 

86 """Reject unsafe backend names before they touch the filesystem.""" 

87 if not backend or not isinstance(backend, str): 

88 raise ValueError(f"backend must be a non-empty string, got {backend!r}") 

89 if not backend.replace("_", "").replace("-", "").isalnum(): 

90 raise ValueError( 

91 f"backend name must be alphanumeric / underscore / dash only, " 

92 f"got {backend!r}" 

93 ) 

94 if backend.startswith("."): 

95 raise ValueError(f"backend name must not start with a dot: {backend!r}") 

96 

97 

98def venv_path(backend: str) -> str: 

99 """Return the directory for a specific backend's venv.""" 

100 _validate_backend_name(backend) 

101 return os.path.join(venv_root(), backend) 

102 

103 

104def venv_python(backend: str) -> str: 

105 """Return the canonical path to the Python executable inside a backend's venv. 

106 

107 The path is returned whether or not the venv exists on disk. Use 

108 ``venv_python_if_exists`` for the existence-checked variant. 

109 """ 

110 vpath = venv_path(backend) 

111 if sys.platform == "win32": 

112 return os.path.join(vpath, "Scripts", "python.exe") 

113 return os.path.join(vpath, "bin", "python") 

114 

115 

116def venv_python_if_exists(backend: Optional[str]) -> Optional[str]: 

117 """Return the venv's python.exe path if it exists on disk, else None. 

118 

119 The HARTOS spawn path uses this resolver: ``None`` lets the caller 

120 fall through to the bundled python-embed (the right behavior for 

121 backends that don't have their own venv yet). 

122 """ 

123 if not backend: 

124 return None 

125 try: 

126 candidate = venv_python(backend) 

127 except ValueError: 

128 return None 

129 return candidate if os.path.isfile(candidate) else None