path: root/lib/project/tests/test_pipeline.py

import json

import numpy as np
import pytest
import soundfile as sf

from voiceforge import pipeline
from voiceforge.audio import loudness, measure
from voiceforge.config import resolve


@pytest.mark.parametrize("kind", ["missing", "input-extension", "output-extension"])
def test_invalid_paths_fail_before_tool_setup(wav, tmp_path, bypass, progress, monkeypatch, kind):
    source = wav(name="input.txt" if kind == "input-extension" else "input.wav")
    if kind == "missing":
        source = tmp_path / "missing.wav"
    target = tmp_path / ("master.mp3" if kind == "output-extension" else "master.wav")
    monkeypatch.setattr(pipeline, "ensure_ffmpeg", lambda: pytest.fail("Invalid paths must fail preflight"))
    with pytest.raises(ValueError, match="WAV file|.wav extension"):
        pipeline.process(source, target, bypass, progress)
    assert not target.exists()


@pytest.mark.parametrize("alias", ["same", "symlink", "hardlink"])
@pytest.mark.parametrize("overwrite", [False, True])
def test_original_is_never_overwritten(wav, tmp_path, bypass, progress, monkeypatch, alias, overwrite):
    source = wav()
    original = source.read_bytes()
    target = source if alias == "same" else tmp_path / "alias.wav"
    if alias == "symlink":
        target.symlink_to(source)
    elif alias == "hardlink":
        target.hardlink_to(source)
    monkeypatch.setattr(pipeline, "ensure_ffmpeg", lambda: pytest.fail("Preflight must reject the input"))
    with pytest.raises(ValueError, match="original recording"):
        pipeline.process(source, target, bypass, progress, overwrite=overwrite)
    assert source.read_bytes() == original


@pytest.mark.parametrize("suffix", [".wav", ".report.json", ".mp3"])
def test_existing_output_or_sidecar_rejected_before_processing(wav, tmp_path, bypass, progress, monkeypatch, suffix):
    source = wav()
    target = tmp_path / "master.wav"
    existing = target.with_suffix(suffix)
    existing.write_bytes(b"existing output")
    bypass.mp3 = True
    monkeypatch.setattr(pipeline, "ensure_ffmpeg", lambda: pytest.fail("Preflight must reject existing outputs"))
    with pytest.raises(FileExistsError, match="Output exists"):
        pipeline.process(source, target, bypass, progress)
    assert existing.read_bytes() == b"existing output"


@pytest.mark.parametrize("kind,message", [
    ("empty", "empty"), ("silence", "silent or too quiet"),
    ("quiet", "silent or too quiet"), ("surround", "Only mono or stereo"),
    ("nonfinite", "non-finite"), ("stereo", "select --channel"),
    ("mono-right", "right channel of a mono"),
])
def test_unusable_inputs_leave_no_outputs(wav, tmp_path, bypass, progress, ffmpeg, kind, message):
    samples = {
        "empty": np.empty(0), "silence": np.zeros(48000),
        "quiet": np.full(48000, 1e-6), "surround": np.ones((4800, 3)) * 0.1,
        "nonfinite": np.array([0.1, np.nan]), "stereo": np.ones((4800, 2)) * 0.1,
        "mono-right": np.ones(4800) * 0.1,
    }[kind]
    source = wav(samples)
    target = tmp_path / "outputs" / "master.wav"
    if kind == "mono-right":
        bypass.channel = "right"
    with pytest.raises(ValueError, match=message):
        pipeline.process(source, target, bypass, progress)
    assert not target.parent.exists()


def test_measure_reports_channels_peaks_clipping_and_progress(wav):
    samples = np.tile([1.0, -0.25], (4800, 1))
    events = []
    result = measure(wav(samples), lambda *event: events.append(event))
    assert result["duration_seconds"] == 0.1
    assert result["channels"] == 2
    assert result["sample_peak_dbfs"] == 0
    assert result["channel_rms_dbfs"] == pytest.approx([0, 20 * np.log10(0.25)])
    assert result["dc_offset"] == pytest.approx([1, -0.25])
    assert result["near_full_scale_samples"] == 4800
    assert events[0][1:] == (0, 4800)
    assert events[-1][1:] == (4800, 4800)


