slop rewriteHEADmain

1 files changed, 273 insertions, 0 deletions

diff --git a/lib/project/tests/test_pipeline.py b/lib/project/tests/test_pipeline.py
new file mode 100644
index 0000000..0fa1c52
--- /dev/null
+++ b/lib/project/tests/test_pipeline.py
@@ -0,0 +1,273 @@
+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()