diff options
Diffstat (limited to 'lib/project/tests/test_pipeline.py')
| -rw-r--r-- | lib/project/tests/test_pipeline.py | 65 |
1 files changed, 10 insertions, 55 deletions
diff --git a/lib/project/tests/test_pipeline.py b/lib/project/tests/test_pipeline.py index 3f6000d..797bc12 100644 --- a/lib/project/tests/test_pipeline.py +++ b/lib/project/tests/test_pipeline.py @@ -41,7 +41,8 @@ def test_original_is_never_overwritten(wav, tmp_path, bypass, progress, monkeypa 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 = pipeline.report_for(target) if suffix == ".report.json" else target.with_suffix(suffix) + existing.parent.mkdir(parents=True, exist_ok=True) existing.write_bytes(b"existing output") bypass.mp3 = True monkeypatch.setattr(pipeline, "ensure_ffmpeg", lambda: pytest.fail("Preflight must reject existing outputs")) @@ -53,20 +54,16 @@ def test_existing_output_or_sidecar_rejected_before_processing(wav, tmp_path, by @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"), + ("nonfinite", "non-finite"), ("stereo", "downmix to 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() @@ -86,26 +83,6 @@ def test_measure_reports_channels_peaks_clipping_and_progress(wav): 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_32" - - @pytest.mark.parametrize("rate", [44100, 48000]) def test_normalization_peak_ceiling_and_report(wav, tmp_path, progress, ffmpeg, rate): source = wav() @@ -123,7 +100,7 @@ def test_normalization_peak_ceiling_and_report(wav, tmp_path, progress, ffmpeg, 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 json.loads(pipeline.report_for(target).read_text()) == report assert source.read_bytes() == original assert not list(target.parent.glob(".producer-*")) @@ -173,30 +150,6 @@ def test_leveling_skipped_when_speech_estimate_is_unreliable(wav, tmp_path, bypa 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(".producer-*")) - - -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") @@ -214,14 +167,15 @@ def test_fft_and_mp3_export_can_replace_generated_outputs(wav, tmp_path, bypass, 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") + for path in (target, pipeline.report_for(target), target.with_suffix(".mp3")): + path.parent.mkdir(parents=True, exist_ok=True) + path.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 json.loads(pipeline.report_for(target).read_text()) == report assert sf.info(target).channels == 1 assert source.read_bytes() == original @@ -230,8 +184,9 @@ def test_render_failure_preserves_existing_outputs_and_removes_temporary_files(w source = wav() target = tmp_path / "master.wav" bypass.mp3 = True - outputs = [target.with_suffix(suffix) for suffix in (".wav", ".report.json", ".mp3")] + outputs = [target, pipeline.report_for(target), target.with_suffix(".mp3")] for path in outputs: + path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(b"previous publication") real_run = pipeline.ffmpeg_run |