path: root/lib/project/tests/test_gui.py

"""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]