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

"""Opt-in real Chromium/loopback smoke; no AI models. Uploads come from the browser.

Install Playwright externally, then run:
PRODUCER_BROWSER_TESTS=1 python -m pytest tests/test_gui_browser.py -q
Screenshots go to PRODUCER_BROWSER_ARTIFACTS (default /tmp/producer-browser-smoke).
The long overview case uses mocked fixed-size scan metadata, not 4-hour AI inference.
"""
from dataclasses import asdict
import hashlib
import json
import os
from pathlib import Path
import socket
import threading
import time

import numpy as np
import pytest

pytestmark = pytest.mark.skipif(os.environ.get("PRODUCER_BROWSER_TESTS") != "1", reason="Opt-in Playwright browser smoke")


@pytest.fixture
def studio(tmp_path, wav, bypass, ffmpeg, monkeypatch):
    pytest.importorskip("playwright.sync_api")
    import uvicorn
    from producer.gui import create_app
    from producer import gui_jobs

    source = wav(0.15 * np.sin(2 * np.pi * 440 * np.arange(48000 * 8) / 48000), name="Narration.wav")
    wav(np.zeros((100, 2)), name="Stereo.wav")
    (tmp_path / "delivery").mkdir()
    control = {"hold": False, "started": threading.Event(), "release": threading.Event()}
    actual = gui_jobs.render_preview

    def controlled_preview(*args):
        if control["hold"]:
            control["started"].set()
            assert control["release"].wait(15), "Browser test did not release held preview"
            control["hold"] = False
        return actual(*args)

    monkeypatch.setattr(gui_jobs, "render_preview", controlled_preview)
    sock = socket.socket()
    sock.bind(("127.0.0.1", 0))
    sock.listen(128)
    port = sock.getsockname()[1]
    app = create_app(port, output_dir=tmp_path, overrides=asdict(bypass))
    server = uvicorn.Server(uvicorn.Config(app, log_level="error", access_log=False, proxy_headers=False))
    thread = threading.Thread(target=server.run, kwargs={"sockets": [sock]}, daemon=True)
    thread.start()
    deadline = time.monotonic() + 10
    while not server.started and time.monotonic() < deadline:
        time.sleep(0.02)
    assert server.started
    try:
        yield {"url": f"http://127.0.0.1:{port}", "root": tmp_path, "source": source, "control": control}
    finally:
        control["release"].set()
        server.should_exit = True
        thread.join(15)
        sock.close()
        assert not thread.is_alive()


