diff --git a/src/local_transcriber/backends/faster_whisper.py b/src/local_transcriber/backends/faster_whisper.py index abf4d41..c75fe87 100644 --- a/src/local_transcriber/backends/faster_whisper.py +++ b/src/local_transcriber/backends/faster_whisper.py @@ -167,7 +167,6 @@ class FasterWhisperBackend: except Exception as exc: # noqa: BLE001 — диагностика не должна ронять запуск cuda_devices = f"недоступно ({exc})" return { - "engine": "faster-whisper", "faster_whisper": package_version("faster-whisper"), "ctranslate2": package_version("ctranslate2"), "cuda_devices": cuda_devices, diff --git a/src/local_transcriber/backends/onnx_asr.py b/src/local_transcriber/backends/onnx_asr.py index ab04c9c..3c4eb13 100644 --- a/src/local_transcriber/backends/onnx_asr.py +++ b/src/local_transcriber/backends/onnx_asr.py @@ -229,7 +229,6 @@ class OnnxAsrBackend: configured = ", ".join(self._providers) return { - "engine": "onnx-asr", "onnxruntime": package_version("onnxruntime"), "onnx_asr": package_version("onnx-asr"), "available_providers": ", ".join(onnxruntime.get_available_providers()), diff --git a/src/local_transcriber/backends/openvino.py b/src/local_transcriber/backends/openvino.py index 4048128..e5a1be7 100644 --- a/src/local_transcriber/backends/openvino.py +++ b/src/local_transcriber/backends/openvino.py @@ -209,7 +209,6 @@ class OpenVINOBackend: from openvino import Core return { - "engine": "openvino", "openvino": package_version("openvino"), "available_devices": ", ".join(Core().available_devices), "device": self.actual_ov_device or "", diff --git a/src/local_transcriber/cli.py b/src/local_transcriber/cli.py index eed8f4d..554025a 100644 --- a/src/local_transcriber/cli.py +++ b/src/local_transcriber/cli.py @@ -13,8 +13,6 @@ from rich.status import Status from .config import ( CliValues, - ResolvedConfig, - apply_device_defaults, load_config, resolve_defaults, ) @@ -39,11 +37,12 @@ from .quality import ( ) from .speaker_diarizer import SpeakerDiarizer, load_speaker_diarizer from .transcriber import ( + ExecutionInfo, + ExecutionRequest, Segment, + Transcriber, TranscribeResult, - _transcribe_file, cuda_error_hint, - load_model, ) from .types import ( UNKNOWN_LANGUAGE, @@ -53,10 +52,7 @@ from .types import ( ) from .utils import ( build_output_path, - detect_device, expand_globs, - get_gpu_name, - get_intel_gpu_name, has_existing_transcript, validate_input_file, ) @@ -95,21 +91,6 @@ def _with_cli_warning_renderer( return wrapped -def _format_device_info(device_used: str) -> str: - """Формирует строку устройства для шапки транскрипта.""" - if device_used == "cuda": - gpu_name = get_gpu_name() - return f"CUDA ({gpu_name or 'Unknown GPU'})" - if device_used == "openvino-gpu": - gpu_name = get_intel_gpu_name() - return f"OpenVINO ({gpu_name or 'Intel GPU'})" - if device_used in ("openvino", "openvino-cpu"): - return "OpenVINO (CPU)" - if device_used == "onnx": - return "ONNX (CPU)" - return "CPU" - - def _format_language_mode(requested_language: str, result: TranscribeResult) -> str: """Описывает источник языка, не выдавая профиль модели за детектор.""" if requested_language != "auto": @@ -348,7 +329,7 @@ def main( ) raise SystemExit(2) - resolved_device: str | None = None + requested_device: str | None = None try: config = load_config() cli_values: CliValues = { @@ -360,11 +341,19 @@ def main( } defaults = resolve_defaults(cli_values, config) diarize_enabled = defaults["diarize"] is True or speakers is not None + requested_device = defaults["device"] - resolved_device = detect_device(defaults["device"]) - defaults = apply_device_defaults(defaults, resolved_device, cli_values, config) - - ct_explicit = compute_type is not None or "compute_type" in config + # Явные model/compute_type; умолчания исполнения разрешает module. + request = ExecutionRequest( + device=requested_device, + model=model if model is not None else config.get("model"), + compute_type=( + compute_type if compute_type is not None else config.get("compute_type") + ), + language=defaults["language"], + cpu_threads=threads, + require_word_timestamps=diarize_enabled, + ) expanded = expand_globs(files) if not expanded: @@ -381,22 +370,18 @@ def main( if is_batch: _run_batch( expanded, - defaults, + request, verbose, force, - ct_explicit, - cpu_threads=threads, diarize=diarize_enabled, speakers=speakers, ) else: _run_single( expanded[0], - defaults, + request, output, verbose, - ct_explicit, - cpu_threads=threads, diarize=diarize_enabled, speakers=speakers, ) @@ -406,15 +391,15 @@ def main( except SystemExit: raise except ValueError as exc: - _print_cuda_hint(exc, resolved_device) + _print_cuda_hint(exc, requested_device) console.print(f"Ошибка: {exc}", style="red bold") raise SystemExit(1) except (FileNotFoundError,) as exc: - _print_cuda_hint(exc, resolved_device) + _print_cuda_hint(exc, requested_device) console.print(f"Ошибка: {exc}", style="red bold") raise SystemExit(1) except Exception as exc: - _print_cuda_hint(exc, resolved_device) + _print_cuda_hint(exc, requested_device) if verbose: console.print_exception() else: @@ -423,13 +408,48 @@ def main( raise SystemExit(1) +def _print_execution_header(info: ExecutionInfo, verbose: bool) -> None: + """Печатает выбранное исполнение; в --verbose — версии runtime для диагностики.""" + console.print( + f"Модель: [bold]{info.model}[/bold] " + f"Устройство: [bold]{info.device}[/bold] " + f"Compute: [bold]{info.compute_type}[/bold]" + ) + if info.device == "openvino-gpu" and info.model != "large-v3": + console.print( + "Совет: --model large-v3 даёт лучшее качество на GPU (~2x дольше)", + style="dim", + ) + if info.device != info.resolved_device: + console.print( + f"Запрошено {info.requested_device}, используется {info.device}", + style="yellow", + ) + if verbose: + threads = info.cpu_threads or "по умолчанию библиотеки" + console.print(f"Движок: {info.engine} Потоки: {threads}", style="dim") + for key, value in info.runtime.items(): + console.print(f" {key}: {value}", style="dim", markup=False) + + +def _load_diarizer( + diarize: bool, speakers: int | None, cpu_threads: int +) -> SpeakerDiarizer | None: + """Готовит диаризатор до первого ASR; пословный контракт уже проверен module.""" + if not diarize: + return None + return load_speaker_diarizer( + speakers=speakers, + threads=cpu_threads, + on_status=lambda message: console.print(message), + ) + + def _run_single( file: Path, - defaults: ResolvedConfig, + request: ExecutionRequest, output: Path | None, verbose: bool, - compute_type_explicit: bool = False, - cpu_threads: int = 0, diarize: bool = False, speakers: int | None = None, ) -> None: @@ -437,9 +457,6 @@ def _run_single( start = time.monotonic() validated_file = validate_input_file(file) - requested_device = defaults["device"] - resolved_device = detect_device(requested_device) - strict = requested_device != "auto" output_path = build_output_path(validated_file, output) console.print(f"Файл: [bold]{validated_file.name}[/bold]") @@ -447,59 +464,18 @@ def _run_single( def on_segment(seg: Segment) -> None: console.print(f" [{seg.start:.2f}s] {seg.text.strip()}") - model_obj, actual_device, backend, model_path = load_model( - defaults["model"], - resolved_device, - defaults["compute_type"], - on_status=lambda msg: console.print(msg), - strict_device=strict, - compute_type_explicit=compute_type_explicit, - cpu_threads=cpu_threads, - ) - actual_ct = ( - getattr(backend, "actual_compute_type", defaults["compute_type"]) - or defaults["compute_type"] - ) - console.print( - f"Модель: [bold]{defaults['model']}[/bold] " - f"Устройство: [bold]{actual_device}[/bold] " - f"Compute: [bold]{actual_ct}[/bold]" - ) - if actual_device == "openvino-gpu" and defaults["model"] != "large-v3": - console.print( - "Совет: --model large-v3 даёт лучшее качество на GPU (~2x дольше)", - style="dim", - ) - - speaker_diarizer = None - if diarize: - if not backend.word_timestamps_available: - raise ValueError( - "Выбранный движок или модель не поддерживает пословные таймкоды" - ) - speaker_diarizer = load_speaker_diarizer( - speakers=speakers, - threads=cpu_threads, - on_status=lambda message: console.print(message), - ) + transcriber = Transcriber(request) + info = transcriber.prepare(on_status=lambda msg: console.print(msg)) + _print_execution_header(info, verbose) + speaker_diarizer = _load_diarizer(diarize, speakers, request.cpu_threads) with Status("Подготавливаю запуск...", console=console) as status: - tfr = _transcribe_file( - model=model_obj, - actual_device=actual_device, - backend=backend, - model_path=model_path, - file_path=validated_file, - model_name=defaults["model"], - compute_type=defaults["compute_type"], - language=defaults["language"] if defaults["language"] != "auto" else None, + result = transcriber.transcribe( + validated_file, on_segment=on_segment if verbose else None, on_status=status.update, - strict_device=strict, - cpu_threads=cpu_threads, ) - result = tfr.result speaker_transcript = None diarization_warning = None diarization_degraded = False @@ -526,18 +502,12 @@ def _run_single( if diarization_warning is not None: console.print(f"Внимание: {diarization_warning}", style="yellow") - if tfr.actual_device != resolved_device: - if requested_device == "auto": - console.print( - f"Определено устройство {resolved_device}, " - f"но использовано {tfr.actual_device} (fallback)", - style="yellow", - ) - else: - console.print( - f"Запрошено {requested_device}, использовано {tfr.actual_device}", - style="yellow", - ) + execution = transcriber.execution + if execution.device != info.device: + console.print( + f"Запрошено {execution.requested_device}, использовано {execution.device}", + style="yellow", + ) if len(result.segments) == 0: message = f"Речь не обнаружена в файле {validated_file.name}" @@ -545,14 +515,13 @@ def _run_single( message += "; диаризация не запускалась" console.print(message, style="yellow") - device_info = _format_device_info(result.device_used) - language_mode = _format_language_mode(defaults["language"], result) + language_mode = _format_language_mode(request.language or "auto", result) content = format_transcript( result=result, source_filename=validated_file.name, - model_name=defaults["model"], - device_info=device_info, + model_name=execution.model, + device_info=execution.description, language_mode=language_mode, speaker_transcript=speaker_transcript, diarization_warning=diarization_warning, @@ -569,11 +538,9 @@ def _run_single( def _run_batch( files: list[Path], - defaults: ResolvedConfig, + request: ExecutionRequest, verbose: bool, force: bool, - compute_type_explicit: bool = False, - cpu_threads: int = 0, diarize: bool = False, speakers: int | None = None, ) -> None: @@ -603,50 +570,11 @@ def _run_batch( raise SystemExit(1) return - # Phase 2: Load model (ensure + create в одном вызове) - requested_device = defaults["device"] - resolved_device = detect_device(requested_device) - strict = requested_device != "auto" - model_obj, actual_device, backend, model_path = load_model( - defaults["model"], - resolved_device, - defaults["compute_type"], - on_status=lambda msg: console.print(msg), - strict_device=strict, - compute_type_explicit=compute_type_explicit, - cpu_threads=cpu_threads, - ) - - if actual_device == "openvino-gpu" and defaults["model"] != "large-v3": - console.print( - "Совет: --model large-v3 даёт лучшее качество на GPU (~2x дольше)", - style="dim", - ) - - if actual_device != resolved_device: - if requested_device == "auto": - console.print( - f"Определено устройство {resolved_device}, " - f"но используется {actual_device} (fallback)", - style="yellow", - ) - else: - console.print( - f"Запрошено {requested_device}, используется {actual_device}", - style="yellow", - ) - - speaker_diarizer = None - if diarize: - if not backend.word_timestamps_available: - raise ValueError( - "Выбранный движок или модель не поддерживает пословные таймкоды" - ) - speaker_diarizer = load_speaker_diarizer( - speakers=speakers, - threads=cpu_threads, - on_status=lambda message: console.print(message), - ) + # Phase 2: module выполнения создаётся, когда есть работа + transcriber = Transcriber(request) + info = transcriber.prepare(on_status=lambda msg: console.print(msg)) + _print_execution_header(info, verbose) + speaker_diarizer = _load_diarizer(diarize, speakers, request.cpu_threads) # Phase 3: Transcribe processed = 0 @@ -663,39 +591,24 @@ def _run_batch( def on_segment(seg: Segment) -> None: console.print(f" [{seg.start:.2f}s] {seg.text.strip()}") + device_before = transcriber.execution.device with Status(f"{prefix}...", console=console) as status: - tfr = _transcribe_file( - model=model_obj, - actual_device=actual_device, - backend=backend, - model_path=model_path, - file_path=file, - model_name=defaults["model"], - compute_type=defaults["compute_type"], - language=defaults["language"] - if defaults["language"] != "auto" - else None, + result = transcriber.transcribe( + file, on_segment=on_segment if verbose else None, on_status=status.update if not verbose else lambda msg: console.print(msg), - strict_device=strict, - cpu_threads=cpu_threads, ) - if tfr.actual_device != actual_device: + execution = transcriber.execution + if execution.device != device_before: console.print( - f" {file.name}: fallback на {tfr.actual_device} при транскрипции", + f" {file.name}: fallback на {execution.device} при транскрипции", style="yellow", ) - # Обновляем после возможного mid-stream fallback - model_obj = tfr.model - actual_device = tfr.actual_device - backend = tfr.backend - model_path = tfr.model_path - result = tfr.result - language_mode = _format_language_mode(defaults["language"], result) + language_mode = _format_language_mode(request.language or "auto", result) speaker_transcript = None diarization_warning = None file_degraded = False @@ -735,13 +648,11 @@ def _run_batch( message += "; диаризация не запускалась" console.print(message, style="yellow") - device_info = _format_device_info(result.device_used) - content = format_transcript( result=result, source_filename=file.name, - model_name=defaults["model"], - device_info=device_info, + model_name=execution.model, + device_info=execution.description, language_mode=language_mode, speaker_transcript=speaker_transcript, diarization_warning=diarization_warning, @@ -760,7 +671,7 @@ def _run_batch( except KeyboardInterrupt: raise except Exception as exc: - _print_cuda_hint(exc, actual_device) + _print_cuda_hint(exc, request.device) if verbose: console.print_exception() else: diff --git a/src/local_transcriber/config.py b/src/local_transcriber/config.py index 0f36361..66f0896 100644 --- a/src/local_transcriber/config.py +++ b/src/local_transcriber/config.py @@ -171,23 +171,3 @@ def _resolve_value(cli_value: _T | None, config_value: _T | None, default: _T) - if config_value is not None: return config_value return default - - -def apply_device_defaults( - defaults: ResolvedConfig, - resolved_device: str, - cli_values: CliValues, - config: ConfigValues, -) -> ResolvedConfig: - """Применяет device-aware дефолты для model и compute_type, - если они не были явно заданы через CLI или конфиг.""" - device_defs = DEVICE_DEFAULTS.get(resolved_device, {}) - if not device_defs: - return defaults - - result = dict(defaults) - for key in ("model", "compute_type"): - if cli_values.get(key) is None and key not in config: - if key in device_defs: - result[key] = device_defs[key] - return cast(ResolvedConfig, result) diff --git a/src/local_transcriber/transcriber.py b/src/local_transcriber/transcriber.py index 1de3a27..d67a5e3 100644 --- a/src/local_transcriber/transcriber.py +++ b/src/local_transcriber/transcriber.py @@ -10,9 +10,8 @@ from local_transcriber.backends import get_backend from local_transcriber.config import DEVICE_DEFAULTS, HARDCODED_DEFAULTS # Re-export из types.py для обратной совместимости -from local_transcriber.types import ( +from local_transcriber.types import ( # noqa: F401 Segment, - TranscribeFileResult, TranscribeResult, WordTimestampsUnavailableError, ) @@ -228,120 +227,6 @@ def _describe_device(device: str) -> str: return "CPU" -def load_model( - model_name: str, - device: str, - compute_type: str, - on_status: Callable[[str], None] | None = None, - strict_device: bool = False, - compute_type_explicit: bool = False, - cpu_threads: int = 0, -) -> tuple[Any, str, Any, str]: - """Загружает модель: ensure + create с fallback. - - Возвращает (model, actual_device, backend, model_path). - compute_type_explicit: True если пользователь явно указал --compute-type. - cpu_threads: число потоков для CPU inference (0 = дефолт библиотеки). - """ - backend = get_backend(device, compute_type_explicit=compute_type_explicit) - actual_device = device - - model_path = backend.ensure_model_available(model_name, compute_type, on_status) - - try: - _notify_status(on_status, f"Инициализирую модель на {device}...") - model = backend.create_model( - model_path, device, compute_type, cpu_threads=cpu_threads - ) - # Резолвим actual_device по реальному OpenVINO device - ov_dev = getattr(backend, "actual_ov_device", None) - if ov_dev == "GPU" and actual_device != "openvino-gpu": - actual_device = "openvino-gpu" - elif ( - ov_dev == "CPU" - and actual_device.startswith("openvino") - and actual_device != "openvino-cpu" - ): - actual_device = "openvino-cpu" - except (RuntimeError, ValueError) as exc: - if device != "cpu" and _is_backend_error(exc, device): - if strict_device: - raise - warnings.warn( - f"Не удалось загрузить модель на {device}: {exc}. Переключение на CPU.", - stacklevel=2, - ) - actual_device = "cpu" - backend = get_backend("cpu") - model_path = backend.ensure_model_available( - model_name, compute_type, on_status - ) - _notify_status(on_status, "Инициализирую модель на cpu...") - model = backend.create_model( - model_path, "cpu", compute_type, cpu_threads=cpu_threads - ) - else: - raise - - return model, actual_device, backend, model_path - - -def _transcribe_file( - model: Any, - actual_device: str, - backend: Any, - model_path: str, - file_path: Path, - model_name: str, - compute_type: str, - language: str | None = None, - on_segment: Callable[[Segment], None] | None = None, - on_status: Callable[[str], None] | None = None, - strict_device: bool = False, - cpu_threads: int = 0, -) -> TranscribeFileResult: - """Транскрибирует один файл. При mid-stream fallback перезагружает модель.""" - lang_arg = language if language and language != "auto" else None - - try: - _notify_status(on_status, "Транскрибирую...") - result = backend.transcribe(model, file_path, lang_arg, on_segment, on_status) - result.device_used = actual_device - except (RuntimeError, ValueError) as exc: - if actual_device != "cpu" and _is_backend_error(exc, actual_device): - if strict_device: - raise - warnings.warn( - f"Ошибка при транскрипции на {actual_device}: {exc}. " - "Переключение на CPU и повтор.", - stacklevel=2, - ) - actual_device = "cpu" - backend = get_backend("cpu") - model_path = backend.ensure_model_available( - model_name, compute_type, on_status - ) - _notify_status(on_status, "Инициализирую модель на cpu...") - model = backend.create_model( - model_path, "cpu", compute_type, cpu_threads=cpu_threads - ) - _notify_status(on_status, "Транскрибирую...") - result = backend.transcribe( - model, file_path, lang_arg, on_segment, on_status - ) - result.device_used = actual_device - else: - raise - - return TranscribeFileResult( - result=result, - model=model, - actual_device=actual_device, - backend=backend, - model_path=model_path, - ) - - def transcribe( file_path: Path, model_name: str | None = None, @@ -377,8 +262,6 @@ def ensure_model_available( on_status: Callable[[str], None] | None = None, ) -> str: """Публичный helper: гарантирует наличие модели для указанного бэкенда.""" - from local_transcriber.config import DEVICE_DEFAULTS, HARDCODED_DEFAULTS - if compute_type is None: device_defs = DEVICE_DEFAULTS.get(device, {}) compute_type = device_defs.get( diff --git a/src/local_transcriber/types.py b/src/local_transcriber/types.py index d8b49c4..dd96461 100644 --- a/src/local_transcriber/types.py +++ b/src/local_transcriber/types.py @@ -2,7 +2,6 @@ from collections.abc import Callable from dataclasses import dataclass, field -from typing import Any # Единый признак «язык неизвестен» для всех бэкендов UNKNOWN_LANGUAGE = "unknown" @@ -83,14 +82,5 @@ class TranscribeResult: words: list[Word] = field(default_factory=list) -@dataclass -class TranscribeFileResult: - result: TranscribeResult - model: Any # backend-specific model handle - actual_device: str - backend: Any = None # backend instance (для переиспользования в батче) - model_path: str = "" # путь к модели (меняется при cross-backend fallback) - - StatusCallback = Callable[[str], None] | None SegmentCallback = Callable[[Segment], None] | None diff --git a/tests/test_backend_faster_whisper.py b/tests/test_backend_faster_whisper.py index ed27204..b4cd575 100644 --- a/tests/test_backend_faster_whisper.py +++ b/tests/test_backend_faster_whisper.py @@ -93,7 +93,6 @@ def test_runtime_info_reports_versions_and_cuda_devices(monkeypatch): info = backend.runtime_info() - assert info["engine"] == "faster-whisper" assert info["ctranslate2"] assert info["faster_whisper"] assert info["cuda_devices"] == "0" diff --git a/tests/test_backend_openvino.py b/tests/test_backend_openvino.py index 973416f..385ad8a 100644 --- a/tests/test_backend_openvino.py +++ b/tests/test_backend_openvino.py @@ -476,7 +476,6 @@ def test_runtime_info_reports_available_and_actual_devices(monkeypatch): info = backend.runtime_info() - assert info["engine"] == "openvino" assert info["openvino"] assert info["available_devices"] == "CPU, GPU" assert info["device"] == "GPU" diff --git a/tests/test_cli.py b/tests/test_cli.py index 72627d8..ce5745e 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,23 +1,20 @@ import warnings +from contextlib import ExitStack, contextmanager from pathlib import Path -from unittest.mock import MagicMock, patch +from unittest.mock import ANY, MagicMock, patch import pytest from rich.console import Console from typer.testing import CliRunner -from local_transcriber.cli import _format_device_info, _format_language_mode, app +from local_transcriber.cli import _format_language_mode, app from local_transcriber.formatter import ( LANGUAGE_DETECTED, LANGUAGE_FORCED, LANGUAGE_FROM_MODEL, LANGUAGE_UNKNOWN, ) -from local_transcriber.transcriber import ( - Segment, - TranscribeFileResult, - TranscribeResult, -) +from local_transcriber.transcriber import Segment, TranscribeResult from local_transcriber.types import ( UNKNOWN_LANGUAGE, DiarizationRun, @@ -40,34 +37,53 @@ def _make_result(segments=None, language="ru", device_used="cpu", duration=60.0) ) -def _make_model(): - return MagicMock(name="WhisperModel") - - -def _make_backend(): - return MagicMock(name="Backend") - - -def _make_tfr( +def _make_backend( result=None, - model=None, - actual_device="cpu", - backend=None, - model_path="/models/medium", + word_timestamps=True, + actual_ov_device=None, + runtime=None, ): - if result is None: - result = _make_result() - if model is None: - model = _make_model() - if backend is None: - backend = _make_backend() - return TranscribeFileResult( - result=result, - model=model, - actual_device=actual_device, - backend=backend, - model_path=model_path, + """Fake adapter на internal seam Backend; module выполнения работает по-настоящему.""" + backend = MagicMock(name="Backend") + backend.word_timestamps_available = word_timestamps + backend.actual_compute_type = None + backend.actual_ov_device = actual_ov_device + backend.ensure_model_available.side_effect = ( + lambda model_name, compute_type, on_status=None: f"/models/{model_name}" ) + backend.transcribe.return_value = result if result is not None else _make_result() + backend.runtime_info.return_value = runtime or {} + return backend + + +@contextmanager +def _cli_run(result=None, tmp_file=None, backend=None, config=None, write=True): + """Стандартный CLI-прогон: конфиг, валидация файла, fake adapter, запись. + + Отдаёт (backend, write_transcript); ``tmp_file=None`` пропускает файлы + как есть, ``write=False`` оставляет настоящую запись транскрипта. + """ + if backend is None: + backend = _make_backend(result) + validate = ( + {"return_value": tmp_file} + if tmp_file is not None + else {"side_effect": lambda p: p} + ) + with ExitStack() as stack: + stack.enter_context( + patch("local_transcriber.cli.load_config", return_value=config or {}) + ) + stack.enter_context(patch("local_transcriber.cli.validate_input_file", **validate)) + stack.enter_context( + patch("local_transcriber.transcriber.get_backend", return_value=backend) + ) + write_transcript = ( + stack.enter_context(patch("local_transcriber.cli.write_transcript")) + if write + else None + ) + yield backend, write_transcript @pytest.mark.parametrize( @@ -86,37 +102,44 @@ def test_format_language_mode(requested_language, language, probability, expecte assert _format_language_mode(requested_language, result) == expected -def _single_patches(result=None, tmp_file=None, actual_device="cpu"): - """Patches for a standard single-file CLI happy path.""" - if result is None: - result = _make_result(device_used=actual_device) - model = _make_model() - backend = _make_backend() - tfr = _make_tfr( - result=result, model=model, actual_device=actual_device, backend=backend - ) - return [ - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=tmp_file), - patch("local_transcriber.cli.detect_device", return_value=actual_device), - patch( - "local_transcriber.cli.load_model", - return_value=(model, actual_device, backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), - ] - - def test_cli_happy_path_exit_code_zero(tmp_path): audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - patches = _single_patches(tmp_file=audio) - with patches[0], patches[1], patches[2], patches[3], patches[4], patches[5]: + with _cli_run(tmp_file=audio) as (backend, _write): out = runner.invoke(app, [str(audio)]) assert out.exit_code == 0 + backend.ensure_model_available.assert_called_once_with( + "gigaam-v3-e2e-rnnt", "int8", ANY + ) + assert "Модель: gigaam-v3-e2e-rnnt Устройство: onnx Compute: int8" in out.output + + +def test_cli_verbose_prints_runtime_diagnostics(tmp_path): + """--verbose показывает движок, потоки и версии runtime из данных module.""" + audio = tmp_path / "test.mp3" + audio.write_bytes(b"fake") + backend = _make_backend(runtime={"onnxruntime": "1.28.0", "asr_providers": "CPUExecutionProvider"}) + + with _cli_run(tmp_file=audio, backend=backend): + out = runner.invoke(app, [str(audio), "--verbose", "--threads", "4"]) + + assert out.exit_code == 0 + assert "Движок: onnx-asr Потоки: 4" in out.output + assert "onnxruntime: 1.28.0" in out.output + assert "asr_providers: CPUExecutionProvider" in out.output + + +def test_cli_without_verbose_hides_runtime_diagnostics(tmp_path): + audio = tmp_path / "test.mp3" + audio.write_bytes(b"fake") + backend = _make_backend(runtime={"onnxruntime": "1.28.0"}) + + with _cli_run(tmp_file=audio, backend=backend): + out = runner.invoke(app, [str(audio)]) + + assert "onnxruntime" not in out.output @pytest.mark.parametrize("file_count", [1, 2]) @@ -125,34 +148,20 @@ def test_cli_renders_runtime_warning_without_python_details(tmp_path, file_count for file in files: file.write_bytes(b"fake") - result = _make_result() - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, backend=backend) warning_message = ( "Тестовое [bold]предупреждение[/bold] с длинным текстом, который " "должен остаться одной логической строкой без служебных подробностей Python" ) - def transcribe_with_warning(**_kwargs): + def transcribe_with_warning(*_args, **_kwargs): warnings.warn(warning_message, stacklevel=2) - return tfr + return _make_result() + + backend = _make_backend() + backend.transcribe.side_effect = transcribe_with_warning original_showwarning = warnings.showwarning - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch( - "local_transcriber.cli._transcribe_file", - side_effect=transcribe_with_warning, - ), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(backend=backend): out = runner.invoke(app, [str(file) for file in files]) warning_lines = [line for line in out.output.splitlines() if "Тестовое" in line] @@ -165,30 +174,15 @@ def test_cli_renders_runtime_warning_without_python_details(tmp_path, file_count def test_cli_default_options_passed_to_transcribe(tmp_path): audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - result = _make_result(device_used="onnx") - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, actual_device="onnx", backend=backend) - mock_transcribe_file = MagicMock(return_value=tfr) - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="onnx"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "onnx", backend, "/models/gigaam-v3-e2e-rnnt"), - ), - patch("local_transcriber.cli._transcribe_file", mock_transcribe_file), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(tmp_file=audio) as (backend, _write): out = runner.invoke(app, [str(audio)]) - call_kwargs = mock_transcribe_file.call_args[1] - assert call_kwargs["model_name"] == "gigaam-v3-e2e-rnnt" - assert call_kwargs["compute_type"] == "int8" - assert call_kwargs["language"] == "ru" - assert call_kwargs["on_segment"] is None # verbose=False + backend.ensure_model_available.assert_called_once_with( + "gigaam-v3-e2e-rnnt", "int8", ANY + ) + assert backend.transcribe.call_args[0][2] == "ru" + assert backend.transcribe.call_args[0][3] is None # verbose=False → on_segment assert "Модель: gigaam-v3-e2e-rnnt" in out.output assert "Устройство: onnx" in out.output @@ -196,20 +190,8 @@ def test_cli_default_options_passed_to_transcribe(tmp_path): def test_cli_identifies_onnx_backend_in_transcript_header(tmp_path): audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - result = _make_result(device_used="onnx") - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, actual_device="onnx", backend=backend) - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.detect_device", return_value="onnx"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "onnx", backend, "/models/gigaam-v3-e2e-rnnt"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - ): + with _cli_run(tmp_file=audio, write=False): out = runner.invoke(app, [str(audio)]) content = (tmp_path / "test-transcript.md").read_text(encoding="utf-8") @@ -220,24 +202,8 @@ def test_cli_identifies_onnx_backend_in_transcript_header(tmp_path): def test_cli_custom_options(tmp_path): audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - result = _make_result(device_used="cuda") - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, actual_device="cuda", backend=backend) - mock_transcribe_file = MagicMock(return_value=tfr) - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cuda"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cuda", backend, "/models/small"), - ), - patch("local_transcriber.cli._transcribe_file", mock_transcribe_file), - patch("local_transcriber.cli.write_transcript"), - patch("local_transcriber.cli.get_gpu_name", return_value="RTX 3060"), - ): + with _cli_run(tmp_file=audio) as (backend, _write): runner.invoke( app, [ @@ -253,10 +219,8 @@ def test_cli_custom_options(tmp_path): ], ) - call_kwargs = mock_transcribe_file.call_args[1] - assert call_kwargs["model_name"] == "small" - assert call_kwargs["language"] == "ru" - assert call_kwargs["compute_type"] == "float16" + backend.ensure_model_available.assert_called_once_with("small", "float16", ANY) + assert backend.transcribe.call_args[0][2] == "ru" def test_cli_speakers_enables_diarization_and_writes_speaker_markdown(tmp_path): @@ -271,10 +235,6 @@ def test_cli_speakers_enables_diarization_and_writes_speaker_markdown(tmp_path): Word(0.5, 1.0, "Второй."), Word(1.1, 1.3, "Неясно."), ] - model = _make_model() - backend = _make_backend() - backend.word_timestamps_available = True - tfr = _make_tfr(result=result, model=model, backend=backend) diarizer = MagicMock() diarizer.process.return_value = DiarizationRun( intervals=[ @@ -283,22 +243,13 @@ def test_cli_speakers_enables_diarization_and_writes_speaker_markdown(tmp_path): ], elapsed_seconds=0.2, ) - write = MagicMock() with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), + _cli_run(tmp_file=audio, result=result) as (_backend, write), patch( "local_transcriber.cli.load_speaker_diarizer", return_value=diarizer, ) as load_diarizer, - patch("local_transcriber.cli.write_transcript", write), ): out = runner.invoke( app, @@ -325,28 +276,15 @@ def test_cli_diarization_error_writes_plain_transcript_and_exits_nonzero(tmp_pat duration=10.0, ) result.words = [Word(0.0, 1.0, "Полезный текст.")] - model = _make_model() - backend = _make_backend() - backend.word_timestamps_available = True - tfr = _make_tfr(result=result, model=model, backend=backend) diarizer = MagicMock() diarizer.process.side_effect = RuntimeError("boom") - write = MagicMock() with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), + _cli_run(tmp_file=audio, result=result) as (_backend, write), patch( "local_transcriber.cli.load_speaker_diarizer", return_value=diarizer, ), - patch("local_transcriber.cli.write_transcript", write), ): out = runner.invoke(app, [str(audio), "--diarize"]) @@ -364,10 +302,6 @@ def test_cli_verbose_reports_diarization_counts_and_duration(tmp_path): duration=10.0, ) result.words = [Word(0.0, 0.5, "Раз"), Word(0.5, 1.0, "два")] - model = _make_model() - backend = _make_backend() - backend.word_timestamps_available = True - tfr = _make_tfr(result=result, model=model, backend=backend) diarizer = MagicMock() diarizer.process.return_value = DiarizationRun( intervals=[ @@ -378,19 +312,11 @@ def test_cli_verbose_reports_diarization_counts_and_duration(tmp_path): ) with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), + _cli_run(tmp_file=audio, result=result), patch( "local_transcriber.cli.load_speaker_diarizer", return_value=diarizer, ), - patch("local_transcriber.cli.write_transcript"), ): out = runner.invoke(app, [str(audio), "--diarize", "--verbose"]) @@ -403,26 +329,14 @@ def test_cli_empty_asr_skips_diarizer_and_reports_it(tmp_path): audio = tmp_path / "silence.wav" audio.write_bytes(b"fake") result = _make_result(segments=[]) - model = _make_model() - backend = _make_backend() - backend.word_timestamps_available = True - tfr = _make_tfr(result=result, model=model, backend=backend) diarizer = MagicMock() with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), + _cli_run(tmp_file=audio, result=result), patch( "local_transcriber.cli.load_speaker_diarizer", return_value=diarizer, ), - patch("local_transcriber.cli.write_transcript"), ): out = runner.invoke(app, [str(audio), "--diarize"]) @@ -434,31 +348,37 @@ def test_cli_empty_asr_skips_diarizer_and_reports_it(tmp_path): def test_cli_diarizer_preflight_failure_does_not_start_asr_or_write(tmp_path): audio = tmp_path / "meeting.mp3" audio.write_bytes(b"fake") - model = _make_model() - backend = _make_backend() - backend.word_timestamps_available = True - transcribe_file = MagicMock() - write = MagicMock() with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", transcribe_file), + _cli_run(tmp_file=audio) as (backend, write), patch( "local_transcriber.cli.load_speaker_diarizer", side_effect=RuntimeError("модель повреждена"), ), - patch("local_transcriber.cli.write_transcript", write), ): out = runner.invoke(app, [str(audio), "--diarize"]) assert out.exit_code == 1 - transcribe_file.assert_not_called() + backend.transcribe.assert_not_called() + write.assert_not_called() + + +def test_cli_diarize_rejects_model_without_word_timestamps_before_asr(tmp_path): + """Проверка пословного контракта проходит в module до диаризатора и ASR.""" + audio = tmp_path / "meeting.mp3" + audio.write_bytes(b"fake") + backend = _make_backend(word_timestamps=False) + + with ( + _cli_run(tmp_file=audio, backend=backend) as (backend, write), + patch("local_transcriber.cli.load_speaker_diarizer") as load_diarizer, + ): + out = runner.invoke(app, [str(audio), "--diarize"]) + + assert out.exit_code == 1 + assert "пословные таймкоды" in out.output + load_diarizer.assert_not_called() + backend.transcribe.assert_not_called() write.assert_not_called() @@ -479,28 +399,15 @@ def test_cli_unsuccessful_diarization_shape_writes_plain_text_and_exits_nonzero( duration=10.0, ) result.words = [Word(0.0, 0.5, "Раз"), Word(0.5, 1.0, "два")] - model = _make_model() - backend = _make_backend() - backend.word_timestamps_available = True - tfr = _make_tfr(result=result, model=model, backend=backend) diarizer = MagicMock() diarizer.process.return_value = DiarizationRun(intervals, elapsed_seconds=0.1) - write = MagicMock() with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), + _cli_run(tmp_file=audio, result=result) as (_backend, write), patch( "local_transcriber.cli.load_speaker_diarizer", return_value=diarizer, ), - patch("local_transcriber.cli.write_transcript", write), ): out = runner.invoke(app, [str(audio), "--diarize"]) @@ -522,14 +429,8 @@ def test_cli_rejects_nonpositive_speaker_count(tmp_path): def test_cli_rejects_no_diarize_with_speakers_before_model_load(tmp_path): audio = tmp_path / "meeting.mp3" audio.write_bytes(b"fake") - load_model = MagicMock() - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch("local_transcriber.cli.load_model", load_model), - ): + with patch("local_transcriber.transcriber.get_backend") as get_backend: out = runner.invoke( app, [str(audio), "--no-diarize", "--speakers", "2"], @@ -537,34 +438,19 @@ def test_cli_rejects_no_diarize_with_speakers_before_model_load(tmp_path): assert out.exit_code == 2 assert "--no-diarize и --speakers несовместимы" in out.output - load_model.assert_not_called() + get_backend.assert_not_called() def test_cli_verbose_passes_on_segment_callback(tmp_path): audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - result = _make_result() - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, backend=backend) - mock_transcribe_file = MagicMock(return_value=tfr) - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", mock_transcribe_file), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(tmp_file=audio) as (backend, _write): runner.invoke(app, [str(audio), "--verbose"]) - call_kwargs = mock_transcribe_file.call_args[1] - assert call_kwargs["on_segment"] is not None - assert callable(call_kwargs["on_segment"]) + on_segment = backend.transcribe.call_args[0][3] + assert on_segment is not None + assert callable(on_segment) def test_cli_empty_speech_warning(tmp_path): @@ -572,8 +458,7 @@ def test_cli_empty_speech_warning(tmp_path): audio.write_bytes(b"fake") result = _make_result(segments=[]) - patches = _single_patches(result=result, tmp_file=audio) - with patches[0], patches[1], patches[2], patches[3], patches[4], patches[5]: + with _cli_run(tmp_file=audio, result=result): out = runner.invoke(app, [str(audio)]) assert out.exit_code == 0 @@ -583,27 +468,11 @@ def test_cli_empty_speech_warning(tmp_path): def test_cli_default_output_path(tmp_path): audio = tmp_path / "meeting.mp3" audio.write_bytes(b"fake") - mock_write = MagicMock() - result = _make_result() - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, backend=backend) - - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript", mock_write), - ): + with _cli_run(tmp_file=audio) as (_backend, write): runner.invoke(app, [str(audio)]) - written_path: Path = mock_write.call_args[0][1] + written_path: Path = write.call_args[0][1] assert written_path.name == "meeting-transcript.md" @@ -611,101 +480,46 @@ def test_cli_custom_output_path(tmp_path): audio = tmp_path / "meeting.mp3" audio.write_bytes(b"fake") out_file = tmp_path / "custom.md" - mock_write = MagicMock() - result = _make_result() - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, backend=backend) - - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript", mock_write), - ): + with _cli_run(tmp_file=audio) as (_backend, write): runner.invoke(app, [str(audio), "--output", str(out_file)]) - written_path: Path = mock_write.call_args[0][1] + written_path: Path = write.call_args[0][1] assert written_path == out_file def test_cli_passes_status_callback_to_transcribe(tmp_path): audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - result = _make_result() - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, backend=backend) - mock_transcribe_file = MagicMock(return_value=tfr) - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", mock_transcribe_file), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(tmp_file=audio) as (backend, _write): runner.invoke(app, [str(audio)]) - call_kwargs = mock_transcribe_file.call_args[1] - assert call_kwargs["on_status"] is not None - assert callable(call_kwargs["on_status"]) + on_status = backend.transcribe.call_args[0][4] + assert on_status is not None + assert callable(on_status) def test_cli_load_model_called_with_model_name(tmp_path): - """load_model receives model name from defaults, handles ensure internally.""" + """ensure_model_available получает имя модели из флага CLI.""" audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - result = _make_result() - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, backend=backend) - mock_load_model = MagicMock( - return_value=(model, "cpu", backend, "/models/large-v3") - ) - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch("local_transcriber.cli.load_model", mock_load_model), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(tmp_file=audio) as (backend, _write): runner.invoke(app, [str(audio), "--model", "large-v3"]) - assert mock_load_model.call_args[0][0] == "large-v3" + assert backend.ensure_model_available.call_args[0][0] == "large-v3" def test_cli_windows_cuda_diagnostic(tmp_path): """При проблеме драйвера Windows не предлагает установку Toolkit.""" audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - model = _make_model() backend = _make_backend() + backend.transcribe.side_effect = RuntimeError("CUDA error: no device") with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cuda"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cuda", backend, "/models/medium"), - ), - patch( - "local_transcriber.cli._transcribe_file", - side_effect=RuntimeError("CUDA error: no device"), - ), + _cli_run(tmp_file=audio, backend=backend), patch("local_transcriber.cli.sys") as mock_sys, ): mock_sys.platform = "win32" @@ -722,21 +536,11 @@ def test_cli_linux_cuda_error_no_windows_hint(tmp_path): """CUDA error on Linux does NOT print Windows-specific hint.""" audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - model = _make_model() backend = _make_backend() + backend.transcribe.side_effect = RuntimeError("CUDA error: no device") with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cuda"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cuda", backend, "/models/medium"), - ), - patch( - "local_transcriber.cli._transcribe_file", - side_effect=RuntimeError("CUDA error: no device"), - ), + _cli_run(tmp_file=audio, backend=backend), patch("local_transcriber.cli.sys") as mock_sys, ): mock_sys.platform = "linux" @@ -855,96 +659,14 @@ def test_unsupported_compute_type_hint_requires_cuda_context( assert "Переключение на CPU" not in out.output -def test_cli_device_fallback_warning(tmp_path): - """When auto-detected device differs from actual, show fallback warning.""" - audio = tmp_path / "test.mp3" - audio.write_bytes(b"fake") - result = _make_result(device_used="cpu") - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, actual_device="cpu", backend=backend) - - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cuda"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cuda", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), - ): - out = runner.invoke(app, [str(audio)]) - - assert "fallback" in out.output - - -def test_cli_strict_device_passed_to_transcribe(tmp_path): - """--device cuda passes strict_device=True; default auto passes False.""" - audio = tmp_path / "test.mp3" - audio.write_bytes(b"fake") - result = _make_result(device_used="cuda") - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, actual_device="cuda", backend=backend) - mock_transcribe_file = MagicMock(return_value=tfr) - - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cuda"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cuda", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", mock_transcribe_file), - patch("local_transcriber.cli.write_transcript"), - patch("local_transcriber.cli.get_gpu_name", return_value="RTX 3060"), - ): - runner.invoke(app, [str(audio), "--device", "cuda"]) - - assert mock_transcribe_file.call_args[1]["strict_device"] is True - - mock_transcribe_file.reset_mock() - result_cpu = _make_result(device_used="cpu") - tfr_cpu = _make_tfr(result=result_cpu, model=model, backend=backend) - mock_transcribe_file.return_value = tfr_cpu - - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", mock_transcribe_file), - patch("local_transcriber.cli.write_transcript"), - ): - runner.invoke(app, [str(audio)]) - - assert mock_transcribe_file.call_args[1]["strict_device"] is False - - def test_cli_keyboard_interrupt(tmp_path): """Ctrl+C → exit code 130, 'Прервано пользователем' in output.""" audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - model = _make_model() backend = _make_backend() + backend.transcribe.side_effect = KeyboardInterrupt - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", side_effect=KeyboardInterrupt), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(tmp_file=audio, backend=backend): out = runner.invoke(app, [str(audio)]) assert out.exit_code == 130 @@ -967,23 +689,10 @@ def test_cli_unexpected_error_verbose_traceback(tmp_path): """Unexpected error with --verbose → traceback shown.""" audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - model = _make_model() backend = _make_backend() + backend.transcribe.side_effect = RuntimeError("unexpected boom") - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch( - "local_transcriber.cli._transcribe_file", - side_effect=RuntimeError("unexpected boom"), - ), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(tmp_file=audio, backend=backend): out = runner.invoke(app, [str(audio), "--verbose"]) assert out.exit_code == 1 @@ -994,23 +703,10 @@ def test_cli_unexpected_error_no_verbose_hint(tmp_path): """Unexpected error without --verbose → hint to use --verbose.""" audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - model = _make_model() backend = _make_backend() + backend.transcribe.side_effect = RuntimeError("unexpected boom") - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch( - "local_transcriber.cli._transcribe_file", - side_effect=RuntimeError("unexpected boom"), - ), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(tmp_file=audio, backend=backend): out = runner.invoke(app, [str(audio)]) assert out.exit_code == 1 @@ -1022,31 +718,19 @@ def test_cli_unexpected_error_no_verbose_hint(tmp_path): def test_cli_batch_two_files(tmp_path): + """Два файла: одна загрузка модели, два распознавания.""" a = tmp_path / "a.mp3" b = tmp_path / "b.mp3" a.write_bytes(b"fake") b.write_bytes(b"fake") - result = _make_result() - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, backend=backend) - - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run() as (backend, _write): out = runner.invoke(app, [str(a), str(b)]) assert out.exit_code == 0 assert "2 обработано" in out.output + assert backend.create_model.call_count == 1 + assert backend.transcribe.call_count == 2 @pytest.mark.parametrize( @@ -1065,10 +749,7 @@ def test_cli_batch_reuses_one_diarizer_for_all_nonempty_files( duration=10.0, ) result.words = [Word(0.0, 0.5, "Раз"), Word(0.5, 1.0, "два")] - model = _make_model() - backend = _make_backend() - backend.word_timestamps_available = True - tfr = _make_tfr(result=result, model=model, backend=backend) + backend = _make_backend(result=result) diarizer = MagicMock() diarizer.process.return_value = DiarizationRun( intervals=[ @@ -1079,22 +760,11 @@ def test_cli_batch_reuses_one_diarizer_for_all_nonempty_files( ) with ( - patch("local_transcriber.cli.load_config", return_value=config), - patch( - "local_transcriber.cli.validate_input_file", - side_effect=lambda path: path, - ), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), + _cli_run(backend=backend, config=config) as (_backend, write), patch( "local_transcriber.cli.load_speaker_diarizer", return_value=diarizer, ) as load_diarizer, - patch("local_transcriber.cli.write_transcript") as write, ): out = runner.invoke(app, [str(first), str(second), *cli_args]) @@ -1117,10 +787,7 @@ def test_cli_batch_continues_after_diarization_error_and_exits_nonzero(tmp_path) duration=10.0, ) result.words = [Word(0.0, 0.5, "Раз"), Word(0.5, 1.0, "два")] - model = _make_model() - backend = _make_backend() - backend.word_timestamps_available = True - tfr = _make_tfr(result=result, model=model, backend=backend) + backend = _make_backend(result=result) diarizer = MagicMock() diarizer.process.side_effect = [ RuntimeError("boom"), @@ -1134,22 +801,11 @@ def test_cli_batch_continues_after_diarization_error_and_exits_nonzero(tmp_path) ] with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch( - "local_transcriber.cli.validate_input_file", - side_effect=lambda path: path, - ), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), + _cli_run(backend=backend) as (_backend, write), patch( "local_transcriber.cli.load_speaker_diarizer", return_value=diarizer, ), - patch("local_transcriber.cli.write_transcript") as write, ): out = runner.invoke(app, [str(first), str(second), "--diarize"]) @@ -1167,22 +823,7 @@ def test_cli_batch_skips_existing(tmp_path): b.write_bytes(b"fake") (tmp_path / "a-transcript.md").write_text("existing") - result = _make_result() - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, backend=backend) - - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(): out = runner.invoke(app, [str(a), str(b)]) assert out.exit_code == 0 @@ -1199,23 +840,18 @@ def test_cli_batch_all_skipped_no_model_load(tmp_path): b.write_bytes(b"fake") (tmp_path / "b-transcript.md").write_text("existing") - mock_load_model = MagicMock() - mock_load_diarizer = MagicMock() - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p), - patch("local_transcriber.cli.load_model", mock_load_model), - patch( - "local_transcriber.cli.load_speaker_diarizer", - mock_load_diarizer, - ), + _cli_run() as (backend, _write), + patch("local_transcriber.transcriber.get_backend") as get_backend, + patch("local_transcriber.cli.load_speaker_diarizer") as load_diarizer, ): out = runner.invoke(app, [str(a), str(b), "--diarize"]) assert out.exit_code == 0 - mock_load_model.assert_not_called() - mock_load_diarizer.assert_not_called() + assert "0 обработано, 2 пропущено" in out.output + get_backend.assert_not_called() + backend.create_model.assert_not_called() + load_diarizer.assert_not_called() def test_cli_batch_force_overwrites(tmp_path): @@ -1225,22 +861,7 @@ def test_cli_batch_force_overwrites(tmp_path): b = tmp_path / "b.mp3" b.write_bytes(b"fake") - result = _make_result() - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, backend=backend) - - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(): out = runner.invoke(app, [str(a), str(b), "--force"]) assert out.exit_code == 0 @@ -1253,33 +874,10 @@ def test_cli_batch_per_file_error(tmp_path): b = tmp_path / "b.mp3" a.write_bytes(b"fake") b.write_bytes(b"fake") - - result = _make_result() - model = _make_model() backend = _make_backend() - tfr = _make_tfr(result=result, model=model, backend=backend) - call_count = 0 + backend.transcribe.side_effect = [RuntimeError("oops"), _make_result()] - def transcribe_side_effect(**kwargs): - nonlocal call_count - call_count += 1 - if call_count == 1: - raise RuntimeError("oops") - return tfr - - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch( - "local_transcriber.cli._transcribe_file", side_effect=transcribe_side_effect - ), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(backend=backend): out = runner.invoke(app, [str(a), str(b)]) assert out.exit_code == 1 @@ -1293,29 +891,9 @@ def test_cli_batch_invalid_in_prescan(tmp_path): b = tmp_path / "b.mp3" # b doesn't exist - result = _make_result() - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, backend=backend) - - def validate_side_effect(p): - if not p.exists(): - raise FileNotFoundError(f"Файл не найден: {p}") - return p - with ( patch("local_transcriber.cli.load_config", return_value={}), - patch( - "local_transcriber.cli.validate_input_file", - side_effect=validate_side_effect, - ), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), + patch("local_transcriber.transcriber.get_backend", return_value=_make_backend()), ): out = runner.invoke(app, [str(a), str(b)]) @@ -1350,24 +928,12 @@ def test_cli_batch_empty_glob(tmp_path, monkeypatch): def test_cli_config_applied(tmp_path): audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - model = _make_model() - backend = _make_backend() - result = _make_result() - tfr = _make_tfr(result=result, model=model, backend=backend) - mock_load_model = MagicMock(return_value=(model, "cpu", backend, "/models/tiny")) - with ( - patch("local_transcriber.cli.load_config", return_value={"model": "tiny"}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch("local_transcriber.cli.load_model", mock_load_model), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(tmp_file=audio, config={"model": "tiny"}) as (backend, _write): runner.invoke(app, [str(audio)]) - # load_model receives model name from config - assert mock_load_model.call_args[0][0] == "tiny" + # ensure_model_available receives model name from config + backend.ensure_model_available.assert_called_once_with("tiny", "int8", ANY) @pytest.mark.parametrize( @@ -1385,26 +951,11 @@ def test_cli_resolves_diarization_priority( audio = tmp_path / "meeting.mp3" audio.write_bytes(b"fake") result = _make_result(segments=[]) - model = _make_model() - backend = _make_backend() - backend.word_timestamps_available = True - tfr = _make_tfr(result=result, model=model, backend=backend) load_diarizer = MagicMock() with ( - patch( - "local_transcriber.cli.load_config", - return_value={"diarize": config_value}, - ), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), + _cli_run(tmp_file=audio, result=result, config={"diarize": config_value}), patch("local_transcriber.cli.load_speaker_diarizer", load_diarizer), - patch("local_transcriber.cli.write_transcript"), ): out = runner.invoke(app, [str(audio), *cli_args]) @@ -1415,91 +966,29 @@ def test_cli_resolves_diarization_priority( def test_cli_config_overrides_auto_device(tmp_path): audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - model = _make_model() - backend = _make_backend() - result = _make_result(device_used="openvino-cpu") - tfr = _make_tfr( - result=result, - model=model, - actual_device="openvino-cpu", - backend=backend, - ) - mock_detect_device = MagicMock(return_value="openvino-cpu") - mock_load_model = MagicMock( - return_value=(model, "openvino-cpu", backend, "/models/medium") - ) - with ( - patch( - "local_transcriber.cli.load_config", - return_value={ - "device": "openvino-cpu", - "model": "medium", - "compute_type": "int8", - }, - ), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", mock_detect_device), - patch("local_transcriber.cli.load_model", mock_load_model), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run( + tmp_file=audio, + config={"device": "openvino-cpu", "model": "medium", "compute_type": "int8"}, + ) as (backend, _write): out = runner.invoke(app, [str(audio)]) assert out.exit_code == 0 - assert mock_detect_device.call_args_list[0].args == ("openvino-cpu",) - assert mock_load_model.call_args.args[:3] == ("medium", "openvino-cpu", "int8") + backend.ensure_model_available.assert_called_once_with("medium", "int8", ANY) + backend.create_model.assert_called_once_with( + "/models/medium", "openvino-cpu", "int8", cpu_threads=0 + ) def test_cli_cli_overrides_config(tmp_path): audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - model = _make_model() - backend = _make_backend() - result = _make_result() - tfr = _make_tfr(result=result, model=model, backend=backend) - mock_load_model = MagicMock(return_value=(model, "cpu", backend, "/models/small")) - with ( - patch("local_transcriber.cli.load_config", return_value={"model": "tiny"}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch("local_transcriber.cli.load_model", mock_load_model), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(tmp_file=audio, config={"model": "tiny"}) as (backend, _write): runner.invoke(app, [str(audio), "--model", "small"]) # CLI --model overrides config - assert mock_load_model.call_args[0][0] == "small" - - -def test_cli_batch_fallback_warning(tmp_path): - """Batch mode shows fallback warning when load_model falls back to CPU.""" - a = tmp_path / "a.mp3" - b = tmp_path / "b.mp3" - a.write_bytes(b"fake") - b.write_bytes(b"fake") - - result = _make_result(device_used="cpu") - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, actual_device="cpu", backend=backend) - - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p), - patch("local_transcriber.cli.detect_device", return_value="cuda"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), - ): - out = runner.invoke(app, [str(a), str(b)]) - - assert "fallback" in out.output + backend.ensure_model_available.assert_called_once_with("small", "int8", ANY) def test_cli_batch_empty_speech_warning(tmp_path): @@ -1511,24 +1000,10 @@ def test_cli_batch_empty_speech_warning(tmp_path): result_empty = _make_result(segments=[]) result_ok = _make_result() - model = _make_model() backend = _make_backend() - tfr_empty = _make_tfr(result=result_empty, model=model, backend=backend) - tfr_ok = _make_tfr(result=result_ok, model=model, backend=backend) + backend.transcribe.side_effect = [result_empty, result_ok] - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch( - "local_transcriber.cli._transcribe_file", side_effect=[tfr_empty, tfr_ok] - ), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(backend=backend): out = runner.invoke(app, [str(a), str(b)]) assert out.exit_code == 0 @@ -1536,209 +1011,86 @@ def test_cli_batch_empty_speech_warning(tmp_path): assert "2 обработано" in out.output -def test_cli_batch_midstream_fallback_warning(tmp_path): - """Batch mode shows warning when _transcribe_file falls back mid-stream.""" - a = tmp_path / "a.mp3" - b = tmp_path / "b.mp3" - a.write_bytes(b"fake") - b.write_bytes(b"fake") - - model_gpu = _make_model() - model_cpu = _make_model() - backend = _make_backend() - result = _make_result(device_used="cpu") - tfr_fallback = _make_tfr( - result=result, model=model_cpu, actual_device="cpu", backend=backend - ) - tfr_ok = _make_tfr( - result=result, model=model_cpu, actual_device="cpu", backend=backend - ) - - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p), - patch("local_transcriber.cli.detect_device", return_value="cuda"), - patch( - "local_transcriber.cli.load_model", - return_value=(model_gpu, "cuda", backend, "/models/medium"), - ), - patch( - "local_transcriber.cli._transcribe_file", side_effect=[tfr_fallback, tfr_ok] - ), - patch("local_transcriber.cli.write_transcript"), - ): - out = runner.invoke(app, [str(a), str(b)]) - - assert "fallback" in out.output - assert "2 обработано" in out.output - - -def test_cli_batch_model_loaded_once(tmp_path): - a = tmp_path / "a.mp3" - b = tmp_path / "b.mp3" - a.write_bytes(b"fake") - b.write_bytes(b"fake") - - result = _make_result() - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, backend=backend) - mock_load_model = MagicMock(return_value=(model, "cpu", backend, "/models/medium")) - - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch("local_transcriber.cli.load_model", mock_load_model), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), - ): - out = runner.invoke(app, [str(a), str(b)]) - - assert out.exit_code == 0 - mock_load_model.assert_called_once() - - -# === _format_device_info tests === - - -def test_format_device_info_openvino_gpu(): - with patch( - "local_transcriber.cli.get_intel_gpu_name", - return_value="Intel(R) Arc(TM) 140T GPU", - ): - assert ( - _format_device_info("openvino-gpu") - == "OpenVINO (Intel(R) Arc(TM) 140T GPU)" - ) - - -def test_format_device_info_openvino_gpu_no_name(): - """get_intel_gpu_name вернул None → fallback на 'Intel GPU'.""" - with patch("local_transcriber.cli.get_intel_gpu_name", return_value=None): - assert _format_device_info("openvino-gpu") == "OpenVINO (Intel GPU)" - - -def test_format_device_info_openvino_cpu(): - assert _format_device_info("openvino-cpu") == "OpenVINO (CPU)" - - -def test_format_device_info_openvino_legacy(): - """Обратная совместимость: 'openvino' → OpenVINO (CPU).""" - assert _format_device_info("openvino") == "OpenVINO (CPU)" - - -def test_format_device_info_distinguishes_onnx_from_faster_whisper_cpu(): - assert _format_device_info("onnx") == "ONNX (CPU)" - assert _format_device_info("cpu") == "CPU" - - -def test_format_device_info_cuda(): - with patch("local_transcriber.cli.get_gpu_name", return_value="RTX 4090"): - assert _format_device_info("cuda") == "CUDA (RTX 4090)" - - # === CLI with --device openvino-gpu === def test_cli_openvino_gpu_happy_path(tmp_path): audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - result = _make_result(device_used="openvino-gpu") + backend = _make_backend(actual_ov_device="GPU") - patches = _single_patches( - result=result, tmp_file=audio, actual_device="openvino-gpu" - ) - with patches[0], patches[1], patches[2], patches[3], patches[4], patches[5]: - with patch( - "local_transcriber.cli.get_intel_gpu_name", return_value="Intel Arc 140T" - ): - out = runner.invoke(app, [str(audio), "--device", "openvino-gpu"]) + with ( + _cli_run(tmp_file=audio, backend=backend), + patch( + "local_transcriber.transcriber.get_intel_gpu_name", + return_value="Intel Arc 140T", + ), + ): + out = runner.invoke(app, [str(audio), "--device", "openvino-gpu"]) assert out.exit_code == 0 def test_cli_openvino_alias_resolves_to_gpu(tmp_path): - """--device openvino резолвится через detect_device в openvino-gpu.""" + """--device openvino резолвится в openvino-gpu, шапка берёт имя GPU из данных module.""" audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - result = _make_result(device_used="openvino-gpu") - model = _make_model() - backend = _make_backend() - tfr = _make_tfr( - result=result, model=model, actual_device="openvino-gpu", backend=backend - ) - mock_load_model = MagicMock( - return_value=(model, "openvino-gpu", backend, "/models/medium") - ) + backend = _make_backend(actual_ov_device="GPU") with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="openvino-gpu"), - patch("local_transcriber.cli.load_model", mock_load_model), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), + _cli_run(tmp_file=audio, backend=backend, write=False), + patch("local_transcriber.transcriber.detect_device", return_value="openvino-gpu"), patch( - "local_transcriber.cli.get_intel_gpu_name", return_value="Intel Arc 140T" + "local_transcriber.transcriber.get_intel_gpu_name", + return_value="Intel Arc 140T", ), ): out = runner.invoke(app, [str(audio), "--device", "openvino"]) + content = (tmp_path / "test-transcript.md").read_text(encoding="utf-8") assert out.exit_code == 0 - # detect_device("openvino") resolved to "openvino-gpu", load_model receives it - assert mock_load_model.call_args[0][1] == "openvino-gpu" + assert backend.create_model.call_args[0][1] == "openvino-gpu" + assert "- **Устройство**: OpenVINO (Intel Arc 140T)" in content -# === --threads === +def test_cli_reports_openvino_cpu_when_gpu_was_requested(tmp_path): + """OpenVINO, выбравший CPU вместо запрошенного GPU, виден пользователю.""" + audio = tmp_path / "test.mp3" + audio.write_bytes(b"fake") + backend = _make_backend(actual_ov_device="CPU") + + with ( + _cli_run(tmp_file=audio, backend=backend), + patch("local_transcriber.transcriber.detect_device", return_value="openvino-gpu"), + ): + out = runner.invoke(app, [str(audio), "--device", "openvino"]) + + assert out.exit_code == 0 + assert "Запрошено openvino, используется openvino-cpu" in out.output def test_cli_threads_passed_to_load_model(tmp_path): - """--threads передаётся в load_model как cpu_threads.""" + """--threads передаётся в create_model как cpu_threads.""" audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - result = _make_result() - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, backend=backend) - mock_load_model = MagicMock(return_value=(model, "cpu", backend, "/models/medium")) - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch("local_transcriber.cli.load_model", mock_load_model), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(tmp_file=audio) as (backend, _write): out = runner.invoke(app, [str(audio), "--threads", "8"]) assert out.exit_code == 0 - assert mock_load_model.call_args.kwargs["cpu_threads"] == 8 + assert backend.create_model.call_args.kwargs["cpu_threads"] == 8 def test_cli_threads_default_zero(tmp_path): - """Без --threads load_model получает cpu_threads=0.""" + """Без --threads create_model получает cpu_threads=0.""" audio = tmp_path / "test.mp3" audio.write_bytes(b"fake") - result = _make_result() - model = _make_model() - backend = _make_backend() - tfr = _make_tfr(result=result, model=model, backend=backend) - mock_load_model = MagicMock(return_value=(model, "cpu", backend, "/models/medium")) - with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", return_value=audio), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch("local_transcriber.cli.load_model", mock_load_model), - patch("local_transcriber.cli._transcribe_file", return_value=tfr), - patch("local_transcriber.cli.write_transcript"), - ): + with _cli_run(tmp_file=audio) as (backend, _write): out = runner.invoke(app, [str(audio)]) assert out.exit_code == 0 - assert mock_load_model.call_args.kwargs["cpu_threads"] == 0 + assert backend.create_model.call_args.kwargs["cpu_threads"] == 0 def test_cli_threads_negative_rejected(tmp_path): @@ -1861,14 +1213,8 @@ def test_cli_tail_gap_quality_warning_single(tmp_path): duration=600.0, ) - patches = _single_patches(result=result, tmp_file=audio) with ( - patches[0], - patches[1], - patches[2], - patches[3], - patches[4], - patches[5], + _cli_run(tmp_file=audio, result=result), patch("local_transcriber.cli.console", Console(stderr=True, width=1000)), ): out = runner.invoke(app, [str(audio)]) @@ -1893,14 +1239,8 @@ def test_cli_repetition_quality_warning_single(tmp_path): duration=60.0, ) - patches = _single_patches(result=result, tmp_file=audio) with ( - patches[0], - patches[1], - patches[2], - patches[3], - patches[4], - patches[5], + _cli_run(tmp_file=audio, result=result), patch("local_transcriber.cli.console", Console(stderr=True, width=1000)), ): out = runner.invoke(app, [str(audio)]) @@ -1923,21 +1263,11 @@ def test_cli_quality_warning_batch_includes_file_name(tmp_path): duration=600.0, ) result_ok = _make_result() - model = _make_model() backend = _make_backend() - tfr_warn = _make_tfr(result=result_warn, model=model, backend=backend) - tfr_ok = _make_tfr(result=result_ok, model=model, backend=backend) + backend.transcribe.side_effect = [result_warn, result_ok] with ( - patch("local_transcriber.cli.load_config", return_value={}), - patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p), - patch("local_transcriber.cli.detect_device", return_value="cpu"), - patch( - "local_transcriber.cli.load_model", - return_value=(model, "cpu", backend, "/models/medium"), - ), - patch("local_transcriber.cli._transcribe_file", side_effect=[tfr_warn, tfr_ok]), - patch("local_transcriber.cli.write_transcript"), + _cli_run(backend=backend), patch("local_transcriber.cli.console", Console(stderr=True, width=1000)), ): out = runner.invoke(app, [str(a), str(b)]) @@ -1971,14 +1301,8 @@ def test_cli_repetition_quality_warning_truncates_after_three_blocks(tmp_path): duration=90.0, ) - patches = _single_patches(result=result, tmp_file=audio) with ( - patches[0], - patches[1], - patches[2], - patches[3], - patches[4], - patches[5], + _cli_run(tmp_file=audio, result=result), patch("local_transcriber.cli.console", Console(stderr=True, width=1000)), ): out = runner.invoke(app, [str(audio)]) @@ -1995,8 +1319,7 @@ def test_cli_default_result_has_no_quality_warnings(tmp_path): audio = tmp_path / "normal.mp3" audio.write_bytes(b"fake") - patches = _single_patches(tmp_file=audio) - with patches[0], patches[1], patches[2], patches[3], patches[4], patches[5]: + with _cli_run(tmp_file=audio): out = runner.invoke(app, [str(audio)]) assert out.exit_code == 0 diff --git a/tests/test_config.py b/tests/test_config.py index 2408a01..fcecbcf 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -3,7 +3,6 @@ from unittest.mock import patch import pytest from local_transcriber.config import ( - apply_device_defaults, find_config_file, load_config, resolve_defaults, @@ -136,54 +135,6 @@ def test_resolve_defaults_diarize_priority(cli_value, config_value, expected): assert resolve_defaults(cli, config)["diarize"] is expected -def test_apply_device_defaults_cuda(): - defaults = {"model": "medium", "language": "ru", "device": "auto", "compute_type": "float32"} - cli = {"model": None, "language": None, "device": None, "compute_type": None} - result = apply_device_defaults(defaults, "cuda", cli, {}) - assert result["model"] == "medium" - assert result["compute_type"] == "float16" - - -def test_apply_device_defaults_cpu(): - defaults = {"model": "medium", "language": "ru", "device": "auto", "compute_type": "float32"} - cli = {"model": None, "language": None, "device": None, "compute_type": None} - result = apply_device_defaults(defaults, "cpu", cli, {}) - assert result["model"] == "medium" - assert result["compute_type"] == "float32" - - -def test_apply_device_defaults_onnx_uses_readable_model(): - defaults = { - "model": "medium", - "language": "ru", - "device": "auto", - "compute_type": "float32", - } - cli = {"model": None, "language": None, "device": None, "compute_type": None} - - result = apply_device_defaults(defaults, "onnx", cli, {}) - - assert result["model"] == "gigaam-v3-e2e-rnnt" - assert result["compute_type"] == "int8" - - -def test_apply_device_defaults_cli_overrides(): - defaults = {"model": "large-v3", "language": "ru", "device": "auto", "compute_type": "int8"} - cli = {"model": "large-v3", "language": None, "device": None, "compute_type": "int8"} - result = apply_device_defaults(defaults, "cuda", cli, {}) - assert result["model"] == "large-v3" - assert result["compute_type"] == "int8" - - -def test_apply_device_defaults_config_overrides(): - defaults = {"model": "small", "language": "ru", "device": "auto", "compute_type": "int8"} - cli = {"model": None, "language": None, "device": None, "compute_type": None} - config = {"model": "small", "compute_type": "int8"} - result = apply_device_defaults(defaults, "cuda", cli, config) - assert result["model"] == "small" - assert result["compute_type"] == "int8" - - def test_load_config_openvino_device(tmp_path): config = tmp_path / "config.toml" config.write_text('device = "openvino"\n') @@ -203,27 +154,3 @@ def test_load_config_openvino_cpu_device(tmp_path): config.write_text('device = "openvino-cpu"\n') result = load_config(config) assert result == {"device": "openvino-cpu"} - - -def test_apply_device_defaults_openvino(): - defaults = {"model": "medium", "language": "ru", "device": "auto", "compute_type": "float32"} - cli = {"model": None, "language": None, "device": None, "compute_type": None} - result = apply_device_defaults(defaults, "openvino", cli, {}) - assert result["model"] == "medium" - assert result["compute_type"] == "int8" - - -def test_apply_device_defaults_openvino_gpu(): - defaults = {"model": "medium", "language": "ru", "device": "auto", "compute_type": "float32"} - cli = {"model": None, "language": None, "device": None, "compute_type": None} - result = apply_device_defaults(defaults, "openvino-gpu", cli, {}) - assert result["model"] == "medium" - assert result["compute_type"] == "int8" - - -def test_apply_device_defaults_openvino_cpu(): - defaults = {"model": "medium", "language": "ru", "device": "auto", "compute_type": "float32"} - cli = {"model": None, "language": None, "device": None, "compute_type": None} - result = apply_device_defaults(defaults, "openvino-cpu", cli, {}) - assert result["model"] == "medium" - assert result["compute_type"] == "int8" diff --git a/tests/test_onnx_asr.py b/tests/test_onnx_asr.py index 5ab9aaf..05486df 100644 --- a/tests/test_onnx_asr.py +++ b/tests/test_onnx_asr.py @@ -706,7 +706,6 @@ class TestRuntimeInfo: info = backend.runtime_info() - assert info["engine"] == "onnx-asr" assert info["onnxruntime"] assert info["onnx_asr"] assert info["available_providers"] == "CoreMLExecutionProvider, CPUExecutionProvider" diff --git a/tests/test_transcriber.py b/tests/test_transcriber.py index 31a7a45..6551e70 100644 --- a/tests/test_transcriber.py +++ b/tests/test_transcriber.py @@ -8,10 +8,8 @@ from local_transcriber.transcriber import ( Segment, Transcriber, TranscribeResult, - _transcribe_file, cuda_error_hint, ensure_model_available, - load_model, transcribe, ) from local_transcriber.types import WordTimestampsUnavailableError @@ -297,101 +295,6 @@ def test_transcribe_strict_cuda_error_during_transcription(mock_get_backend): ) -# === load_model() tests === - - -@patch("local_transcriber.transcriber.get_backend") -def test_load_model_cuda_fallback(mock_get_backend): - cuda_backend = _make_backend(create_model_error=RuntimeError("CUDA out of memory")) - cpu_model = MagicMock() - cpu_backend = _make_backend(model=cpu_model, model_path="/mock/cpu/model") - - def backend_for_device(device, **kwargs): - return cuda_backend if device == "cuda" else cpu_backend - - mock_get_backend.side_effect = backend_for_device - - with pytest.warns(UserWarning, match="Переключение на CPU"): - model, actual_device, backend, model_path = load_model("tiny", "cuda", "int8") - - assert actual_device == "cpu" - assert model is cpu_model - - -@patch("local_transcriber.transcriber.get_backend") -def test_load_model_strict_raises(mock_get_backend): - backend = _make_backend(create_model_error=RuntimeError("CUDA out of memory")) - mock_get_backend.return_value = backend - - with pytest.raises(RuntimeError, match="CUDA out of memory"): - load_model("tiny", "cuda", "int8", strict_device=True) - - -@patch("local_transcriber.transcriber.get_backend") -def test_load_model_returns_backend_and_path(mock_get_backend): - backend = _make_backend(model_path="/mock/model/path") - mock_get_backend.return_value = backend - - model, actual_device, returned_backend, model_path = load_model( - "tiny", "cpu", "int8" - ) - - assert returned_backend is backend - assert model_path == "/mock/model/path" - assert actual_device == "cpu" - - -@patch("local_transcriber.transcriber.get_backend") -def test_load_model_passes_cpu_threads_to_backend(mock_get_backend): - backend = _make_backend(model_path="/mock/model/path") - mock_get_backend.return_value = backend - - load_model("tiny", "cpu", "int8", cpu_threads=8) - - assert backend.create_model.call_args.kwargs["cpu_threads"] == 8 - - -@patch("local_transcriber.transcriber.get_backend") -def test_load_model_fallback_preserves_cpu_threads(mock_get_backend): - cuda_backend = _make_backend(create_model_error=RuntimeError("CUDA out of memory")) - cpu_model = MagicMock() - cpu_backend = _make_backend(model=cpu_model, model_path="/mock/cpu/model") - - def backend_for_device(device, **kwargs): - return cuda_backend if device == "cuda" else cpu_backend - - mock_get_backend.side_effect = backend_for_device - - with pytest.warns(UserWarning, match="Переключение на CPU"): - load_model("tiny", "cuda", "int8", cpu_threads=6) - - assert cuda_backend.create_model.call_args.kwargs["cpu_threads"] == 6 - assert cpu_backend.create_model.call_args.kwargs["cpu_threads"] == 6 - - -# === _transcribe_file() tests === - - -def test__transcribe_file_basic(): - result_data = _make_result(count=2) - backend = _make_backend(transcribe_result=result_data) - - tfr = _transcribe_file( - model=MagicMock(), - actual_device="cpu", - backend=backend, - model_path="/mock/model", - file_path=Path("test.mp3"), - model_name="tiny", - compute_type="int8", - ) - - assert len(tfr.result.segments) == 2 - assert tfr.actual_device == "cpu" - assert tfr.backend is backend - assert tfr.model_path == "/mock/model" - - # === ensure_model_available() tests (через FasterWhisperBackend) === @@ -494,202 +397,6 @@ def test_ensure_model_available_rejects_incomplete_local_directory(tmp_path): ensure_model_available(str(model_dir)) -# === Cross-backend fallback (openvino → cpu) === - - -@patch("local_transcriber.transcriber.get_backend") -def test_load_model_openvino_gpu_fallback_to_cpu(mock_get_backend): - """OpenVINO GPU ошибка при init → fallback на CPU (FasterWhisper).""" - ov_backend = _make_backend( - create_model_error=RuntimeError("OpenVINO model load failed"), - ) - cpu_model = MagicMock() - cpu_backend = _make_backend(model=cpu_model, model_path="/mock/cpu/model") - - def backend_for_device(device, **kwargs): - return ov_backend if device == "openvino-gpu" else cpu_backend - - mock_get_backend.side_effect = backend_for_device - - with pytest.warns(UserWarning, match="Переключение на CPU"): - model, actual_device, backend, model_path = load_model( - "medium", - "openvino-gpu", - "fp16", - ) - - assert actual_device == "cpu" - assert model is cpu_model - assert backend is cpu_backend - assert model_path == "/mock/cpu/model" - - -@patch("local_transcriber.transcriber.get_backend") -def test_load_model_openvino_cpu_fallback_to_cpu(mock_get_backend): - """OpenVINO CPU ошибка при init → fallback на CPU (FasterWhisper).""" - ov_backend = _make_backend( - create_model_error=RuntimeError("OpenVINO model load failed"), - ) - cpu_model = MagicMock() - cpu_backend = _make_backend(model=cpu_model, model_path="/mock/cpu/model") - - def backend_for_device(device, **kwargs): - return ov_backend if device == "openvino-cpu" else cpu_backend - - mock_get_backend.side_effect = backend_for_device - - with pytest.warns(UserWarning, match="Переключение на CPU"): - model, actual_device, backend, model_path = load_model( - "medium", - "openvino-cpu", - "int8", - ) - - assert actual_device == "cpu" - assert model is cpu_model - assert backend is cpu_backend - assert model_path == "/mock/cpu/model" - - -@patch("local_transcriber.transcriber.get_backend") -def test_transcribe_file_openvino_gpu_midstream_fallback(mock_get_backend): - """OpenVINO GPU ошибка при транскрипции → fallback на CPU.""" - ov_backend = _make_backend( - transcribe_error=RuntimeError("OpenVINO inference error"), - ) - cpu_backend = _make_backend( - transcribe_result=_make_result(count=2, device_used="cpu"), - model_path="/mock/cpu/model", - ) - - def backend_for_device(device, **kwargs): - return ov_backend if device.startswith("openvino") else cpu_backend - - mock_get_backend.side_effect = backend_for_device - - with pytest.warns(UserWarning, match="Переключение на CPU"): - tfr = _transcribe_file( - model=MagicMock(), - actual_device="openvino-gpu", - backend=ov_backend, - model_path="/mock/ov/model", - file_path=Path("test.mp3"), - model_name="medium", - compute_type="fp16", - ) - - assert tfr.actual_device == "cpu" - assert tfr.backend is cpu_backend - assert tfr.model_path == "/mock/cpu/model" - - -@patch("local_transcriber.transcriber.get_backend") -def test_transcribe_file_does_not_fallback_for_missing_word_timestamps( - mock_get_backend, -): - ov_backend = _make_backend( - transcribe_error=WordTimestampsUnavailableError("нет таймкодов"), - ) - - with pytest.raises(WordTimestampsUnavailableError, match="нет таймкодов"): - _transcribe_file( - model=MagicMock(), - actual_device="openvino-gpu", - backend=ov_backend, - model_path="/mock/ov/model", - file_path=Path("test.mp3"), - model_name="medium", - compute_type="fp16", - ) - - mock_get_backend.assert_not_called() - - -@patch("local_transcriber.transcriber.get_backend") -def test_transcribe_file_midstream_fallback_preserves_cpu_threads(mock_get_backend): - ov_backend = _make_backend( - transcribe_error=RuntimeError("OpenVINO inference error"), - ) - cpu_backend = _make_backend( - transcribe_result=_make_result(count=2, device_used="cpu"), - model_path="/mock/cpu/model", - ) - - def backend_for_device(device, **kwargs): - return ov_backend if device.startswith("openvino") else cpu_backend - - mock_get_backend.side_effect = backend_for_device - - with pytest.warns(UserWarning, match="Переключение на CPU"): - _transcribe_file( - model=MagicMock(), - actual_device="openvino-gpu", - backend=ov_backend, - model_path="/mock/ov/model", - file_path=Path("test.mp3"), - model_name="medium", - compute_type="fp16", - cpu_threads=6, - ) - - assert cpu_backend.create_model.call_args.kwargs["cpu_threads"] == 6 - - -@patch("local_transcriber.transcriber.get_backend") -def test_openvino_gpu_strict_device_no_fallback(mock_get_backend): - """strict_device=True + OpenVINO GPU ошибка → raise.""" - backend = _make_backend( - create_model_error=RuntimeError("OpenVINO model load failed"), - ) - mock_get_backend.return_value = backend - - with pytest.raises(RuntimeError, match="OpenVINO"): - load_model("medium", "openvino-gpu", "fp16", strict_device=True) - - -@patch("local_transcriber.transcriber.get_backend") -def test_openvino_runtime_error_triggers_fallback(mock_get_backend): - """Любой RuntimeError от OpenVINO бэкенда → fallback.""" - ov_backend = _make_backend( - create_model_error=RuntimeError("Exception from src/inference/..."), - ) - cpu_backend = _make_backend(model_path="/mock/cpu/model") - - def backend_for_device(device, **kwargs): - return ov_backend if device.startswith("openvino") else cpu_backend - - mock_get_backend.side_effect = backend_for_device - - with pytest.warns(UserWarning, match="Переключение на CPU"): - _, actual_device, _, _ = load_model("medium", "openvino-cpu", "int8") - - assert actual_device == "cpu" - - -@patch("local_transcriber.transcriber.get_backend") -def test_load_model_resolves_openvino_gpu_device(mock_get_backend): - """load_model обновляет actual_device по backend.actual_ov_device.""" - backend = _make_backend() - backend.actual_ov_device = "GPU" - mock_get_backend.return_value = backend - - _, actual_device, _, _ = load_model("medium", "openvino-cpu", "int8") - - assert actual_device == "openvino-gpu" - - -@patch("local_transcriber.transcriber.get_backend") -def test_load_model_resolves_openvino_cpu_device(mock_get_backend): - """load_model обновляет actual_device по backend.actual_ov_device.""" - backend = _make_backend() - backend.actual_ov_device = "CPU" - mock_get_backend.return_value = backend - - _, actual_device, _, _ = load_model("medium", "openvino-gpu", "fp16") - - assert actual_device == "openvino-cpu" - - def test_ensure_model_available_openvino_default_compute_type(): """ensure_model_available(device='openvino-cpu') без compute_type не падает.""" from local_transcriber.backends.openvino import OpenVINOBackend @@ -775,7 +482,11 @@ def test_run_falls_back_to_cpu_at_load_and_keeps_state_for_next_file( run = Transcriber( ExecutionRequest( - device="cuda", model="tiny", compute_type="int8", strict_device=False + device="cuda", + model="tiny", + compute_type="int8", + strict_device=False, + cpu_threads=3, ) ) with pytest.warns(UserWarning, match="Переключение на CPU"): @@ -785,6 +496,7 @@ def test_run_falls_back_to_cpu_at_load_and_keeps_state_for_next_file( assert first.device_used == "cpu" assert second.device_used == "cpu" assert cpu_backend.create_model.call_count == 1 + assert cpu_backend.create_model.call_args.kwargs["cpu_threads"] == 3 assert run.execution.resolved_device == "cuda" assert run.execution.device == "cpu" assert run.execution.engine == "faster-whisper" @@ -961,3 +673,79 @@ def test_public_transcribe_uses_the_same_execution_module(mock_get_backend): ) assert result.device_used == "onnx" assert len(result.segments) == 1 + + +@pytest.mark.parametrize( + ("device", "ov_device", "gpu_name", "intel_name", "expected"), + [ + ("cpu", None, None, None, "CPU"), + ("onnx", None, "RTX 3060", None, "ONNX (CPU)"), + ("cuda", None, "RTX 3060", None, "CUDA (RTX 3060)"), + ("cuda", None, None, None, "CUDA (Unknown GPU)"), + ("openvino-gpu", "GPU", None, "Intel Arc 140T", "OpenVINO (Intel Arc 140T)"), + ("openvino-gpu", "GPU", None, None, "OpenVINO (Intel GPU)"), + ("openvino-cpu", "CPU", None, None, "OpenVINO (CPU)"), + ], +) +@patch("local_transcriber.transcriber.get_backend") +def test_run_description_names_hardware_only_from_driver_data( + mock_get_backend, device, ov_device, gpu_name, intel_name, expected +): + """Строка исполнения для шапки: ONNX не выдаётся за CPU faster-whisper, GPU — по драйверу.""" + backend = _make_run_backend() + backend.actual_ov_device = ov_device + mock_get_backend.return_value = backend + + with ( + patch("local_transcriber.transcriber.get_gpu_name", return_value=gpu_name), + patch("local_transcriber.transcriber.get_intel_gpu_name", return_value=intel_name), + ): + info = Transcriber(ExecutionRequest(device=device, model="m")).prepare() + + assert info.description == expected + + +@patch("local_transcriber.transcriber.get_backend") +def test_run_openvino_runtime_error_falls_back_when_not_strict(mock_get_backend): + """Любой RuntimeError OpenVINO считается ошибкой движка: не-strict запуск уходит на CPU.""" + ov_backend = _make_run_backend( + create_model_error=RuntimeError("Exception from src/inference/src/core.cpp") + ) + cpu_backend = _make_run_backend() + mock_get_backend.side_effect = lambda device, **_: ( + ov_backend if device.startswith("openvino") else cpu_backend + ) + + run = Transcriber( + ExecutionRequest(device="openvino-gpu", model="medium", strict_device=False) + ) + with pytest.warns(UserWarning, match="Переключение на CPU"): + result = run.transcribe(Path("a.mp3")) + + assert result.device_used == "cpu" + assert run.execution.engine == "faster-whisper" + + +@patch("local_transcriber.transcriber.get_backend") +def test_run_openvino_strict_raises_without_fallback(mock_get_backend): + mock_get_backend.return_value = _make_run_backend( + create_model_error=RuntimeError("GPU plugin failed") + ) + + with pytest.raises(RuntimeError, match="GPU plugin"): + Transcriber(ExecutionRequest(device="openvino-gpu", model="medium")).prepare() + + assert mock_get_backend.call_count == 1 + + +@patch("local_transcriber.transcriber.get_backend") +def test_run_reports_openvino_gpu_chosen_for_cpu_request(mock_get_backend): + """Фактическое устройство OpenVINO важнее запрошенного и в обратную сторону.""" + backend = _make_run_backend() + backend.actual_ov_device = "GPU" + mock_get_backend.return_value = backend + + info = Transcriber(ExecutionRequest(device="openvino-cpu", model="medium")).prepare() + + assert info.resolved_device == "openvino-cpu" + assert info.device == "openvino-gpu"