slop rewriteHEADmain

1 files changed, 366 insertions, 0 deletions

diff --git a/lib/project/tests/test_ai.py b/lib/project/tests/test_ai.py
new file mode 100644
index 0000000..e8c5872
--- /dev/null
+++ b/lib/project/tests/test_ai.py
@@ -0,0 +1,366 @@
+"""Offline AI contract tests; no Torch installation, model download, or GPU needed."""
+import hashlib
+import io
+import json
+import os
+from pathlib import Path
+import sys
+import time
+from types import SimpleNamespace
+from unittest.mock import Mock
+import zipfile
+
+import numpy as np
+import pytest
+import soundfile as sf
+
+from voiceforge import ai, ai_worker, setup
+
+
+MODEL_BYTES = 7986207
+MODEL_HASH = "49c52edc8947ae1f9bf50d81530beaf3a2c3245aeaf34b6f31ff535cd22284d2"
+
+
+@pytest.mark.parametrize("size,message", [
+    (10, "checksum mismatch"),
+    (MODEL_BYTES, "checksum mismatch"),
+    (MODEL_BYTES + 1, "exceeds the expected size"),
+])
+def test_model_rejects_truncated_corrupt_and_oversized_downloads(tmp_path, monkeypatch, size, message):
+    monkeypatch.setattr(ai_worker.urllib.request, "urlopen", Mock(return_value=io.BytesIO(b"x" * size)))
+    progress = Mock()
+    monkeypatch.setattr(ai_worker, "report", progress)
+    with pytest.raises(RuntimeError, match=message):
+        ai_worker.ensure_model(tmp_path / "model")
+    assert list(tmp_path.iterdir()) == []
+    events = [call.args for call in progress.call_args_list]
+    assert events[0] == ("Downloading DeepFilterNet3 model (bytes)", 0, MODEL_BYTES)
+    counts = [completed for _, completed, total in events if total == MODEL_BYTES]
+    assert counts == sorted(counts)
+    assert all(0 <= count <= MODEL_BYTES for count in counts)
+
+
+@pytest.mark.parametrize("unsafe", [False, True])
+def test_mocked_model_archive_progress_extraction_and_reuse(tmp_path, monkeypatch, unsafe):
+    # Mock only the release digest for a synthetic archive; corruption tests above
+    # exercise the real SHA-256 gate. Keep its real byte count and ZIP extraction.
+    def archive(padding):
+        stream = io.BytesIO()
+        with zipfile.ZipFile(stream, "w") as bundle:
+            bundle.writestr("DeepFilterNet3/config.ini", "[df]\nsr=48000\n")
+            bundle.writestr("DeepFilterNet3/checkpoints/model_120.ckpt.best", "checkpoint")
+            bundle.writestr("../escape" if unsafe else "padding", b"x" * padding)
+        return stream.getvalue()
+
+    payload = archive(MODEL_BYTES - len(archive(0)))
+    assert len(payload) == MODEL_BYTES
+    digest = Mock(wraps=hashlib.sha256())
+    digest.hexdigest.return_value = MODEL_HASH
+    monkeypatch.setattr(ai_worker.hashlib, "sha256", lambda: digest)
+    download = Mock(return_value=io.BytesIO(payload))
+    monkeypatch.setattr(ai_worker.urllib.request, "urlopen", download)
+    progress = Mock()
+    monkeypatch.setattr(ai_worker, "report", progress)
+    directory = tmp_path / "model"
+    if unsafe:
+        with pytest.raises(RuntimeError, match="Unsafe path"):
+            ai_worker.ensure_model(directory)
+        assert not (tmp_path / "escape").exists()
+        assert list(tmp_path.iterdir()) == []
+    else:
+        model = ai_worker.ensure_model(directory)
+        assert (model / "config.ini").is_file()
+        assert (model / "checkpoints/model_120.ckpt.best").is_file()
+        assert ai_worker.ensure_model(directory) == model
+        download.assert_called_once()
+    assert b"".join(call.args[0] for call in digest.update.call_args_list) == payload
+    events = [call.args for call in progress.call_args_list]
+    byte_events = [event for event in events if "(bytes)" in event[0]]
+    assert byte_events[0][1:] == (0, MODEL_BYTES)
+    assert byte_events[-1][1:] == (MODEL_BYTES, MODEL_BYTES)
+    assert len(byte_events) > 2
+    assert events[-1] == ("Verifying and extracting DeepFilterNet3 model",)
+
+
+def test_incomplete_model_fails_without_network(tmp_path, monkeypatch):
+    directory = tmp_path / "model"
+    directory.mkdir()
+    download = Mock(side_effect=AssertionError("Unexpected network call"))
+    monkeypatch.setattr(ai_worker.urllib.request, "urlopen", download)
+    with pytest.raises(RuntimeError, match="Incomplete model directory"):
+        ai_worker.ensure_model(directory)
+    download.assert_not_called()
+
+
+@pytest.mark.parametrize("close_output", [False, True])
+def test_subprocess_heartbeat_does_not_block_on_silence_or_closed_output(close_output):
+    events = []
+    code = "import os,time; "
+    if close_output:
+        code += "os.close(1); os.close(2); "
+    code += "time.sleep(0.8)"
+    setup.run_process([sys.executable, "-c", code], "Waiting", lambda *event: events.append(event))
+    assert events[0] == ("Waiting", None, None)
+    elapsed = [completed for _, completed, total in events[1:] if total is None]
+    assert len(elapsed) >= 2
+    assert elapsed == sorted(elapsed)
+
+
+def test_subprocess_parses_split_byte_progress_records():
+    events = []
+    record = "VOICEFORGE_PROGRESS " + json.dumps(["Download (bytes)", 123, 456]) + "\n"
+    code = f"import os,time; os.write(1,{record[:12].encode()!r}); time.sleep(.3); os.write(1,{record[12:].encode()!r})"
+    setup.run_process([sys.executable, "-c", code], "Download", lambda *event: events.append(event))
+    assert events[-1] == ("Download (bytes)", 123, 456)
+
+
+@pytest.mark.parametrize("exception", [KeyboardInterrupt, RuntimeError])
+def test_callback_cancellation_terminates_and_reaps_child(monkeypatch, exception):
+    popen = setup.subprocess.Popen
+    children = []
+
+    def launch(*args, **kwargs):
+        child = popen(*args, **kwargs)
+        children.append(child)
+        return child
+
+    def cancel(stage, completed, total):
+        if completed is not None:
+            raise exception("cancelled")
+
+    monkeypatch.setattr(setup.subprocess, "Popen", launch)
+    started = time.monotonic()
+    with pytest.raises(exception, match="cancelled"):
+        setup.run_process([sys.executable, "-c", "import time; time.sleep(30)"], "Waiting", cancel)
+    assert len(children) == 1 and children[0].poll() is not None
+    assert time.monotonic() - started < 10
+
+
+def test_cancellation_kills_descendants_that_outlive_the_leader(tmp_path, monkeypatch):
+    sentinel = tmp_path / "grandchild.pid"
+    monkeypatch.setenv("GRANDCHILD_PID_FILE", str(sentinel))
+    child_code = (
+        "import os, subprocess, sys, time\n"
+        "subprocess.Popen([sys.executable, '-c', "
+        "\"import os, time; open(os.environ['GRANDCHILD_PID_FILE'], 'w')"
+        ".write(str(os.getpid())); time.sleep(60)\"])\n"
+        "while not os.path.exists(os.environ['GRANDCHILD_PID_FILE']):\n"
+        "    time.sleep(0.02)\n"
+    )
+
+    def cancel(stage, completed, total):
+        if completed is not None and children and children[0].poll() is not None:
+            raise KeyboardInterrupt("cancelled")
+
+    popen = setup.subprocess.Popen
+    children = []
+
+    def launch(*args, **kwargs):
+        child = popen(*args, **kwargs)
+        children.append(child)
+        return child
+
+    monkeypatch.setattr(setup.subprocess, "Popen", launch)
+    with pytest.raises(KeyboardInterrupt, match="cancelled"):
+        setup.run_process([sys.executable, "-c", child_code], "Waiting", cancel)
+
+    def alive(pid):
+        try:
+            for line in Path(f"/proc/{pid}/status").read_text().splitlines():
+                if line.startswith("State:"):
+                    return "Z" not in line
+            return True
+        except FileNotFoundError:
+            return False
+
+    grandchild = int(sentinel.read_text())
+    for _ in range(50):
+        if not alive(grandchild):
+            break
+        time.sleep(0.1)
+    else:
+        pytest.fail("A descendant survived the cancellation of its process group")
+
+
+def test_subprocess_failure_keeps_bounded_diagnostics():
+    code = "import sys; print('x'*100000); print('specific failure'); sys.exit(7)"
+    with pytest.raises(RuntimeError, match="exit 7") as error:
+        setup.run_process([sys.executable, "-c", code], "Worker")
+    assert "specific failure" in str(error.value)
+    assert len(str(error.value)) < 66000
+
+
+@pytest.mark.parametrize("strength", [-1, 2, float("nan"), float("inf")])
+def test_invalid_strength_fails_before_setup(tmp_path, monkeypatch, strength):
+    install = Mock(side_effect=AssertionError("Unexpected setup"))
+    monkeypatch.setattr(ai, "ensure_ai", install)
+    with pytest.raises(ValueError, match="strength"):
+        ai.denoise(tmp_path / "input.wav", tmp_path / "output.wav", strength=strength)
+    install.assert_not_called()
+
+
+def _ready_environment(root, flavor="cpu"):
+    environment = root / f"ai-df-0.5.6-torch-2.5.1-{flavor}-py311-v1"
+    (environment / "bin").mkdir(parents=True)
+    (environment / "bin/python").write_text("#!/bin/sh\n")
+    (environment / ".voiceforge-ready").touch()
+    return environment
+
+
+def test_ready_environment_skips_the_check_worker(tmp_path, monkeypatch):
+    root = tmp_path / "prefix"
+    environment = _ready_environment(root)
+    monkeypatch.setenv("VOICEFORGE_HOME", str(root))
+    monkeypatch.setattr(ai.os, "confstr", lambda *_: "glibc 2.39")
+    executed = []
+    monkeypatch.setattr(ai, "run_process", lambda args, *rest, **kwargs: executed.append(args))
+    python = ai.ensure_ai("cpu")
+    assert executed == []
+    assert python == environment / "bin/python"
+
+
+def test_setup_verifies_even_a_ready_environment(tmp_path, monkeypatch):
+    root = tmp_path / "prefix"
+    _ready_environment(root)
+    monkeypatch.setenv("VOICEFORGE_HOME", str(root))
+    monkeypatch.setattr(ai.os, "confstr", lambda *_: "glibc 2.39")
+    executed = []
+    monkeypatch.setattr(ai, "run_process", lambda args, *rest, **kwargs: executed.append(args))
+    ai.ensure_ai("cpu", verify=True)
+    assert len(executed) == 1
+    assert executed[0][-1] == "--check"
+
+
+def test_broken_environment_is_rebuilt_and_self_tested(tmp_path, monkeypatch):
+    root = tmp_path / "prefix"
+    environment = _ready_environment(root)
+    (environment / "bin/python").unlink()
+    (environment / "bin/python").symlink_to("/nonexistent/python3.11")
+    leftover = environment / "leftover.txt"
+    leftover.write_text("from the previous location")
+    monkeypatch.setenv("VOICEFORGE_HOME", str(root))
+    monkeypatch.setattr(ai.os, "confstr", lambda *_: "glibc 2.39")
+    stages, commands = [], []
+
+    def fake_run(args, stage, progress=None, **kwargs):
+        commands.append(args)
+        stages.append(stage)
+        if stage == "Checking DeepFilterNet3":
+            environment.mkdir(parents=True, exist_ok=True)
+            (environment / ".voiceforge-ready").touch()
+
+    monkeypatch.setattr(ai, "run_process", fake_run)
+    monkeypatch.setattr(ai, "ensure_uv", lambda progress=None: "uv")
+    ai.ensure_ai("cpu")
+    assert not leftover.exists()
+    assert stages == ["Preparing AI Python 3.11", "Installing PyTorch (cpu)",
+                      "Installing DeepFilterNet3", "Checking DeepFilterNet3"]
+    assert commands[0][-1] == str(environment)
+    assert (environment / ".voiceforge-ready").is_file()
+
+
+def test_copied_environment_is_rebuilt_not_trusted(tmp_path, monkeypatch):
+    # Copying an installation while the original remains leaves the copy's
+    # interpreter links resolving into the original prefix; that must not
+    # count as ready, or the copy breaks when the original is removed.
+    root = tmp_path / "prefix"
+    environment = _ready_environment(root)
+    original = _ready_environment(tmp_path / "other")
+    (environment / "bin/python").unlink()
+    (environment / "bin/python").symlink_to(original / "bin/python")
+    monkeypatch.setenv("VOICEFORGE_HOME", str(root))
+    monkeypatch.setattr(ai.os, "confstr", lambda *_: "glibc 2.39")
+    stages = []
+
+    def fake_run(args, stage, progress=None, **kwargs):
+        stages.append(stage)
+        if stage == "Checking DeepFilterNet3":
+            environment.mkdir(parents=True, exist_ok=True)
+            (environment / ".voiceforge-ready").touch()
+
+    monkeypatch.setattr(ai, "run_process", fake_run)
+    monkeypatch.setattr(ai, "ensure_uv", lambda progress=None: "uv")
+    ai.ensure_ai("cpu")
+    assert stages[0] == "Preparing AI Python 3.11"
+
+
+@pytest.fixture
+def mocked_worker(tmp_path, monkeypatch):
+    class Tensor:
+        def __init__(self, array):
+            self.array = array
+
+        def unsqueeze(self, axis):
+            return Tensor(np.expand_dims(self.array, axis))
+
+        def squeeze(self, axis):
+            return Tensor(np.squeeze(self.array, axis))
+
+        def numpy(self):
+            return self.array
+
+    calls = []
+
+    def enhance(model, state, tensor, pad):
+        assert pad is True
+        assert tensor.array.shape[1] % 480 == 0
+        calls.append(tensor.array.shape[1])
+        return Tensor(tensor.array * 0.5)
+
+    state = SimpleNamespace(sr=lambda: 48000)
+    backend = SimpleNamespace(enhance=enhance, init_df=lambda *a, **kw: (None, state, None))
+    for name, module in {
+        "torch": SimpleNamespace(set_num_threads=lambda n: None, from_numpy=Tensor,
+                                 cuda=SimpleNamespace(is_available=lambda: False)),
+        "df": SimpleNamespace(),
+        "df.enhance": backend,
+        "df.model": SimpleNamespace(ModelParams=lambda: SimpleNamespace(
+            sr=48000, fft_size=960, hop_size=480, nb_erb=32, min_nb_freqs=2)),
+        "libdf": SimpleNamespace(DF=lambda **kw: state),
+    }.items():
+        monkeypatch.setitem(sys.modules, name, module)
+    monkeypatch.setattr(ai_worker, "ensure_model", lambda path: path)
+    monkeypatch.setattr(ai_worker.signal, "signal", lambda *args: None)
+    monkeypatch.setenv("DEVICE", "cpu")
+    progress = Mock()
+    monkeypatch.setattr(ai_worker, "report", progress)
+
+    def run(source, target, strength):
+        monkeypatch.setattr(sys, "argv", ["worker", "--device", "cpu", "--model-dir", str(tmp_path),
+                                         "--source", str(source), "--target", str(target),
+                                         "--strength", str(strength)])
+        ai_worker.main()
+
+    return run, calls, backend, progress
+
+
+@pytest.mark.parametrize("frames", [0, 1, 479, 481, 480001, 960017])
+@pytest.mark.parametrize("strength", [0, 0.85, 1])
+def test_mocked_worker_exact_length_mix_and_bounded_windows(tmp_path, wav, mocked_worker, frames, strength):
+    run, calls, _, progress = mocked_worker
+    audio = np.random.default_rng(42).normal(0, 0.1, frames).astype("float32")
+    source = wav(audio)
+    target = tmp_path / "clean.wav"
+    run(source, target, strength)
+    result, rate = sf.read(target, dtype="float32")
+    assert len(result) == frames and rate == 48000
+    assert sf.info(target).subtype == "FLOAT"
+    np.testing.assert_allclose(result, audio * (1 - strength * 0.5), rtol=2e-6, atol=1e-8)
+    if strength == 0:
+        np.testing.assert_array_equal(result, audio)
+        assert calls == []
+    assert max(calls, default=0) <= 674400
+    assert progress.call_args.args == ("Denoising (cpu)", frames, frames)
+    assert not list(tmp_path.glob(".voiceforge-ai-*"))
+
+
+def test_worker_failure_preserves_existing_target(tmp_path, wav, mocked_worker):
+    run, _, backend, _ = mocked_worker
+    backend.enhance = Mock(side_effect=RuntimeError("inference failed"))
+    source = wav()
+    target = tmp_path / "existing.wav"
+    target.write_bytes(b"existing output")
+    with pytest.raises(RuntimeError, match="inference failed"):
+        run(source, target, 0.85)
+    assert target.read_bytes() == b"existing output"
+    assert not list(tmp_path.glob(".voiceforge-ai-*"))