aboutsummaryrefslogtreecommitdiff
path: root/lib/tests/test_cli.py
diff options
context:
space:
mode:
authorhistoria <historiavg@proton.me>2026-09-07 06:47:47 -0400
committerhistoria <historiavg@proton.me>2026-09-07 06:47:47 -0400
commit84dd2d068317998f6fb59400c534ef5be6b51b53 (patch)
tree025293e9d9229e02960374771ae522d9de2628ce /lib/tests/test_cli.py
parent39b0f2bbed74f6487a41b82501ae3c6799e4b5c4 (diff)
downloadproducer-main.tar.gz
slop rewriteHEADmain
Diffstat (limited to 'lib/tests/test_cli.py')
-rw-r--r--lib/tests/test_cli.py357
1 files changed, 0 insertions, 357 deletions
diff --git a/lib/tests/test_cli.py b/lib/tests/test_cli.py
deleted file mode 100644
index 76952ba..0000000
--- a/lib/tests/test_cli.py
+++ /dev/null
@@ -1,357 +0,0 @@
-import sys
-from types import SimpleNamespace
-
-import numpy as np
-import pytest
-import soundfile as sf
-from conftest import speechish
-
-from producer import io as pio
-from producer.cli import _apply_args, build_parser, process_one
-from producer.config import Options
-
-
-def _mk_wav(tmp_path, name="in.wav", stereo=False):
- x = speechish(3.0, level_dbfs=-30.0)
- if stereo:
- data = np.stack([x, x * 0.5], axis=1)
- sf.write(str(tmp_path / name), data, 44100, subtype="PCM_16")
- else:
- sf.write(str(tmp_path / name), x, 44100, subtype="PCM_16")
- return tmp_path / name
-
-
-def _opts(**kw):
- opts = Options()
- opts.denoise = "off"
- opts.enhance = "off"
- for k, v in kw.items():
- setattr(opts, k, v)
- return opts
-
-
-def test_decode_stereo_mixdown(tmp_path):
- p = _mk_wav(tmp_path, "st.wav", stereo=True)
- x, sr = pio.decode(p)
- assert sr == 44100
- assert x.dtype.name == "float32"
- assert x.ndim == 1
-
-
-def test_encode_wav_bitdepths(tmp_path):
- x = speechish(2.0, level_dbfs=-20.0)
- for depth in (16, 24, 32):
- out = tmp_path / f"o{depth}.wav"
- pio.encode(x, 44100, out, "wav", depth)
- y, sr = pio.decode(out)
- assert sr == 44100
- assert (
- abs(
- float(np.sqrt(np.mean(y.astype(np.float64) ** 2)))
- - float(np.sqrt(np.mean(x.astype(np.float64) ** 2)))
- )
- < 1e-3
- )
-
-
-def test_encode_flac(tmp_path):
- x = speechish(2.0, level_dbfs=-20.0)
- out = tmp_path / "o.flac"
- pio.encode(x, 44100, out, "flac", 24)
- y, sr = pio.decode(out)
- assert sr == 44100
- assert np.corrcoef(x, y)[0, 1] > 0.999
-
-
-def test_mp3_roundtrip(tmp_path):
- import pytest as _pt
-
- if not pio.ffmpeg_available():
- _pt.skip("ffmpeg missing")
- x = speechish(4.0, level_dbfs=-20.0)
- out = tmp_path / "o.mp3"
- pio.encode(x, 44100, out, "mp3", 16)
- y, sr = pio.decode(out)
- assert sr == 44100
- from producer import meters
-
- assert abs(meters.rms_db(x) - meters.rms_db(y)) < 0.7
-
-
-def test_decode_via_ffmpeg_fallback(tmp_path):
- import pytest as _pt
-
- if not pio.ffmpeg_available():
- _pt.skip("ffmpeg missing")
- x = speechish(4.0, level_dbfs=-20.0)
- out = tmp_path / "o.mp3"
- pio.encode(x, 44100, out, "mp3", 16)
- y, sr = pio.decode(out)
- assert sr == 44100
- assert y.size > 0
-
-
-def test_process_one_end_to_end(tmp_path, capsys):
- import json
-
- from producer import meters
-
- inp = _mk_wav(tmp_path, "e2e.wav")
- out = tmp_path / "e2e_processed.wav"
- rc = process_one(inp, _opts(report=True), single=True)
- assert rc == out
- assert out.exists()
- rep_path = out.with_name("e2e_processed.report.json")
- assert rep_path.exists()
- rep = json.loads(rep_path.read_text())
- y, sr = pio.decode(out)
- assert abs(meters.rms_db(y) + 20.0) < 0.6
- assert meters.true_peak_db(y, sr) <= -2.9
- assert rep["after"]["rms_db"] != 0
-
-
-def test_dry_run_listing(tmp_path, capsys):
- from producer.cli import main
-
- inp = _mk_wav(tmp_path, "dry.wav")
- rc = main([str(inp), "--dry-run", "--denoise", "off"])
- assert rc == 0
- out = capsys.readouterr().out
- assert "denoise" in out and "dsp" in out and "levelling" in out
-
-
-def test_arg_parsing_precedence():
- parser = build_parser()
- args = parser.parse_args(
- ["in.wav", "--profile", "podcast", "--warmth", "0.1", "--ceiling", "-2.0"]
- )
- opts = Options()
- _apply_args(opts, args)
- assert opts.profile == "podcast"
- assert opts.loudness_mode() == "lufs"
- assert opts.strengths["warmth"] == 0.1
- assert opts.ceiling() == -2.0
- assert opts.strengths["air"] is None
-
-
-def test_radio_profile_has_tuned_defaults():
- from producer import pipeline
-
- opts = Options(profile="radio")
- assert opts.eff("tape") > 0 and opts.eff("soothe") > 0
- assert opts.loudness_mode() == "lufs"
- stages = pipeline.build_stages(opts)
- assert [st.name for st in stages] == ["denoise", "enhance", "dsp", "levelling"]
- assert opts.denoise_strength is not None
-
-
-def test_tape_and_soothe_flags_override():
- args = build_parser().parse_args(
- ["in.wav", "--profile", "radio", "--tape", "0.4", "--soothe", "0.7"]
- )
- opts = Options()
- _apply_args(opts, args)
- assert opts.profile == "radio"
- assert opts.strengths["tape"] == 0.4
- assert opts.strengths["soothe"] == 0.7
-
-
-def test_default_output_suffix_processed(tmp_path):
- from producer.cli import _resolve_output
-
- inp = tmp_path / "song.wav"
- assert _resolve_output(inp, _opts(), single=True) == tmp_path / "song_processed.wav"
- odir = tmp_path / "out"
- opts = _opts(output=str(odir))
- assert _resolve_output(inp, opts, single=False) == odir / "song_processed.wav"
-
-
-def test_engine_chunk_flags():
- args = build_parser().parse_args(["in.wav", "--engine-chunk", "15", "--engine-overlap", "1.0"])
- opts = Options()
- _apply_args(opts, args)
- assert opts.engine_chunk_s == 15.0
- assert opts.engine_overlap_s == 1.0
-
-
-def test_default_options_whole_file_and_soft_denoise():
- opts = Options()
- assert opts.engine_chunk_s == 0.0
- assert opts.denoise_strength == 0.9
-
-
-def test_engine_chunk_validation():
- with pytest.raises(SystemExit):
- args = build_parser().parse_args(["in.wav", "--engine-chunk", "5", "--engine-overlap", "5"])
- _apply_args(Options(), args)
- with pytest.raises(SystemExit):
- args = build_parser().parse_args(["in.wav", "--engine-chunk", "-1"])
- _apply_args(Options(), args)
-
-
-def test_engine_chunk_config_override(tmp_path):
- from producer import config as cfgmod
-
- cfg = tmp_path / "config.toml"
- cfg.write_text("engine_chunk = 10.0\nengine_overlap = 1.0\n")
- opts = Options()
- cfgmod.apply_config(opts, cfgmod.load_config(cfg))
- assert opts.engine_chunk_s == 10.0
- assert opts.engine_overlap_s == 1.0
-
-
-def test_denoise_pf_flag_and_config_plumbing(tmp_path):
- from producer import config as cfgmod
-
- opts = Options()
- _apply_args(opts, build_parser().parse_args(["in.wav"]))
- assert opts.denoise_pf is False
-
- opts = Options()
- _apply_args(opts, build_parser().parse_args(["in.wav", "--denoise-pf"]))
- assert opts.denoise_pf is True
-
- cfg = tmp_path / "config.toml"
- cfg.write_text('[denoise]\nengine = "dfn3"\nstrength = 0.8\npf = true\n')
- opts = Options()
- cfgmod.apply_config(opts, cfgmod.load_config(cfg))
- assert opts.denoise_strength == 0.8
- assert opts.denoise_pf is True
-
-
-def test_spectral_denoise_choice(tmp_path):
- from producer import config as cfgmod
-
- opts = Options()
- _apply_args(opts, build_parser().parse_args(["in.wav", "--denoise", "spectral"]))
- assert opts.denoise == "spectral"
-
- cfg = tmp_path / "config.toml"
- cfg.write_text('[denoise]\nengine = "spectral"\nstrength = 0.8\n')
- opts = Options()
- cfgmod.apply_config(opts, cfgmod.load_config(cfg))
- assert opts.denoise == "spectral"
- assert opts.denoise_strength == 0.8
-
-
-def test_conflict_auto_renames_when_not_a_tty(tmp_path):
- inp = _mk_wav(tmp_path, "in.wav")
- first = process_one(inp, _opts(), single=True)
- assert first == tmp_path / "in_processed.wav"
- assert process_one(inp, _opts(), single=True) == tmp_path / "in_processed_1.wav"
- assert process_one(inp, _opts(), single=True) == tmp_path / "in_processed_2.wav"
- assert (tmp_path / "in_processed.wav").exists()
-
-
-def test_conflict_prompt_overwrite(tmp_path, monkeypatch):
- inp = _mk_wav(tmp_path, "in.wav")
- out = tmp_path / "in_processed.wav"
- process_one(inp, _opts(), single=True)
- monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True))
- monkeypatch.setattr("builtins.input", lambda _prompt: "o")
- assert process_one(inp, _opts(), single=True) == out
- assert out.exists()
-
-
-def test_conflict_prompt_rename(tmp_path, monkeypatch):
- inp = _mk_wav(tmp_path, "in.wav")
- out = tmp_path / "in_processed.wav"
- process_one(inp, _opts(), single=True)
- monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True))
- monkeypatch.setattr("builtins.input", lambda _prompt: "r")
- second = process_one(inp, _opts(), single=True)
- assert second == tmp_path / "in_processed_1.wav"
- assert second.exists()
- assert out.exists()
-
-
-def test_conflict_prompt_invalid_then_overwrite(tmp_path, monkeypatch):
- inp = _mk_wav(tmp_path, "in.wav")
- out = tmp_path / "in_processed.wav"
- process_one(inp, _opts(), single=True)
- monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True))
- answers = iter(["maybe", "O"])
- monkeypatch.setattr("builtins.input", lambda _prompt: next(answers))
- assert process_one(inp, _opts(), single=True) == out
-
-
-def test_conflict_prompt_cancel(tmp_path, monkeypatch, capsys):
- inp = _mk_wav(tmp_path, "in.wav")
- out = tmp_path / "in_processed.wav"
- process_one(inp, _opts(), single=True)
- before = out.read_bytes()
- monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True))
- monkeypatch.setattr("builtins.input", lambda _prompt: "c")
- assert process_one(inp, _opts(), single=True) is None
- assert out.read_bytes() == before
- assert "skipped" in capsys.readouterr().out
-
-
-def test_conflict_prompt_eof_cancels(tmp_path, monkeypatch):
- inp = _mk_wav(tmp_path, "in.wav")
- process_one(inp, _opts(), single=True)
- monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True))
-
- def _eof(_prompt):
- raise EOFError
-
- monkeypatch.setattr("builtins.input", _eof)
- assert process_one(inp, _opts(), single=True) is None
-
-
-def test_process_one_reports_stage_status(tmp_path, capsys):
- inp = _mk_wav(tmp_path, "status.wav")
- out_path = process_one(inp, _opts(), single=True)
- assert out_path is not None
- out = capsys.readouterr().out
- assert f"[producer] {inp} (" in out
- assert "dsp done in" in out
- assert "levelling done in" in out
- assert f"[producer] wrote {out_path}" in out
-
-
-def test_multi_file_run_gets_position_prefixes(tmp_path, capsys):
- from producer.cli import main
-
- a = _mk_wav(tmp_path, "a.wav")
- b = _mk_wav(tmp_path, "b.wav")
- rc = main([str(a), str(b), "--denoise", "off"])
- assert rc == 0
- out = capsys.readouterr().out
- assert "[1/2]" in out and "[2/2]" in out
- assert (tmp_path / "a_processed.wav").exists()
- assert (tmp_path / "b_processed.wav").exists()
-
-
-def test_output_file_rejected_for_multiple_inputs(tmp_path):
- from producer.cli import main
-
- a = _mk_wav(tmp_path, "a.wav")
- b = _mk_wav(tmp_path, "b.wav")
- with pytest.raises(SystemExit):
- main([str(a), str(b), "-o", str(tmp_path / "out.wav"), "--denoise", "off"])
-
-
-def test_output_dir_accepted_for_multiple_inputs(tmp_path):
- from producer.cli import main
-
- a = _mk_wav(tmp_path, "a.wav")
- b = _mk_wav(tmp_path, "b.wav")
- outdir = tmp_path / "masters"
- rc = main([str(a), str(b), "-o", str(outdir), "--denoise", "off"])
- assert rc == 0
- assert (outdir / "a_processed.wav").exists()
- assert (outdir / "b_processed.wav").exists()
-
-
-def test_batch_failure_continues_to_next_file(tmp_path, capsys):
- from producer.cli import main
-
- good = _mk_wav(tmp_path, "good.wav")
- missing = tmp_path / "missing.wav"
- rc = main([str(missing), str(good), "--denoise", "off"])
- assert rc == 1
- captured = capsys.readouterr()
- assert "[2/2]" in captured.out
- assert "[producer] ERROR" in captured.err
- assert (tmp_path / "good_processed.wav").exists()