def test_gui_browser_real_workflow(studio):
    from playwright.sync_api import sync_playwright, expect

    screenshots = Path(os.environ.get("PRODUCER_BROWSER_ARTIFACTS", "/tmp/producer-browser-smoke"))
    screenshots.mkdir(parents=True, exist_ok=True)
    before = hashlib.sha256(studio["source"].read_bytes()).hexdigest()
    with sync_playwright() as playwright:
        browser = playwright.chromium.launch(headless=True, channel=os.environ.get("PRODUCER_BROWSER_CHANNEL") or None, args=["--no-sandbox"])
        context = browser.new_context(viewport={"width": 1360, "height": 1000}, accept_downloads=True)
        page = context.new_page()
        errors, console_errors, media_requests = [], [], []
        page.on("pageerror", lambda exc: errors.append(str(exc)))
        page.on("console", lambda msg: console_errors.append(msg.text) if msg.type == "error" else None)
        page.on("request", lambda req: media_requests.append(req.url) if "/api/results/" in req.url else None)
        page.goto(studio["url"])
        expect(page.locator("#workspace")).not_to_have_attribute("disabled", "")
        assert "token" not in page.locator("body").inner_text()
        assert page.locator("[data-setting]").count() == 47
        expect(page.locator("#setting-denoiser")).to_have_value("none")
        # Dependent controls of an inactive module are hidden entirely.
        expect(page.locator("#setting-denoise_strength")).to_be_hidden()
        expect(page.locator("#setting-fft_reduction_db")).to_be_hidden()
        expect(page.locator("#setting-highpass_hz")).to_be_hidden()
        expect(page.locator("#setting-highpass_hz")).to_have_value("70")
        # Stereo input is downmixed safely to mono and both recordings load.
        page.locator("#pick-files").set_input_files([studio["root"] / "Stereo.wav", studio["source"]])
        expect(page.locator("#error")).not_to_be_visible()
        expect(page.locator("#wave-status")).to_contain_text("2,048", timeout=15000)
        expect(page.locator("#input-info")).to_contain_text("00:00:08")
        # Both drag selection and keyboard-accessible exact inputs.
        page.locator("#waveform").scroll_into_view_if_needed()
        box = page.locator("#waveform").bounding_box()
        page.mouse.move(box["x"] + box["width"] * .25, box["y"] + 80)
        page.mouse.down()
        page.mouse.move(box["x"] + box["width"] * .65, box["y"] + 80, steps=5)
        page.mouse.up()
        assert page.locator("#selection-start").input_value() != "00:00:00"
        page.locator("#selection-start").fill("00:00:02")
        page.locator("#selection-end").fill("00:00:05")
        page.locator("#zoom-selection").click()
        page.locator("#zoom-full").click()
        # Explicit render, requested pacing+MP3 preserved in snapshot but bypassed internally.
        page.locator("#setting-trim_silence").check()
        page.locator("#setting-mp3").check()
        expect(page.locator("#setting-mp3_bitrate")).not_to_be_hidden()
        page.locator("#render-preview").click()
        expect(page.locator("#comparison-status")).to_contain_text("Ready", timeout=30000)
        assert len(media_requests) == 2
        page.locator("#compare").scroll_into_view_if_needed()
        page.locator("#play").click()
        expect(page.locator("#play")).to_have_text("Pause")
        page.wait_for_timeout(350)
        position = page.locator("#audio-original").evaluate("audio => audio.currentTime")
        page.locator("#listen-processed").click()
        expect(page.locator("#listen-processed")).to_have_attribute("aria-pressed", "true")
        assert page.locator("#audio-processed").evaluate("audio => audio.currentTime") >= position - .15
        assert page.locator("#audio-original").evaluate("audio => audio.paused")
        assert page.locator("#audio-processed").evaluate("audio => audio.loop")
        page.locator("#match").check()
        expect(page.locator("#monitoring-note")).to_contain_text("Monitoring only")
        page.locator("#play").click()
        expect(page.locator("#meters")).to_contain_text("not a full-export measurement")
        page.locator("#comparison-name").fill("Gentle <snapshot>")
        page.locator("#save-comparison").click()
        expect(page.locator("#comparisons")).to_contain_text("Gentle <snapshot>")
        assert page.locator("#comparisons snapshot").count() == 0
        page.locator("#presets-panel summary").click()
        page.locator("#preset-name").fill("My sound")
        page.locator("#save-preset").click()
        expect(page.locator("#presets")).to_contain_text("My sound")
        with page.expect_download() as downloaded:
            page.locator("#download-preset").click()
        toml_path = screenshots / "roundtrip-preset.toml"
        downloaded.value.save_as(toml_path)
        assert "trim_silence = true" in toml_path.read_text()
        # Editing a control invalidates the comparison; importing the preset restores it.
        page.locator("#setting-eq").check()
        expect(page.locator("#comparison-status")).to_contain_text("changed since preview")
        page.locator("#import-preset").set_input_files(toml_path)
        expect(page.locator("#setting-eq")).not_to_be_checked()
        # The saved copy cannot change when controls are edited later.
        page.locator("#setting-eq").check()
        saved = page.evaluate("JSON.parse(localStorage.getItem(Object.keys(localStorage).find(k => k.startsWith('producer-gui-v1'))))")
        assert saved["comparisons"][0]["snapshot"]["settings"]["eq"] is False
        assert saved["presets"][0]["settings"]["eq"] is False
        # Delay real preview completion. Editing after submission must never publish it as current.
        studio["control"]["hold"] = True
        page.locator("#render-preview").click()
        assert studio["control"]["started"].wait(10)
        old_source = page.locator("#comparison-source").inner_text()
        page.locator("#setting-eq").uncheck()
        studio["control"]["release"].set()
        expect(page.locator("#jobs")).to_contain_text("Older snapshot; not loaded", timeout=30000)
        assert page.locator("#comparison-source").inner_text() == old_source
        assert len(media_requests) == 2
        # Restore is an explicit render and restores the original immutable requested settings.
        page.locator("#comparisons button", has_text="Restore & render").click()
        expect(page.locator("#setting-eq")).not_to_be_checked()
        expect(page.locator("#comparison-status")).to_contain_text("Ready", timeout=30000)
        # Cross-field errors do not silently enable the limiter.
        page.locator("#setting-normalize").check()
        expect(page.locator("#setting-target_lufs")).not_to_be_hidden()
        page.locator("#render-preview").click()
        expect(page.locator("#error-text")).to_contain_text("limiter")
        assert not page.locator("#setting-limiter").is_checked()
        page.locator("#dismiss-error").click()
        page.locator("#setting-normalize").uncheck()
        page.locator("#output-dir").fill(str(studio["root"] / "delivery"))
        preview_requests = len(media_requests)
        page.locator("#prepare-export").click()
        # Audio is written directly into the chosen output directory; no
        # pacing dialog exists anymore, and nothing exists to overwrite yet.
        expect(page.locator("#export-results")).to_contain_text("Full recording exported", timeout=30000)
        assert (studio["root"] / "delivery" / "Narration.custom.wav").is_file()
        assert (studio["root"] / "delivery" / "Narration.custom.mp3").is_file()
        reports = list((studio["root"] / "producer-root" / "logs").glob("Narration.custom.*.report.json"))
        assert len(reports) == 1
        import soundfile as sf
        assert sf.info(studio["root"] / "delivery" / "Narration.custom.wav").duration == pytest.approx(8, abs=.01)
        assert len(media_requests) == preview_requests, "Full exports must not be fetched into browser blobs"
        # A second export targets the same audio file: the browser asks, and
        # the red Overwrite confirmation replaces it.
        page.locator("#prepare-export").click()
        expect(page.locator("#overwrite-dialog")).to_be_visible()
        expect(page.locator("#overwrite-paths")).to_contain_text("Narration.custom.wav")
        page.locator("#overwrite-confirm").click()
        # Only the latest export stays on display; a new one replaces the card.
        expect(page.locator("#export-results .export-result")).to_have_count(1, timeout=30000)
        # The replaced export also leaves its own timestamped report.
        deadline = time.monotonic() + 30
        while time.monotonic() < deadline and len(list((studio["root"] / "producer-root" / "logs").glob("Narration.custom.*.report.json"))) < 2:
            time.sleep(0.2)
        assert (studio["root"] / "delivery" / "Narration.custom.wav").is_file()
        assert not list((studio["root"] / "delivery").glob("producer-export-*"))
        assert len(list((studio["root"] / "producer-root" / "logs").glob("Narration.custom.*.report.json"))) == 2
        assert hashlib.sha256(studio["source"].read_bytes()).hexdigest() == before
        page.locator("#compare").scroll_into_view_if_needed()
        page.screenshot(path=str(screenshots / "desktop-comparison.png"), full_page=True)
        page.locator("#compare").screenshot(path=str(screenshots / "comparison-panel.png"))
        # Reload keeps local snapshots, not misleading current audio.
        page.reload()
        expect(page.locator("#workspace")).not_to_have_attribute("disabled", "")
        expect(page.locator("#comparisons")).to_contain_text("Gentle <snapshot>")
        expect(page.locator("#play")).to_be_disabled()
        expect(page.locator("#export-results .export-result")).to_have_count(1, timeout=15000)
        page.set_viewport_size({"width": 390, "height": 844})
        page.screenshot(path=str(screenshots / "mobile-studio.png"), full_page=True)
        page.evaluate("window.scrollTo(0, 0)")
        page.screenshot(path=str(screenshots / "mobile-entry.png"))
        assert page.evaluate("document.documentElement.scrollWidth <= innerWidth")
        assert errors == []
        # Expected HTTP errors for stereo and the invalid constraint are allowed.
        assert not [msg for msg in console_errors if "Content Security Policy" in msg or "Refused" in msg]
        browser.close()


