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

"""Optional GUI API, origin boundary, uploads, 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

URL = "http://127.0.0.1:8765"


@pytest.fixture
def client(tmp_path):
    with TestClient(create_app(8765, output_dir=tmp_path), base_url=URL) as client:
        yield client


_UPLOADED = {}


def upload(client, path, name=None):
    """Upload once per (file, name) and reuse the entry, like the browser UI."""
    key = (id(path), name)
    if key not in _UPLOADED:
        response = client.post("/api/uploads", params={"name": name or path.name}, content=path.read_bytes())
        assert response.status_code == 200, response.text
        _UPLOADED[key] = response.json()
    return _UPLOADED[key]


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):
    entry = upload(client, path)
    response = client.post("/api/jobs", json={"kind": kind, "upload_id": entry["upload_id"],
                           "input_id": entry["input_id"], **kwargs})
    assert response.status_code == 202, response.text
    return response.json()


def test_security_origins_hosts_and_headers(tmp_path):
    with TestClient(create_app(8765), base_url=URL) as client:
        assert client.get("/").status_code == 200
        assert client.get("/api/schema").status_code == 200
        assert "token" not in client.get("/").text.lower()
        for headers in [{"Host": "evil.example:8765"},
                        {"Host": "localhost:8765"},
                        {"Origin": "https://evil.example"},
                        {"Origin": "null"},
                        {"Origin": "http://127.0.0.1:9999"},
                        {"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={"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 client.get("/openapi.json").status_code == 404
        assert client.post("/api/resolve", content=b"x" * 65537).status_code == 413
        assert client.post("/api/resolve", content="{").status_code == 400


def test_upload_pipeline_validates_names_and_content(client, tmp_path, wav):
    source = wav()
    entry = upload(client, source)
    assert entry["name"] == "input.wav"
    assert entry["duration"] == pytest.approx(3)
    info = client.get(f"/api/uploads/{entry['upload_id']}").json()
    assert info["input_id"] == entry["input_id"]
    # Display names are sanitized; they never reach the filesystem.
    escaped = upload(client, source, name="../evil name<>.wav")
    assert ".." not in escaped["name"] and "/" not in escaped["name"]
    assert client.get("/api/uploads/unknown-id").status_code == 400
    (tmp_path / "note.txt").write_text("not a wav")
    assert client.post("/api/uploads", params={"name": "note.txt"}, content=b"nope").status_code == 400
    assert client.post("/api/uploads", params={"name": "x.wav"}, content=b"not a wav").status_code == 400
    stereo = wav(np.zeros((20, 2)), name="stereo.wav")
    response = client.post("/api/uploads", params={"name": "stereo.wav"}, content=stereo.read_bytes())
    assert response.status_code == 200 and response.json()["channels"] == 2
    assert "downmixed" in response.json()["notice"]
    assert "downmixed" in client.get(f"/api/uploads/{response.json()['upload_id']}").text
    surround = wav(np.zeros((20, 3)), name="surround.wav")
    assert client.post("/api/uploads", params={"name": "surround.wav"}, content=surround.read_bytes()).status_code == 400
    # Jobs require a known upload and a matching identity.
    assert client.post("/api/jobs", json={"kind": "scan", "upload_id": "unknown",
                       "input_id": "x"}).status_code == 400
    assert client.post("/api/jobs", json={"kind": "scan", "upload_id": entry["upload_id"],
                       "input_id": "wrong"}).status_code == 400
    assert "changed" in client.post("/api/jobs", json={"kind": "scan", "upload_id": entry["upload_id"],
                                    "input_id": "wrong"}).text


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"])
    help_for = {item["name"]: item["help"] for item in schema["fields"]}
    assert len(set(help_for.values())) == len(help_for)
    assert "cannot be 0" in help_for["fft_reduction_db"]
    assert "Uncheck" in help_for["highpass"]
    assert "Uncheck" in help_for["pause_silence"] and "Uncheck" in help_for["trim_silence"]
    assert "0 to disable" in help_for["hum_hz"]
    assert schema["settings"] == asdict(Settings())
    assert Path(schema["output_dir"]).is_absolute()
    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(8765, config=config, profile="radio", overrides={"target_lufs": -22})
    config.write_text("this file changed after startup")
    with TestClient(app, base_url=URL) 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


def test_real_preview_context_exact_crop_bypass_and_export_trim(client, tmp_path, 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,
                                       output_dir=str(tmp_path))["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 Path(path).is_absolute() and Path(path).is_file()
    assert len(list(tmp_path.glob("producer-export-*"))) == 2
    assert hashlib.sha256(source.read_bytes()).hexdigest() == before


def test_export_output_directory_rules(client, wav, bypass, tmp_path):
    source = wav()
    entry = upload(client, source)
    requested = asdict(bypass)
    base = {"kind": "export", "upload_id": entry["upload_id"], "input_id": entry["input_id"],
            "settings": requested}
    assert client.post("/api/jobs", json={**base, "output_dir": "relative/folder"}).status_code == 400
    assert client.post("/api/jobs", json={**base, "output_dir": str(tmp_path / "missing")}).status_code == 400
    missing = client.post("/api/jobs", json={**base})
    assert missing.status_code == 202
    job = wait(client, missing.json()["id"])
    assert job["status"] == "succeeded", job
    for path in job["result"]["paths"]:
        assert Path(path).is_absolute() and Path(path).is_file()
        Path(path).relative_to(tmp_path)

def test_preview_bounds_and_empty_inputs(client, wav):
    source = wav()
    entry = upload(client, source)
    for start, duration in ((-1, 1), (0, 121), (0, 0), (2, 2), (True, 1)):
        response = client.post("/api/jobs", json={"kind": "preview", "upload_id": entry["upload_id"],
            "input_id": entry["input_id"], "start": start, "duration": duration})
        assert response.status_code == 400


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]