@pytest.mark.parametrize("channel", ["left", "right", "mix"])
def test_explicit_stereo_selection_preserves_signal_and_duration(wav, tmp_path, bypass, progress, ffmpeg, channel):
    time = np.arange(48000 * 3) / 48000
    left = 0.2 * np.sin(2 * np.pi * 440 * time)
    right = 0.1 * np.sin(2 * np.pi * 880 * time)
    source = wav(np.column_stack([left, right]))
    bypass.channel = channel
    target = tmp_path / "master.wav"
    report = pipeline.process(source, target, bypass, progress)
    actual, rate = sf.read(target)
    expected = {"left": left, "right": right, "mix": (left + right) / 2}[channel]
    assert actual.ndim == 1
    assert rate == 48000
    assert len(actual) == len(expected)
    np.testing.assert_allclose(actual, expected, atol=5e-7, rtol=0)
    assert report["normalization_mode"] == "disabled"
    assert report["level_correction_db"] == 0
    assert sf.info(target).subtype == "PCM_24"


@pytest.mark.parametrize("rate", [44100, 48000])
def test_normalization_peak_ceiling_and_report(wav, tmp_path, progress, ffmpeg, rate):
    source = wav()
    original = source.read_bytes()
    target = tmp_path / "nested" / "master.wav"
    settings = resolve(overrides={"denoiser": "none", "sample_rate": rate})
    report = pipeline.process(source, target, settings, progress)
    stats = sf.info(target)
    assert stats.channels == 1
    assert stats.samplerate == rate
    assert stats.subtype == "PCM_24"
    assert abs(stats.duration - 3) <= 0.02
    verified = loudness(ffmpeg, target, settings, stats.duration, progress)
    assert verified["input_i"] == pytest.approx(settings.target_lufs, abs=0.5)
    assert verified["input_tp"] <= settings.true_peak_db + 0.1
    assert report["master"]["near_full_scale_samples"] == 0
    assert report["master"]["sample_peak_dbfs"] <= settings.true_peak_db + 0.1
    assert json.loads(target.with_suffix(".report.json").read_text()) == report
    assert source.read_bytes() == original
    assert not list(target.parent.glob(".voiceforge-*"))


def test_limiter_without_normalization(wav, tmp_path, bypass, progress, ffmpeg):
    time = np.arange(48000 * 3) / 48000
    source = wav(0.99 * np.sin(2 * np.pi * 440 * time))
    bypass.limiter = True
    bypass.true_peak_db = -6
    report = pipeline.process(source, tmp_path / "master.wav", bypass, progress)
    assert report["normalization_mode"] == "disabled"
    assert report["master_loudness"]["input_tp"] <= -5.9
    assert report["master"]["sample_peak_dbfs"] > -7


def test_speech_estimate_survives_pause_dominated_recordings(wav):
    time = np.arange(48000 * 3) / 48000
    audio = np.concatenate([np.zeros(48000 * 27), 0.02 * np.sin(2 * np.pi * 440 * time)])
    stats = measure(wav(audio), lambda *_: None)
    assert stats["quiet_blocks_dbfs"] <= -110
    assert stats["speech_level_estimate_dbfs"] == pytest.approx(-37, abs=2)


def test_leveling_uses_active_speech_not_pauses(wav, tmp_path, progress, ffmpeg):
    time = np.arange(48000 * 3) / 48000
    audio = np.concatenate([np.zeros(48000 * 27), 0.02 * np.sin(2 * np.pi * 440 * time)])
    source = wav(audio)
    settings = resolve(profile="cleanup-only", overrides={"denoiser": "none", "leveling": True})
    report = pipeline.process(source, tmp_path / "master.wav", settings, progress)
    assert 10 < report["level_correction_db"] < settings.max_gain_db
    assert not any("gain limit" in warning for warning in report["warnings"])


def test_leveling_skipped_when_speech_estimate_is_unreliable(wav, tmp_path, bypass, progress, ffmpeg, monkeypatch):
    real_measure = pipeline.measure

    def measured(path, callback, stage="Analyzing audio"):
        stats = real_measure(path, callback, stage)
        if stage == "Measuring cleaned voice":
            stats["speech_level_estimate_dbfs"] = -120
        return stats

    monkeypatch.setattr(pipeline, "measure", measured)
    bypass.leveling = True
    report = pipeline.process(wav(), tmp_path / "master.wav", bypass, progress)
    assert report["level_correction_db"] == 0
    assert any("leveling skipped" in warning for warning in report["warnings"])