def test_gui_browser_long_overview(studio):
    from playwright.sync_api import sync_playwright, expect

    with sync_playwright() as playwright:
        browser = playwright.chromium.launch(headless=True, channel=os.environ.get("PRODUCER_BROWSER_CHANNEL") or None, args=["--no-sandbox"])
        page = browser.new_page(viewport={"width": 1200, "height": 900})
        requests, errors = [], []
        page.on("request", lambda request: requests.append(request.url))
        page.on("pageerror", lambda exc: errors.append(str(exc)))
        page.goto(studio["url"])
        # Four-hour metadata plus fixed-size peaks: browser never receives full recording samples.
        info = {"upload_id": "long-upload", "name": "Narration.wav", "input_id": "long-identity",
                "frames": 14400 * 8000, "duration": 14400, "sample_rate": 8000,
                "channels": 1, "subtype": "PCM_16"}
        result = {"input_id": "long-identity", "duration": 14400, "frames": info["frames"], "sample_rate": 8000,
                  "bins": 2048, "min": [-.3] * 2048, "max": [.4] * 2048}
        job = {"id": "mock-scan", "kind": "scan",
               "snapshot": {"upload_id": "long-upload", "name": "Narration.wav", "input_id": "long-identity"},
               "status": "succeeded", "progress": {"stage": "Done", "completed": 1, "total": 1},
               "created": time.time(), "result": result, "error": None, "cancel_requested": False, "cached": False}
        page.route("**/api/uploads*", lambda route: route.fulfill(json=info))
        page.route("**/api/jobs", lambda route: route.fulfill(json=job if route.request.method == "POST" else {"jobs": [job]}))
        page.locator("#pick-files").set_input_files(studio["source"])
        expect(page.locator("#wave-status")).to_contain_text("2,048", timeout=10000)
        expect(page.locator("#wave-end")).to_have_text("04:00:00")
        page.locator("#selection-start").fill("03:59:30")
        page.locator("#selection-end").fill("04:00:00")
        expect(page.locator("#selection-note")).to_contain_text("30 seconds selected")
        page.locator("#zoom-selection").click()
        expect(page.locator("#wave-start")).to_have_text("03:59:15")
        assert len(json.dumps(result)) < 50000
        assert not [url for url in requests if "/api/results/" in url]
        assert errors == []
        browser.close()