aboutsummaryrefslogtreecommitdiff
path: root/lib/project/tests/test_pipeline.py
diff options
context:
space:
mode:
Diffstat (limited to 'lib/project/tests/test_pipeline.py')
-rw-r--r--lib/project/tests/test_pipeline.py59
1 files changed, 57 insertions, 2 deletions
diff --git a/lib/project/tests/test_pipeline.py b/lib/project/tests/test_pipeline.py
index 797bc12..fe04519 100644
--- a/lib/project/tests/test_pipeline.py
+++ b/lib/project/tests/test_pipeline.py
@@ -37,11 +37,11 @@ def test_original_is_never_overwritten(wav, tmp_path, bypass, progress, monkeypa
assert source.read_bytes() == original
-@pytest.mark.parametrize("suffix", [".wav", ".report.json", ".mp3"])
+@pytest.mark.parametrize("suffix", [".wav", ".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 = pipeline.report_for(target) if suffix == ".report.json" else target.with_suffix(suffix)
+ existing = target.with_suffix(suffix)
existing.parent.mkdir(parents=True, exist_ok=True)
existing.write_bytes(b"existing output")
bypass.mp3 = True
@@ -51,6 +51,16 @@ def test_existing_output_or_sidecar_rejected_before_processing(wav, tmp_path, by
assert existing.read_bytes() == b"existing output"
+def test_existing_report_log_is_replaced_without_overwrite(wav, tmp_path, bypass, progress, ffmpeg):
+ source = wav()
+ target = tmp_path / "master.wav"
+ existing = pipeline.report_for(target)
+ existing.parent.mkdir(parents=True, exist_ok=True)
+ existing.write_bytes(b"stale report")
+ report = pipeline.process(source, target, bypass, progress)
+ assert json.loads(existing.read_text()) == report
+
+
@pytest.mark.parametrize("kind,message", [
("empty", "empty"), ("silence", "silent or too quiet"),
("quiet", "silent or too quiet"), ("surround", "Only mono or stereo"),
@@ -226,3 +236,48 @@ def test_publish_new_output_moves_staged_file(tmp_path):
pipeline.publish(staged, target, overwrite=False)
assert target.read_bytes() == b"finished"
assert not staged.exists()
+
+
+def test_trim_gaps_shortens_silence_and_keeps_speech_exact(wav, tmp_path, progress):
+ from producer.audio import trim_gaps
+ sr = 48000
+ speech_a = 0.2 * np.sin(2 * np.pi * 440 * np.arange(2 * sr) / sr)
+ room_tone = 0.001 * np.random.default_rng(7).normal(0, 1, int(3 * sr)) # -60 dBFS
+ speech_b = 0.2 * np.sin(2 * np.pi * 880 * np.arange(2 * sr) / sr)
+ source = tmp_path / "gaps.wav"
+ sf.write(source, np.concatenate([speech_a, room_tone, speech_b]), sr, format="WAV", subtype="FLOAT")
+ target = tmp_path / "trimmed.wav"
+ trimmed, removed = trim_gaps(source, target, threshold_db=-50, max_gap_s=1.5,
+ pad_s=0.6, progress=progress)
+ result, rate = sf.read(target, dtype="float32")
+ assert rate == 48000 and trimmed == 1
+ assert removed == pytest.approx(3 - 0.6, abs=0.02) # 0.3 s pad kept at each edge
+ # Speech is copied through unattenuated and sample-exact.
+ np.testing.assert_allclose(result[:len(speech_a)], speech_a, atol=1e-7)
+ np.testing.assert_allclose(result[-len(speech_b):], speech_b, atol=1e-6)
+ kept_gap = result[len(speech_a):len(speech_a) + int(0.6 * sr)]
+ assert np.count_nonzero(kept_gap) == int(0.6 * sr) # room tone preserved, no gate
+
+
+def test_trim_and_clearvoice_dispatch_flow(wav, tmp_path, bypass, progress, ffmpeg, monkeypatch):
+ def fake_denoise(source, target, device, strength, progress, backend):
+ assert backend == "clearvoice"
+ import shutil
+ shutil.copyfile(source, target)
+
+ monkeypatch.setattr("producer.ai.denoise", fake_denoise)
+ bypass.denoiser = "clearvoice"
+ bypass.denoise_strength = 1.0
+ bypass.trim_silence = True
+ bypass.max_gap_s = 1.0
+ bypass.gap_pad_s = 0.5
+ sr = 48000
+ speech = 0.15 * np.sin(2 * np.pi * 440 * np.arange(sr) / sr)
+ room_tone = 0.001 * np.random.default_rng(7).normal(0, 1, int(2 * sr))
+ source = wav(np.concatenate([speech, room_tone, speech]))
+ report = pipeline.process(source, tmp_path / "master.wav", bypass, progress)
+ assert report["gaps_trimmed"] == 1
+ assert report["silence_removed_seconds"] > 0
+ assert report["master"]["duration_seconds"] == pytest.approx(
+ report["input"]["duration_seconds"] - report["silence_removed_seconds"], abs=0.02)
+ assert report["warnings"] == []