def test_antiphase_mix_is_rejected_after_selection(wav, tmp_path, bypass, progress, ffmpeg):
    time = np.arange(48000 * 3) / 48000
    tone = 0.2 * np.sin(2 * np.pi * 440 * time)
    source = wav(np.column_stack([tone, -tone]))
    bypass.channel = "mix"
    target = tmp_path / "master.wav"
    with pytest.raises(ValueError, match="selected audio is silent or too quiet"):
        pipeline.process(source, target, bypass, progress)
    assert not target.exists()
    assert not target.with_suffix(".report.json").exists()
    assert not list(tmp_path.glob(".voiceforge-*"))


def test_silent_selected_channel_is_rejected_after_selection(wav, tmp_path, bypass, progress, ffmpeg):
    time = np.arange(48000 * 3) / 48000
    tone = 0.2 * np.sin(2 * np.pi * 440 * time)
    source = wav(np.column_stack([tone, np.zeros_like(tone)]))
    bypass.channel = "right"
    target = tmp_path / "master.wav"
    with pytest.raises(ValueError, match="selected audio is silent or too quiet"):
        pipeline.process(source, target, bypass, progress)
    assert not target.exists()


def test_zero_denoise_strength_skips_ai(wav, tmp_path, bypass, progress, ffmpeg, monkeypatch):
    def unexpected(*args, **kwargs):
        pytest.fail("Zero denoise strength must not invoke AI")

    monkeypatch.setattr("voiceforge.ai.denoise", unexpected)
    bypass.denoiser = "deepfilter"
    bypass.denoise_strength = 0
    report = pipeline.process(wav(), tmp_path / "master.wav", bypass, progress)
    assert report["cleaned"]["frames"] == report["input"]["frames"]


def test_fft_and_mp3_export_can_replace_generated_outputs(wav, tmp_path, bypass, progress, ffmpeg):
    source = wav()
    original = source.read_bytes()
    bypass.denoiser = "fft"
    bypass.mp3 = True
    target = tmp_path / "master.wav"
    for suffix in (".wav", ".report.json", ".mp3"):
        target.with_suffix(suffix).write_bytes(b"old output")
    events = []
    report = pipeline.process(source, target, bypass, lambda *event: events.append(event), overwrite=True)
    assert any(event[0] == "FFT noise reduction" for event in events)
    assert target.with_suffix(".mp3").stat().st_size > 1000
    assert report["mp3_loudness"]["input_i"] is not None
    assert json.loads(target.with_suffix(".report.json").read_text()) == report
    assert sf.info(target).channels == 1
    assert source.read_bytes() == original


def test_render_failure_preserves_existing_outputs_and_removes_temporary_files(wav, tmp_path, bypass, progress, ffmpeg, monkeypatch):
    source = wav()
    target = tmp_path / "master.wav"
    bypass.mp3 = True
    outputs = [target.with_suffix(suffix) for suffix in (".wav", ".report.json", ".mp3")]
    for path in outputs:
        path.write_bytes(b"previous publication")
    real_run = pipeline.ffmpeg_run

    def fail_encoding(executable, args, stage, duration, callback):
        if stage == "Encoding MP3":
            raise RuntimeError("injected encoder failure")
        return real_run(executable, args, stage, duration, callback)

    monkeypatch.setattr(pipeline, "ffmpeg_run", fail_encoding)
    with pytest.raises(RuntimeError, match="injected encoder failure"):
        pipeline.process(source, target, bypass, progress, overwrite=True)
    assert all(path.read_bytes() == b"previous publication" for path in outputs)
    assert not list(tmp_path.glob(".voiceforge-*"))


@pytest.mark.parametrize("overwrite", [False, True])
def test_publish_collision_respects_overwrite(tmp_path, overwrite):
    staged = tmp_path / "staged"
    target = tmp_path / "target"
    staged.write_bytes(b"new")
    target.write_bytes(b"concurrent output")
    if overwrite:
        pipeline.publish(staged, target, overwrite=True)
        assert target.read_bytes() == b"new"
        assert not staged.exists()
    else:
        with pytest.raises(FileExistsError):
            pipeline.publish(staged, target, overwrite=False)
        assert target.read_bytes() == b"concurrent output"
        assert staged.read_bytes() == b"new"


def test_publish_new_output_moves_staged_file(tmp_path):
    staged = tmp_path / "staged"
    target = tmp_path / "target"
    staged.write_bytes(b"finished")
    pipeline.publish(staged, target, overwrite=False)
    assert target.read_bytes() == b"finished"
    assert not staged.exists()