diff --git a/ROADMAP.md b/ROADMAP.md index 281f54a..28dc354 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -25,6 +25,51 @@ It is a planning note, not a promise. - Add a lightweight launch smoke test (boot the app offscreen and exit cleanly) so CI confirms the GUI actually starts, not just that mocked logic passes - Export transcribed text as a shareable audio file (e.g. OGG/Opus or MP3) so longer transcripts can be sent as a voice message via WhatsApp or similar (builds on the TTS pipeline) +## Paket H — Audit hardening (2026-06-21) + +Derived from an external code audit, cross-verified by three independent +reviews (Claude, `security-reviewer`, Codex). Verdict: neither of the two +findings flagged "critical" is actually critical for this single-user, +no-network desktop app. The items below are real-but-lower-severity hardening +and quality improvements, ordered for step-by-step execution. + +Phase 1 — worthwhile hardening (recommended): + +- [x] H1 `config.py` `_load()`: split the broad `except Exception` (lines + ~115-117) into `json.JSONDecodeError` and `PermissionError`/`OSError`, + each logged with `exc_info=True` so failures are debuggable. (MEDIUM) +- [x] H2 `run.sh` secrets sourcing (lines ~24-30): add an ownership/permission + check before `source` (`[[ -O "$SECRETS_FILE" ]]` + warn if perms are + looser than 600); keep `source`. Do NOT use the audit's + `export $(grep ... | xargs)` fix — it corrupts keys with spaces/special + chars via word-splitting. (MEDIUM, defense-in-depth) + +Phase 2 — optional cleanups (LOW): + +- [x] H3 `transcribe.py` (line ~62): `Path(wav_file).resolve()` as + defense-in-depth (no real traversal vector exists; path is app-generated). +- [x] H4 `llm_service.py`: replace the `MagicMock` production fallback with a + small `_NullLLMClient` stub so `unittest.mock` is not on the prod path. +- [ ] H5 `hotkey_service.py` (line ~319): optional short `time.sleep` after the + inner `break` as cheap insurance (no real busy-loop — `select(...,1.0)` + and fd-pop already prevent CPU spin). Deferred intentionally: optional only, + no reproduced bug and no strong red test yet. + +Phase 3 — only if desired / verify first: + +- [x] H6 Make `PasteService` key delay configurable via config and wire it through app init/runtime updates. +- [x] H7 `transcribe._transcribe_openai`: avoid leaking `CUDA_VISIBLE_DEVICES` + mutations beyond the OpenAI Whisper call; local override restored after use. +- [x] H8 Collapse the duplicate default model source (`llm_service.MODEL` vs + `config.py` default) to a single source of truth. +- [x] H9 Verify (do not blindly remove) the `tts_openai_voice` validation set: + "marin"/"cedar" appear to be real newer OpenAI voices — the audit is + likely wrong here. Verified against the installed OpenAI client; no code + change needed. + +Explicitly out: treating finding 2 (path traversal) as critical; adopting the +audit's flawed `run.sh` xargs fix. + ## Not in scope - Hosted backend services diff --git a/app/blitztext_linux.py b/app/blitztext_linux.py index 0b2ad9c..d06e4d7 100644 --- a/app/blitztext_linux.py +++ b/app/blitztext_linux.py @@ -30,7 +30,7 @@ if PROJECT_DIR not in sys.path: sys.path.insert(0, PROJECT_DIR) -from app.config import Config, VALID_HOTKEY_KEYS +from app.config import Config, DEFAULTS, VALID_HOTKEY_KEYS from app.llm_service import LLMService, WorkflowType, LLM_WORKFLOWS, LLMServiceError from app.writing_presets import WRITING_PRESET_KEYS, get_preset, preset_index from app.hotkey_service import HotkeyWorker @@ -273,7 +273,7 @@ def init_ui(self) -> None: self.edit_llm_model = QLineEdit() self.edit_llm_model.setText(self.config.llm_model) - self.edit_llm_model.setPlaceholderText("gpt-4o-mini") + self.edit_llm_model.setPlaceholderText(DEFAULTS["llm_model"]) self.combo_tone = QComboBox() self.combo_tone.addItems(["formal", "neutral", "locker"]) @@ -604,7 +604,7 @@ def __init__(self, app: QApplication) -> None: self.llm_service = self._build_llm_service() self.audio_recorder = AudioRecorder() - self.paste_service = PasteService(autopaste=self.config.autopaste) + self.paste_service = PasteService(autopaste=self.config.autopaste, key_delay_ms=self.config.paste_key_delay_ms) # State machine state: "IDLE", "RECORDING", "TRANSCRIBING", "LLM_REWRITING" self.state = "IDLE" @@ -964,6 +964,7 @@ def _stop_recording_and_process(self) -> None: # Ensure PasteService has the latest autopaste configuration self.paste_service.autopaste = self.config.autopaste + self.paste_service.key_delay_ms = self.config.paste_key_delay_ms # Create the transcribe worker worker = _TranscribeWorker( diff --git a/app/config.py b/app/config.py index 33ca3c3..727f313 100644 --- a/app/config.py +++ b/app/config.py @@ -31,6 +31,7 @@ "llm_base_url": "", "llm_model": "gpt-4o-mini", "autopaste": True, + "paste_key_delay_ms": 80, "audio_device": "@DEFAULT_SOURCE@", "notes_folder": str(Path.home() / "Blitztext-Notizen"), "history_size": 50, @@ -112,8 +113,11 @@ def _load(self) -> dict[str, Any]: sanitized = dict(data) sanitized.pop("openai_api_key", None) return _deep_merge(DEFAULTS, sanitized) - except Exception: - logger.warning("Config could not be loaded, using defaults") + except json.JSONDecodeError: + logger.warning("Config file is not valid JSON, using defaults", exc_info=True) + return _deep_merge(DEFAULTS, {}) + except OSError: + logger.warning("Config file could not be read, using defaults", exc_info=True) return _deep_merge(DEFAULTS, {}) def save(self) -> None: @@ -253,6 +257,19 @@ def autopaste(self) -> bool: def autopaste(self, value: bool) -> None: self._data["autopaste"] = bool(value) + @property + def paste_key_delay_ms(self) -> int: + raw = self._data.get("paste_key_delay_ms", DEFAULTS["paste_key_delay_ms"]) + try: + value = int(raw) + except (TypeError, ValueError): + return DEFAULTS["paste_key_delay_ms"] + return max(0, min(1000, value)) + + @paste_key_delay_ms.setter + def paste_key_delay_ms(self, value: int) -> None: + self._data["paste_key_delay_ms"] = max(0, min(1000, int(value))) + @property def audio_device(self) -> str: return self._data.get("audio_device", "@DEFAULT_SOURCE@") diff --git a/app/llm_service.py b/app/llm_service.py index c0edfae..d308aae 100644 --- a/app/llm_service.py +++ b/app/llm_service.py @@ -4,13 +4,14 @@ import logging from typing import Any, Optional +from app.config import DEFAULTS from app.workflows import WorkflowType from app.writing_presets import DEFAULT_PRESET_KEY, get_preset logger = logging.getLogger("blitztext.llm_service") LLM_WORKFLOWS = {WorkflowType.TEXT_IMPROVER, WorkflowType.DAMPF_ABLASSEN, WorkflowType.EMOJI_TEXT} -MODEL = "gpt-4o-mini" +DEFAULT_LLM_MODEL = DEFAULTS["llm_model"] _DAMPF_SYSTEM = ( "Du erhältst ein emotional gesprochenes Transkript. Erkenne zuerst das eigentliche " @@ -40,6 +41,20 @@ class LLMServiceError(Exception): """Raised when an LLM call fails.""" +class _NullLLMClient: + """Minimaler Platzhalter, damit der Produktionspfad kein unittest.mock nutzt.""" + + class _Chat: + class _Completions: + @staticmethod + def create(*args, **kwargs): + raise RuntimeError("Null client cannot create completions") + + completions = _Completions() + + chat = _Chat() + + class LLMService: """Wraps OpenAI API calls for BlitztextLinux rewrite workflows.""" @@ -59,7 +74,7 @@ def __init__( self.api_key = api_key or "" self.api_key_env = api_key_env or "OPENAI_API_KEY" self.base_url = (base_url or "").strip() - self.model = (model or "").strip() or MODEL + self.model = (model or "").strip() or DEFAULT_LLM_MODEL self.tone = tone self.emoji_density = emoji_density self.dampf_system_prompt = dampf_system_prompt @@ -76,9 +91,7 @@ def __init__( except ImportError: self._openai_installed = False self._client_is_fallback_mock = True - from unittest.mock import MagicMock - - self.client = MagicMock() + self.client = _NullLLMClient() else: if self.api_key and self.api_key.strip(): self.client = openai.OpenAI(api_key=self.api_key, base_url=self.base_url or None) @@ -87,9 +100,7 @@ def __init__( # werfen bereits im Konstruktor bei leerem Key. Der eigentliche # Fehler wird zur Aufrufzeit über _check_openai() klar gemeldet, # damit die App auch ohne gesetzten Key startet. - from unittest.mock import MagicMock - - self.client = MagicMock() + self.client = _NullLLMClient() def is_available(self) -> bool: return bool(self.api_key and self.api_key.strip()) diff --git a/app/paste_service.py b/app/paste_service.py index 0dae3d9..783e5fc 100644 --- a/app/paste_service.py +++ b/app/paste_service.py @@ -56,12 +56,14 @@ class PasteService: svc.paste("Hallo Welt") """ - def __init__(self, autopaste: bool = True) -> None: + def __init__(self, autopaste: bool = True, key_delay_ms: int = _KEY_DELAY_MS) -> None: """ Args: autopaste: True = nach wl-copy automatisch Ctrl+V via ydotool senden. + key_delay_ms: Verzögerung zwischen ydotool-Keyevents in Millisekunden. """ self.autopaste = autopaste + self.key_delay_ms = max(0, int(key_delay_ms)) def paste(self, text: str) -> None: """Text ins Clipboard schreiben und optional einfuegen. @@ -154,7 +156,7 @@ def _ydotool_paste(self) -> None: time.sleep(_PASTE_DELAY) try: result = subprocess.run( - ["ydotool", "key", "--key-delay", str(_KEY_DELAY_MS), *_CTRL_V_KEYCODES], + ["ydotool", "key", "--key-delay", str(self.key_delay_ms), *_CTRL_V_KEYCODES], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, diff --git a/app/transcribe.py b/app/transcribe.py index a3245a8..6dc8e55 100644 --- a/app/transcribe.py +++ b/app/transcribe.py @@ -59,7 +59,7 @@ def transcribe( TranscribeError: Bei fehlender Abhaengigkeit, fehlendem File oder Modell-Fehler. """ - wav_file = Path(wav_file) + wav_file = Path(wav_file).resolve() if model not in VALID_MODEL_NAMES: raise TranscribeError( @@ -150,16 +150,25 @@ def _load_faster_whisper_model_class(): def _transcribe_openai(wav_file: str, model_name: str, language: str, hint: str | None = None) -> str: warnings.filterwarnings("ignore", message="FP16 is not supported on CPU") - if _should_force_cpu_for_openai(): + original_cuda_visible_devices = os.environ.get("CUDA_VISIBLE_DEVICES") + should_force_cpu = _should_force_cpu_for_openai() + if should_force_cpu: os.environ["CUDA_VISIBLE_DEVICES"] = "" - whisper = _load_openai_whisper_module() + try: + whisper = _load_openai_whisper_module() - model = whisper.load_model(model_name) - result = model.transcribe( - wav_file, - language=_normalize_language(language), - initial_prompt=hint, - ) + model = whisper.load_model(model_name) + result = model.transcribe( + wav_file, + language=_normalize_language(language), + initial_prompt=hint, + ) + finally: + if should_force_cpu: + if original_cuda_visible_devices is None: + os.environ.pop("CUDA_VISIBLE_DEVICES", None) + else: + os.environ["CUDA_VISIBLE_DEVICES"] = original_cuda_visible_devices if not isinstance(result, dict): logger.warning("Transcription result is not a dict: %r", type(result).__name__) return "" diff --git a/run.sh b/run.sh index 8c62d77..24cca1c 100755 --- a/run.sh +++ b/run.sh @@ -23,10 +23,18 @@ trap 'rm -f "${LOCKFILE}"' EXIT INT TERM # --- secrets.env (optional) --- if [[ -f "${SECRETS_FILE}" ]]; then - set -a - # shellcheck disable=SC1090 - source "${SECRETS_FILE}" - set +a + if [[ ! -O "${SECRETS_FILE}" ]]; then + echo "WARNUNG: ${SECRETS_FILE} gehört nicht dem aktuellen Nutzer und wird nicht geladen." >&2 + else + SECRETS_PERMS=$(stat -c '%a' "${SECRETS_FILE}" 2>/dev/null || true) + if [[ -n "${SECRETS_PERMS}" ]] && (( 10#${SECRETS_PERMS} > 600 )); then + echo "WARNUNG: ${SECRETS_FILE} hat zu offene Rechte (${SECRETS_PERMS}); erwartet 600 oder restriktiver." >&2 + fi + set -a + # shellcheck disable=SC1090 + source "${SECRETS_FILE}" + set +a + fi fi # --- venv-Prüfung --- diff --git a/tests/test_config.py b/tests/test_config.py index 19157df..848ffdc 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -2,6 +2,7 @@ from __future__ import annotations import json +import logging from pathlib import Path import pytest @@ -262,6 +263,43 @@ def test_env_value_wins_over_legacy_fallback(self, config_dir, monkeypatch): assert loaded.resolve_openai_api_key() == "env-placeholder" +class TestLoadFailures: + def test_invalid_json_logs_json_decode_error_with_exc_info(self, config_dir, caplog): + config_dir.mkdir(parents=True, exist_ok=True) + (config_dir / "config.json").write_text("{kaputt", encoding="utf-8") + + with caplog.at_level(logging.WARNING, logger="blitztext.config"): + loaded = BlitztextConfig(config_dir=config_dir) + + assert loaded.model == "base" + records = [r for r in caplog.records if "valid JSON" in r.message] + assert len(records) == 1 + assert records[0].exc_info is not None + assert records[0].exc_info[0] is json.JSONDecodeError + + def test_read_error_logs_os_error_with_exc_info(self, config_dir, caplog, monkeypatch): + config_dir.mkdir(parents=True, exist_ok=True) + config_file = config_dir / "config.json" + config_file.write_text("{}", encoding="utf-8") + original_read_text = Path.read_text + + def fake_read_text(self, *args, **kwargs): + if self == config_file: + raise PermissionError("denied") + return original_read_text(self, *args, **kwargs) + + monkeypatch.setattr(Path, "read_text", fake_read_text) + + with caplog.at_level(logging.WARNING, logger="blitztext.config"): + loaded = BlitztextConfig(config_dir=config_dir) + + assert loaded.model == "base" + records = [r for r in caplog.records if "could not be read" in r.message] + assert len(records) == 1 + assert records[0].exc_info is not None + assert records[0].exc_info[0] is PermissionError + + class TestUILanguage: """Tests für ui_language Config-Property.""" diff --git a/tests/test_features.py b/tests/test_features.py index 2378808..eed8053 100644 --- a/tests/test_features.py +++ b/tests/test_features.py @@ -35,6 +35,7 @@ def test_defaults(self, tmp_path): assert cfg.tts_openai_model == "gpt-4o-mini-tts" assert cfg.tts_openai_voice == "marin" assert cfg.notes_folder.endswith("Blitztext-Notizen") + assert cfg.paste_key_delay_ms == 80 def test_history_size_clamped(self, tmp_path): cfg = Config.load(tmp_path / "config.json") @@ -59,6 +60,7 @@ def test_roundtrip_save_load(self, tmp_path): cfg.tts_openai_voice = "nova" cfg.notes_folder = str(Path.home() / "Notizen") cfg.history_size = 25 + cfg.paste_key_delay_ms = 135 cfg.save() cfg2 = Config.load(path) assert cfg2.tts_voice == "de_DE-thorsten-medium.onnx" @@ -67,6 +69,7 @@ def test_roundtrip_save_load(self, tmp_path): assert cfg2.tts_openai_voice == "nova" assert cfg2.history_size == 25 assert cfg2.notes_folder.endswith("Notizen") + assert cfg2.paste_key_delay_ms == 135 def test_sanitize_bad_values(self, tmp_path): path = tmp_path / "config.json" diff --git a/tests/test_llm_service.py b/tests/test_llm_service.py index 7ddab6d..1edb856 100644 --- a/tests/test_llm_service.py +++ b/tests/test_llm_service.py @@ -6,7 +6,8 @@ import pytest -from app.llm_service import LLMService, LLMServiceError +from app.config import DEFAULTS +from app.llm_service import LLMService, LLMServiceError, _NullLLMClient from app.workflows import WorkflowType from app.writing_presets import WRITING_PRESETS @@ -51,8 +52,15 @@ def test_custom_terms_are_stored(self, mock_client): class TestProviderConfig: - def test_default_model_is_gpt_4o_mini(self, service): - assert service.model == "gpt-4o-mini" + + def test_empty_api_key_uses_null_client_instead_of_mock(self): + fake_openai = MagicMock() + with patch.dict("sys.modules", {"openai": fake_openai}): + service = LLMService(api_key="", base_url="") + assert isinstance(service.client, _NullLLMClient) + + def test_default_model_comes_from_config_defaults(self, service): + assert service.model == DEFAULTS["llm_model"] def test_custom_model_is_used_in_requests(self, mock_client): service = LLMService(api_key=DUMMY_API_KEY, client=mock_client, model="openai/gpt-4o") @@ -60,9 +68,9 @@ def test_custom_model_is_used_in_requests(self, mock_client): kwargs = mock_client.chat.completions.create.call_args.kwargs assert kwargs["model"] == "openai/gpt-4o" - def test_empty_model_falls_back_to_default(self, mock_client): + def test_empty_model_falls_back_to_config_default(self, mock_client): service = LLMService(api_key=DUMMY_API_KEY, client=mock_client, model="") - assert service.model == "gpt-4o-mini" + assert service.model == DEFAULTS["llm_model"] def test_base_url_stored_on_service(self, mock_client): service = LLMService(api_key=DUMMY_API_KEY, client=mock_client, base_url="https://openrouter.ai/api/v1") diff --git a/tests/test_run_script.py b/tests/test_run_script.py new file mode 100644 index 0000000..b80db04 --- /dev/null +++ b/tests/test_run_script.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import os +import subprocess +from pathlib import Path + + +def _prepare_run_script(tmp_path: Path, repo_root: Path) -> Path: + script_path = tmp_path / "run.sh" + script_path.write_text((repo_root / "run.sh").read_text(encoding="utf-8"), encoding="utf-8") + script_path.chmod(0o755) + + stub_python = tmp_path / ".venv" / "bin" / "python" + stub_python.parent.mkdir(parents=True, exist_ok=True) + stub_python.write_text( + "#!/usr/bin/env bash\nprintf 'BT_SAMPLE_VAR=%s\\n' \"${BT_SAMPLE_VAR:-}\"\n", + encoding="utf-8", + ) + stub_python.chmod(0o755) + + app_dir = tmp_path / "app" + app_dir.mkdir(parents=True, exist_ok=True) + (app_dir / "blitztext_linux.py").write_text("print('noop')\n", encoding="utf-8") + return script_path + + +def _run_script(script_path: Path, home_dir: Path, runtime_dir: Path) -> subprocess.CompletedProcess[str]: + env = os.environ.copy() + env.update({ + "HOME": str(home_dir), + "XDG_RUNTIME_DIR": str(runtime_dir), + }) + proc = subprocess.Popen( + ["bash", str(script_path)], + cwd=str(script_path.parent), + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + stdout, stderr = proc.communicate() + return subprocess.CompletedProcess(proc.args, proc.returncode, stdout, stderr) + + +def test_run_script_warns_when_file_is_more_permissive_than_0600(tmp_path: Path): + repo_root = Path(__file__).resolve().parents[1] + script_path = _prepare_run_script(tmp_path, repo_root) + home_dir = tmp_path / "home" + runtime_dir = tmp_path / "runtime" + runtime_dir.mkdir(parents=True, exist_ok=True) + env_file = home_dir / ".config" / "blitztext-linux" / "secrets.env" + env_file.parent.mkdir(parents=True, exist_ok=True) + env_file.write_text("BT_SAMPLE_VAR=from-env\n", encoding="utf-8") + env_file.chmod(0o644) + + result = _run_script(script_path, home_dir, runtime_dir) + + assert result.returncode == 0 + assert "BT_SAMPLE_VAR=from-env" in result.stdout + assert "644" in result.stderr + assert "600" in result.stderr + + +def test_run_script_stays_quiet_when_file_permissions_are_0600(tmp_path: Path): + repo_root = Path(__file__).resolve().parents[1] + script_path = _prepare_run_script(tmp_path, repo_root) + home_dir = tmp_path / "home" + runtime_dir = tmp_path / "runtime" + runtime_dir.mkdir(parents=True, exist_ok=True) + env_file = home_dir / ".config" / "blitztext-linux" / "secrets.env" + env_file.parent.mkdir(parents=True, exist_ok=True) + env_file.write_text("BT_SAMPLE_VAR=from-env\n", encoding="utf-8") + env_file.chmod(0o600) + + result = _run_script(script_path, home_dir, runtime_dir) + + assert result.returncode == 0 + assert "BT_SAMPLE_VAR=from-env" in result.stdout + assert result.stderr == "" diff --git a/tests/test_settings_dialog.py b/tests/test_settings_dialog.py index b4a06e1..b5501a5 100644 --- a/tests/test_settings_dialog.py +++ b/tests/test_settings_dialog.py @@ -204,6 +204,7 @@ def test_app_init_applies_configured_ui_language(tmp_path): config = BlitztextConfig(config_dir=tmp_path / ".config" / "blitztext-linux") config.ui_language = "en" + config.paste_key_delay_ms = 135 fake_qapp = Mock() try: @@ -212,10 +213,11 @@ def test_app_init_applies_configured_ui_language(tmp_path): patch.object(BlitztextApp, "setup_tray"), \ patch.object(BlitztextApp, "start_hotkey_worker"), \ patch("app.blitztext_linux.AudioRecorder"), \ - patch("app.blitztext_linux.PasteService"): + patch("app.blitztext_linux.PasteService") as paste_service_cls: BlitztextApp(fake_qapp) assert get_language() == "en" + paste_service_cls.assert_called_once_with(autopaste=config.autopaste, key_delay_ms=135) finally: set_language(DEFAULT_LANGUAGE) diff --git a/tests/test_state_machine.py b/tests/test_state_machine.py index b886e2c..a3d34ec 100644 --- a/tests/test_state_machine.py +++ b/tests/test_state_machine.py @@ -60,6 +60,21 @@ def test_paste_passes_timeout_to_subprocess(self): svc.paste("text") assert run_mock.call_args.kwargs.get("timeout") is not None + def test_paste_uses_configured_key_delay_for_ydotool(self): + svc = PasteService(autopaste=True, key_delay_ms=135) + with patch("app.paste_service.shutil.which", return_value="/usr/bin/tool"), \ + patch("app.paste_service.time.sleep"), \ + patch("app.paste_service.subprocess.run") as run_mock: + def side_effect(cmd, *args, **kwargs): + if cmd[0] == "wl-copy": + return subprocess.CompletedProcess(cmd, 0, b"", b"") + return subprocess.CompletedProcess(cmd, 0, b"", b"") + run_mock.side_effect = side_effect + svc.paste("hallo welt") + + ydotool_call = run_mock.call_args_list[1] + assert ydotool_call.args[0][:4] == ["ydotool", "key", "--key-delay", "135"] + def test_paste_missing_ydotoold_does_not_raise(self, caplog): svc = PasteService(autopaste=True) caplog.set_level(logging.WARNING, logger="blitztext.paste_service") diff --git a/tests/test_transcribe.py b/tests/test_transcribe.py new file mode 100644 index 0000000..3e361bc --- /dev/null +++ b/tests/test_transcribe.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import os +from pathlib import Path +from unittest.mock import patch + +from app.transcribe import transcribe + + +def test_transcribe_resolves_path_before_backend_call(tmp_path: Path): + wav_file = tmp_path / "sample.wav" + wav_file.write_bytes(b"RIFFdata") + + with patch("app.transcribe.Path.resolve", autospec=True, side_effect=lambda self: self) as resolve_mock, \ + patch("app.transcribe._transcribe_openai", return_value="OK") as backend_mock: + result = transcribe(wav_file, backend="openai-whisper") + + assert result == "OK" + resolve_mock.assert_called_once() + backend_mock.assert_called_once() + + +def test_transcribe_openai_cpu_override_is_local_to_the_call(tmp_path: Path, monkeypatch): + wav_file = tmp_path / "sample.wav" + wav_file.write_bytes(b"RIFFdata") + monkeypatch.delenv("WHISPER_USE_CUDA", raising=False) + monkeypatch.setenv("CUDA_VISIBLE_DEVICES", "7") + + observed = {} + + class FakeModel: + def transcribe(self, *_args, **_kwargs): + observed["during_transcribe"] = os.environ.get("CUDA_VISIBLE_DEVICES") + return {"text": " Hallo "} + + fake_model = FakeModel() + + class FakeWhisperModule: + @staticmethod + def load_model(_model_name): + observed["during_load_model"] = os.environ.get("CUDA_VISIBLE_DEVICES") + return fake_model + + with patch("app.transcribe._load_openai_whisper_module", return_value=FakeWhisperModule()): + result = transcribe(wav_file, backend="openai-whisper") + + assert result == "Hallo" + assert observed == { + "during_load_model": "", + "during_transcribe": "", + } + assert os.environ.get("CUDA_VISIBLE_DEVICES") == "7"