test_diagnostics.py

from __future__ import annotations

from pathlib import Path

from app import schemas
from app.diagnostics import NEVER_INCLUDED, build_diagnostics_bundle, redact_home_path

TEST_RAM_GB = 16.0
TEST_LOGICAL_CPU_COUNT = 8
TEST_EFFECTIVE_CPU_COUNT = 8.0
TEST_GATEWAY_PORT = 8765
TEST_UPTIME_SECONDS = 42
TEST_LATENCY_MS = 500
WHISPER_CPP_ENGINE = "whisper.cpp"


def _status(paths: schemas.PathStatus) -> schemas.AdminStatusResponse:
    return schemas.AdminStatusResponse(
        version="0.2.0",
        engine=schemas.EngineStatus(
            id=WHISPER_CPP_ENGINE,
            name="whisper.cpp:ggml-base.en.bin",
            ready=True,
        ),
        system=schemas.SystemStatus(
            os="Darwin",
            arch="arm64",
            chip="Apple M2",
            ram_gb=TEST_RAM_GB,
            is_apple_silicon=True,
            logical_cpus=TEST_LOGICAL_CPU_COUNT,
            effective_cpus=TEST_EFFECTIVE_CPU_COUNT,
            containerized=False,
            accelerators=["CPU", "Metal/Core ML"],
            cpu_features=[],
        ),
        dependencies=[
            schemas.DependencyStatus(name="FFmpeg", available=True, path="/usr/bin/ffmpeg")
        ],
        paths=paths,
        bind_host="0.0.0.0",
        port=TEST_GATEWAY_PORT,
        setup=schemas.SetupChecklist(
            token_configured=True,
            ffmpeg_available=True,
            engine_binary_available=True,
            model_installed=True,
            engine_ready=True,
        ),
        metrics=schemas.OperationalMetricsStatus(
            uptime_seconds=TEST_UPTIME_SECONDS,
            queue_depth=0,
            active_transcriptions=0,
            concurrency_limit=1,
            successful_transcriptions=3,
            failed_transcriptions=0,
            rejected_transcriptions=0,
            average_latency_ms=TEST_LATENCY_MS,
            last_latency_ms=TEST_LATENCY_MS,
        ),
        readiness=schemas.ReadinessStatus(
            probe_age_seconds=1.0, warmup_state="complete", warmed_bytes=0
        ),
    )


def _config() -> schemas.ConfigResponse:
    return schemas.ConfigResponse(engine="auto", available_engines=["auto", WHISPER_CPP_ENGINE])


def test_redact_home_path_replaces_home_prefix() -> None:
    home = str(Path.home())
    assert redact_home_path(f"{home}/.local/share/vocagateway") == "~/.local/share/vocagateway"
    assert redact_home_path(home) == "~"


def test_redact_home_path_leaves_other_path_aa() -> None:
    assert redact_home_path("/data/models") == "/data/models"


def test_build_diagnostics_bundle_redacts_p_aaa() -> None:
    home = str(Path.home())
    status = _status(
        schemas.PathStatus(
            data_dir=f"{home}/.local/share/vocagateway",
            models_dir=f"{home}/.local/share/vocagateway/models",
            config_file=f"{home}/.config/vocagateway/config.json",
            token_file="~/.config/vocagateway/token",
        )
    )
    bundle = build_diagnostics_bundle(status, _config())
    assert (
        bundle.paths.data_dir,
        bundle.paths.models_dir,
        bundle.paths.config_file,
    ) == (
        "~/.local/share/vocagateway",
        "~/.local/share/vocagateway/models",
        "~/.config/vocagateway/config.json",
    )
    assert bundle.never_included == list(NEVER_INCLUDED)
    dumped = bundle.model_dump_json()
    assert home not in dumped
    assert "Bearer" not in dumped


def test_build_diagnostics_bundle_redacts_c_a8a1b() -> None:
    """whisper_model/whisperkit_model/faster_whisper_model are absolute paths
    (set via `str(path)` in EngineManager.select_model), unlike
    moonshine_model/sherpa_model/mlx_audio_model, which are opaque catalog
    ids — both must survive the bundle, but only the paths get redacted."""
    home = str(Path.home())
    status = _status(
        schemas.PathStatus(
            data_dir=f"{home}/.local/share/vocagateway",
            models_dir=f"{home}/.local/share/vocagateway/models",
            config_file=f"{home}/.config/vocagateway/config.json",
            token_file="~/.config/vocagateway/token",
        )
    )
    config = schemas.ConfigResponse(
        engine=WHISPER_CPP_ENGINE,
        available_engines=["auto", WHISPER_CPP_ENGINE],
        whisper_model=f"{home}/.local/share/vocagateway/models/whisper.cpp/ggml-base.en.bin",
        whisperkit_model=f"{home}/.local/share/vocagateway/models/whisperkit/openai_whisper-tiny",
        faster_whisper_model=f"{home}/.local/share/vocagateway/models/faster-whisper/tiny.en",
        sherpa_model="sherpa-onnx:sensevoice-small-int8",
    )

    bundle = build_diagnostics_bundle(status, config)

    assert (
        bundle.config.whisper_model
        == "~/.local/share/vocagateway/models/whisper.cpp/ggml-base.en.bin"
    )
    assert (
        bundle.config.whisperkit_model
        == "~/.local/share/vocagateway/models/whisperkit/openai_whisper-tiny"
    )
    assert (
        bundle.config.faster_whisper_model
        == "~/.local/share/vocagateway/models/faster-whisper/tiny.en"
    )
    assert bundle.config.sherpa_model == "sherpa-onnx:sensevoice-small-int8"
    assert home not in bundle.model_dump_json()