1 files changed, 260 insertions, 0 deletions
diff --git a/lib/project/tests/test_gui.py b/lib/project/tests/test_gui.py
new file mode 100644
index 0000000..c9e5285
--- /dev/null
+++ b/lib/project/tests/test_gui.py
@@ -0,0 +1,260 @@
+"""Optional GUI API, origin/root boundary, snapshots, and real sound previews."""
+from dataclasses import asdict
+import hashlib
+from pathlib import Path
+import struct
+import time
+
+import numpy as np
+import pytest
+import soundfile as sf
+
+pytest.importorskip("fastapi")
+from fastapi.testclient import TestClient
+
+from producer.config import Settings
+from producer.gui import create_app
+from producer import gui_jobs
+
+TOKEN = "test-session-credential"
+AUTH = {"Authorization": f"Bearer {TOKEN}"}
+URL = "http://127.0.0.1:8765"
+
+
+@pytest.fixture
+def client(tmp_path):
+    with TestClient(create_app(tmp_path, 8765, TOKEN), base_url=URL, headers=AUTH) as client:
+        yield client
+
+
+def wait(client, job_id):
+    deadline = time.monotonic() + 30
+    while time.monotonic() < deadline:
+        job = client.get(f"/api/jobs/{job_id}").json()
+        if job["status"] in ("succeeded", "failed", "cancelled"):
+            return job
+        time.sleep(0.02)
+    pytest.fail("Job failed to terminate")
+
+
+def submit(client, path, kind, **kwargs):
+    info = client.get("/api/input", params={"path": path.name})
+    assert info.status_code == 200, info.text
+    response = client.post("/api/jobs", json={"kind": kind, "path": path.name,
+                           "input_id": info.json()["input_id"], **kwargs})
+    assert response.status_code == 202, response.text
+    return response.json()
+
+
+def test_security_credentials_origins_hosts_and_headers(tmp_path):
+    with TestClient(create_app(tmp_path, 8765, TOKEN), base_url=URL) as client:
+        assert client.get("/").status_code == 200
+        assert client.get("/api/schema").status_code == 401
+        assert client.get(f"/api/schema?token={TOKEN}").status_code == 401
+        assert client.get("/api/schema", headers={"Cookie": f"token={TOKEN}"}).status_code == 401
+        for headers in [AUTH | {"Host": "evil.example:8765"},
+                        AUTH | {"Host": "localhost:8765"},
+                        AUTH | {"Origin": "https://evil.example"},
+                        AUTH | {"Origin": "null"},
+                        AUTH | {"Origin": "http://127.0.0.1:9999"},
+                        AUTH | {"Sec-Fetch-Site": "cross-site"}]:
+            assert client.get("/api/schema", headers=headers).status_code == 403
+            assert client.post("/api/jobs", json={}, headers=headers).status_code == 403
+        response = client.get("/api/schema", headers=AUTH | {"Origin": URL})
+        assert response.status_code == 200
+        assert response.headers["Cache-Control"] == "no-store"
+        assert response.headers["Referrer-Policy"] == "no-referrer"
+        assert "frame-ancestors 'none'" in response.headers["Content-Security-Policy"]
+        assert "Access-Control-Allow-Origin" not in response.headers
+        assert TOKEN not in response.text
+        assert client.get("/openapi.json").status_code == 404
+        assert client.post("/api/resolve", content=b"x" * 65537, headers=AUTH).status_code == 413
+        assert client.post("/api/resolve", content="{", headers=AUTH).status_code == 400
+
+
+def test_root_browser_symlinks_traversal_and_no_arbitrary_download(client, tmp_path, wav):
+    source = wav()
+    outside = tmp_path.parent / "outside.wav"
+    outside.write_bytes(b"not accessible")
+    (tmp_path / "escape.wav").symlink_to(outside)
+    (tmp_path / "escape-dir").symlink_to(tmp_path.parent, target_is_directory=True)
+    (tmp_path / "inside.wav").symlink_to(source)
+    (tmp_path / "note.txt").write_text("not exposed")
+    (tmp_path / "subdir").mkdir()
+    listing = client.get("/api/files").json()
+    assert listing["parent"] is None
+    assert {item["name"] for item in listing["entries"]} == {"input.wav", "inside.wav", "subdir"}
+    for path in ("../outside.wav", str(outside), "escape.wav", "escape-dir", "subdir/../../outside.wav"):
+        assert client.get("/api/input", params={"path": path}).status_code == 400
+    assert client.get("/api/files", params={"path": ".."}).status_code == 400
+    assert client.get("/api/files", params={"limit": 201}).status_code == 400
+    assert len(client.get("/api/files", params={"limit": 1}).json()["entries"]) == 1
+    assert client.get("/api/files", params={"limit": 1}).json()["next_offset"] == 1
+    assert client.get("/assets/../gui.py").status_code == 404
+    assert client.get("/assets/%2e%2e/gui.py").status_code == 404
+    assert client.get("/api/results/missing/original.wav").status_code == 404
+
+
+def test_schema_complete_strict_resolution_and_preset_roundtrip(client, bypass):
+    schema = client.get("/api/schema").json()
+    assert {item["name"] for item in schema["fields"]} == set(asdict(Settings()))
+    assert all(item["group"] and item["help"] for item in schema["fields"])
+    assert schema["settings"] == asdict(Settings())
+    for value in ({"unknown": 1}, {"normalize": "false"}, {"target_lufs": -100},
+                  {"normalize": True, "limiter": False}, {"sample_rate": 1234}):
+        assert client.post("/api/resolve", json={"settings": value}).status_code == 400
+    response = client.post("/api/presets/download", json={"settings": asdict(bypass)})
+    assert response.status_code == 200
+    assert "attachment" in response.headers["Content-Disposition"]
+    imported = client.post("/api/presets/import", json={"toml": response.text})
+    assert imported.json()["settings"] == asdict(bypass)
+    assert client.post("/api/presets/import", json={"toml": "no_such_setting = 2"}).status_code == 400
+    assert client.post("/api/presets/import", json={"toml": "["}).status_code == 400
+
+
+def test_config_precedence_and_startup_snapshot(tmp_path):
+    config = tmp_path / "config.toml"
+    config.write_text('profile = "narrator"\ntarget_lufs = -25\n')
+    app = create_app(tmp_path, 8765, TOKEN, config=config, profile="radio", overrides={"target_lufs": -22})
+    config.write_text("this file changed after startup")
+    with TestClient(app, base_url=URL, headers=AUTH) as client:
+        settings = client.get("/api/schema").json()["settings"]
+        assert settings["profile"] == "radio" and settings["compressor_ratio"] == 3.5
+        assert settings["target_lufs"] == -22
+        resolved = client.post("/api/resolve", json={"settings": {"target_lufs": -18, "profile": "natural"}})
+        assert resolved.json()["settings"]["target_lufs"] == -18
+        assert resolved.json()["settings"]["compressor_ratio"] == 2
+
+
+def test_scan_fixed_bins_cache_and_stale_identity(client, wav, monkeypatch):
+    source = wav()
+    original = source.read_bytes()
+    first = wait(client, submit(client, source, "scan")["id"])
+    assert first["status"] == "succeeded", first
+    result = first["result"]
+    assert len(result["min"]) == len(result["max"]) == 2048
+    assert max(result["max"]) == pytest.approx(0.15, abs=1e-6)
+    monkeypatch.setattr(gui_jobs, "waveform", lambda *args: pytest.fail("Expected repeated-result cache"))
+    second = wait(client, submit(client, source, "scan")["id"])
+    assert second["cached"] and second["result"] == result
+    assert source.read_bytes() == original
+    old_identity = first["snapshot"]["input_id"]
+    with source.open("ab") as handle:
+        handle.write(b"changed")
+    response = client.post("/api/jobs", json={"kind": "scan", "path": source.name, "input_id": old_identity})
+    assert response.status_code == 400
+    assert "changed" in response.text
+
+
+def test_real_preview_context_exact_crop_bypass_and_export_trim(client, wav, bypass, ffmpeg):
+    t = np.arange(48000 * 8) / 48000
+    samples = 0.15 * np.sin(2 * np.pi * 440 * t)
+    source = wav(samples)
+    before = hashlib.sha256(source.read_bytes()).hexdigest()
+    requested = asdict(bypass) | {"trim_silence": True, "mp3": True}
+    queued = submit(client, source, "preview", settings=requested, start=2, duration=3)
+    preview = wait(client, queued["id"])
+    assert preview["status"] == "succeeded", preview
+    result = preview["result"]
+    assert preview["snapshot"]["settings"] == requested
+    assert result["requested_settings"] == requested
+    assert result["effective_settings"] == requested | {"trim_silence": False, "mp3": False}
+    assert result["normalization_mode"] == "disabled"
+    assert result["context_start"] == 0 and result["context_end"] == 8
+    assert result["approximate"] is True
+    assert "Pacing edits are not previewed" in result["notice"]
+    import io
+    originals = client.get(result["audio"]["original"])
+    processed = client.get(result["audio"]["processed"])
+    for response in (originals, processed):
+        assert response.status_code == 200
+        info = sf.info(io.BytesIO(response.content))
+        assert info.frames == 3 * 48000
+        assert info.subtype == "PCM_16"
+    original_audio, _ = sf.read(io.BytesIO(originals.content))
+    processed_audio, _ = sf.read(io.BytesIO(processed.content))
+    np.testing.assert_allclose(original_audio, processed_audio, atol=1 / 32768)
+    np.testing.assert_allclose(original_audio, samples[2 * 48000:5 * 48000], atol=1 / 32768)
+    assert result["monitoring"]["available"]
+    assert result["monitoring"]["original_gain_db"] <= 0
+    assert result["monitoring"]["processed_gain_db"] <= 0
+    repeated = wait(client, submit(client, source, "preview", settings=requested, start=2, duration=3)["id"])
+    assert repeated["cached"] and repeated["result"]["result_id"] == result["result_id"]
+    changed = wait(client, submit(client, source, "preview", settings=requested | {"highpass": True}, start=2, duration=3)["id"])
+    assert not changed["cached"] and changed["result"]["result_id"] != result["result_id"]
+    for _ in range(2):
+        exported = wait(client, submit(client, source, "export", settings=requested)["id"])
+        assert exported["status"] == "succeeded", exported
+        assert exported["result"]["settings"] == requested
+        assert len(exported["result"]["paths"]) == 3
+        assert "not auditioned" in exported["result"]["notice"]
+        for path in exported["result"]["paths"]:
+            assert (source.parent / path).is_file()
+    assert len(list(source.parent.glob("producer-export-*"))) == 2
+    assert hashlib.sha256(source.read_bytes()).hexdigest() == before
+    # A cached audio URL cannot read stale results after the recording changes.
+    with source.open("ab") as handle:
+        handle.write(b"changed")
+    assert client.get(result["audio"]["original"]).status_code == 400
+
+
+def test_preview_bounds_settings_errors_and_unsupported_input(client, wav):
+    source = wav()
+    for start, duration in ((-1, 1), (0, 121), (0, 0), (2, 2), (True, 1)):
+        info = client.get("/api/input", params={"path": source.name}).json()
+        response = client.post("/api/jobs", json={"kind": "preview", "path": source.name,
+            "input_id": info["input_id"], "start": start, "duration": duration})
+        assert response.status_code == 400
+    stereo = wav(np.zeros((20, 2)), name="stereo.wav")
+    assert "Stereo" in client.get("/api/input", params={"path": stereo.name}).json()["detail"]
+    surround = wav(np.zeros((20, 3)), name="surround.wav")
+    assert client.get("/api/input", params={"path": surround.name}).status_code == 400
+    source.write_bytes(b"not a wav")
+    response = client.get("/api/input", params={"path": source.name})
+    assert response.status_code == 400 and "Unsupported" in response.text
+
+
+def test_no_total_duration_cap_and_seek_to_end_of_four_hour_input(tmp_path, progress):
+    # Sparse mono PCM16 WAV. No real recordings touched and no huge allocation.
+    source = tmp_path / "four-hours.wav"
+    sr, seconds = 8000, 4 * 60 * 60
+    data_bytes = sr * seconds * 2
+    header = (b"RIFF" + struct.pack("<I", 36 + data_bytes) + b"WAVEfmt "
+              + struct.pack("<IHHIIHH", 16, 1, 1, sr, sr * 2, 2, 16)
+              + b"data" + struct.pack("<I", data_bytes))
+    with source.open("wb") as handle:
+        handle.write(header)
+        handle.truncate(44 + data_bytes)
+    assert gui_jobs.input_info(source)["duration"] == seconds
+    target = tmp_path / "end.wav"
+    gui_jobs.copy_range(source, target, seconds - 2, 2, progress)
+    assert sf.info(target).frames == 2 * sr
+
+
+def test_waveform_buffers_bounded_and_scan_cancellable(wav, monkeypatch):
+    source = wav()
+    actual = sf.SoundFile
+    reads = []
+    class CheckedFile:
+        def __init__(self, *args, **kwargs):
+            self.source = actual(*args, **kwargs)
+        def __enter__(self):
+            return self
+        def __exit__(self, *args):
+            self.source.close()
+        def __len__(self):
+            return len(self.source)
+        def __getattr__(self, name):
+            return getattr(self.source, name)
+        def blocks(self, **kwargs):
+            assert kwargs["blocksize"] <= 65536
+            for block in self.source.blocks(**kwargs):
+                reads.append(len(block))
+                yield block
+    monkeypatch.setattr(gui_jobs.sf, "SoundFile", CheckedFile)
+    def cancel(stage, completed, total):
+        if completed:
+            raise gui_jobs.Cancelled()
+    with pytest.raises(gui_jobs.Cancelled):
+        gui_jobs.waveform(source, cancel)
+    assert reads == [65536]