aboutsummaryrefslogtreecommitdiff
path: root/lib/project/tests
diff options
context:
space:
mode:
Diffstat (limited to 'lib/project/tests')
-rw-r--r--lib/project/tests/gui_state.test.mjs10
-rw-r--r--lib/project/tests/test_gui.py162
-rw-r--r--lib/project/tests/test_gui_assets.py14
-rw-r--r--lib/project/tests/test_gui_browser.py76
-rw-r--r--lib/project/tests/test_gui_browser_races.py27
-rw-r--r--lib/project/tests/test_gui_jobs.py20
-rw-r--r--lib/project/tests/test_gui_launcher.py16
-rw-r--r--lib/project/tests/test_pipeline.py49
8 files changed, 214 insertions, 160 deletions
diff --git a/lib/project/tests/gui_state.test.mjs b/lib/project/tests/gui_state.test.mjs
index 10775eb..0556a4a 100644
--- a/lib/project/tests/gui_state.test.mjs
+++ b/lib/project/tests/gui_state.test.mjs
@@ -1,6 +1,6 @@
import test from 'node:test';
import assert from 'node:assert/strict';
-import {parseTime, formatTime, selection, stable, matches, monitoringGain, absolutePath} from '../src/producer/web/state.js';
+import {parseTime, formatTime, selection, stable, matches, monitoringGain} from '../src/producer/web/state.js';
test('exact excerpt times work beyond four hours without wrapping', () => {
assert.equal(parseTime('05:12:03.125'), 18723.125);
@@ -12,11 +12,11 @@ test('exact excerpt times work beyond four hours without wrapping', () => {
for (const [start, end] of [['0', '121'], ['2', '1'], ['0', '0'], ['14399', '14401']]) assert.throws(() => selection(start, end, 14400));
});
-test('preview identity includes source, exact selection and every requested setting', () => {
- const snapshot = {path: 'take.wav', input_id: 'identity', start: 10, duration: 30, settings: {mp3: true, trim_silence: true, denoiser: 'none'}};
+test('preview identity includes uploaded source, exact selection and every requested setting', () => {
+ const snapshot = {upload_id: 'upload', input_id: 'identity', start: 10, duration: 30, settings: {mp3: true, trim_silence: true, denoiser: 'none'}};
assert.ok(matches(snapshot, structuredClone(snapshot)));
assert.ok(matches(snapshot, {...snapshot, settings: {denoiser: 'none', trim_silence: true, mp3: true}}));
- for (const change of [{path: 'other.wav'}, {input_id: 'changed'}, {start: 11}, {duration: 29}, {settings: {...snapshot.settings, trim_silence: false}}]) assert.equal(matches(snapshot, {...snapshot, ...change}), false);
+ for (const change of [{upload_id: 'other'}, {input_id: 'changed'}, {start: 11}, {duration: 29}, {settings: {...snapshot.settings, trim_silence: false}}]) assert.equal(matches(snapshot, {...snapshot, ...change}), false);
assert.equal(matches(snapshot, null), false);
assert.equal(stable({b: 2, a: 1}), stable({a: 1, b: 2}));
});
@@ -26,6 +26,4 @@ test('monitoring is attenuation only and independent of export settings', () =>
assert.equal(monitoringGain(6, true), 1);
assert.equal(monitoringGain(NaN, true), 1);
assert.ok(Math.abs(monitoringGain(-6, true) - 0.501187) < 0.000001);
- assert.equal(absolutePath('/recordings', 'delivery/master.wav'), '/recordings/delivery/master.wav');
- assert.equal(absolutePath('/', './master.wav'), '/master.wav');
});
diff --git a/lib/project/tests/test_gui.py b/lib/project/tests/test_gui.py
index c9e5285..e490e99 100644
--- a/lib/project/tests/test_gui.py
+++ b/lib/project/tests/test_gui.py
@@ -1,4 +1,4 @@
-"""Optional GUI API, origin/root boundary, snapshots, and real sound previews."""
+"""Optional GUI API, origin boundary, uploads, snapshots, and real sound previews."""
from dataclasses import asdict
import hashlib
from pathlib import Path
@@ -16,17 +16,28 @@ 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:
+ 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:
@@ -38,68 +49,79 @@ def wait(client, job_id):
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})
+ 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_credentials_origins_hosts_and_headers(tmp_path):
- with TestClient(create_app(tmp_path, 8765, TOKEN), base_url=URL) as client:
+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 == 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").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=AUTH | {"Origin": URL})
+ 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 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
+ assert client.post("/api/resolve", content=b"x" * 65537).status_code == 413
+ assert client.post("/api/resolve", content="{").status_code == 400
-def test_root_browser_symlinks_traversal_and_no_arbitrary_download(client, tmp_path, wav):
+def test_upload_pipeline_validates_names_and_content(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
+ 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
@@ -115,9 +137,9 @@ def test_schema_complete_strict_resolution_and_preset_roundtrip(client, bypass):
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})
+ 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, headers=AUTH) as client:
+ 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
@@ -138,15 +160,9 @@ def test_scan_fixed_bins_cache_and_stale_identity(client, wav, monkeypatch):
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):
+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)
@@ -183,35 +199,41 @@ def test_real_preview_context_exact_crop_bypass_and_export_trim(client, wav, byp
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"])
+ 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 (source.parent / path).is_file()
- assert len(list(source.parent.glob("producer-export-*"))) == 2
+ 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
- # 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):
+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)):
- 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})
+ 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
- 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):
diff --git a/lib/project/tests/test_gui_assets.py b/lib/project/tests/test_gui_assets.py
index b7f949a..961f15e 100644
--- a/lib/project/tests/test_gui_assets.py
+++ b/lib/project/tests/test_gui_assets.py
@@ -25,8 +25,8 @@ class AssetReferences(HTMLParser):
self.stylesheets.append(attrs.get("href"))
-def test_gui_packaged_external_assets_and_guided_workflow(tmp_path):
- with TestClient(create_app(tmp_path, 8765, "test-token"), base_url="http://127.0.0.1:8765") as client:
+def test_gui_packaged_external_assets_and_condensed_workflow(tmp_path):
+ with TestClient(create_app(8765, output_dir=tmp_path), base_url="http://127.0.0.1:8765") as client:
page = client.get("/")
assert page.status_code == 200
parser = AssetReferences()
@@ -35,12 +35,14 @@ def test_gui_packaged_external_assets_and_guided_workflow(tmp_path):
assert parser.stylesheets == ["/assets/style.css"]
assert not parser.inline_attributes
assert "unsafe-inline" not in page.headers["Content-Security-Policy"]
- assert "Approximate sound comparison, not an exact export preview." in page.text
- for name in ("Choose your recording", "Try a short passage", "Compare the sound", "Export the full recording"):
- assert name in page.text
+ assert "approximate, not an exact export preview" in page.text
+ for name in ("Recording", "Settings", "Compare", "Export", "Activity"):
+ assert f">{name}</h2>" in page.text or name in page.text
+ for banned in ("#token", "private by design", "Make it sound like you"):
+ assert banned not in page.text
for path in parser.scripts + parser.stylesheets + ["/assets/state.js"]:
response = client.get(path)
assert response.status_code == 200, path
assert response.headers["Cache-Control"] == "no-store"
assert len(response.content) > 100
- assert client.get("/api/schema").status_code == 401
+ assert client.get("/api/schema").status_code == 200
diff --git a/lib/project/tests/test_gui_browser.py b/lib/project/tests/test_gui_browser.py
index 1e38964..385de44 100644
--- a/lib/project/tests/test_gui_browser.py
+++ b/lib/project/tests/test_gui_browser.py
@@ -1,4 +1,4 @@
-"""Opt-in real Chromium/loopback smoke; no AI models or recording uploads.
+"""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
@@ -8,8 +8,6 @@ The long overview case uses mocked fixed-size scan metadata, not 4-hour AI infer
from dataclasses import asdict
import hashlib
import json
-import os
-import re
from pathlib import Path
import socket
import threading
@@ -31,7 +29,6 @@ def studio(tmp_path, wav, bypass, ffmpeg, monkeypatch):
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()
- (tmp_path / "subfolder").mkdir()
control = {"hold": False, "started": threading.Event(), "release": threading.Event()}
actual = gui_jobs.render_preview
@@ -43,12 +40,11 @@ def studio(tmp_path, wav, bypass, ffmpeg, monkeypatch):
return actual(*args)
monkeypatch.setattr(gui_jobs, "render_preview", controlled_preview)
- token = "browser-test-private-credential"
sock = socket.socket()
sock.bind(("127.0.0.1", 0))
sock.listen(128)
port = sock.getsockname()[1]
- app = create_app(tmp_path, port, token, overrides=asdict(bypass))
+ 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()
@@ -57,7 +53,7 @@ def studio(tmp_path, wav, bypass, ffmpeg, monkeypatch):
time.sleep(0.02)
assert server.started
try:
- yield {"url": f"http://127.0.0.1:{port}", "token": token, "root": tmp_path, "source": source, "control": control}
+ yield {"url": f"http://127.0.0.1:{port}", "root": tmp_path, "source": source, "control": control}
finally:
control["release"].set()
server.should_exit = True
@@ -80,19 +76,21 @@ def test_gui_browser_real_workflow(studio):
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"] + "/#token=" + studio["token"])
+ page.goto(studio["url"])
expect(page.locator("#workspace")).not_to_have_attribute("disabled", "")
- assert "#" not in page.url
- assert studio["token"] not in page.locator("body").inner_text()
+ 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")
- expect(page.locator("#folder-up")).to_be_disabled()
+ # Dependent controls of a disabled effect are visibly inactive.
+ expect(page.locator("#setting-denoise_strength")).to_be_disabled()
+ expect(page.locator("#setting-fft_reduction_db")).to_be_disabled()
+ expect(page.locator("#setting-highpass_hz")).to_be_disabled()
+ expect(page.locator("#setting-highpass_hz")).to_have_value("70")
# Unsupported input errors are human-readable, not raw HTML.
- page.get_by_role("button", name=re.compile(r"WAV\s*Stereo.wav")).click()
+ page.locator("#pick-files").set_input_files([studio["root"] / "Stereo.wav", studio["source"]])
expect(page.locator("#error")).to_be_visible()
expect(page.locator("#error-text")).to_contain_text("Stereo")
page.locator("#dismiss-error").click()
- page.get_by_role("button", name=re.compile(r"WAV\s*Narration.wav")).click()
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.
@@ -110,6 +108,7 @@ def test_gui_browser_real_workflow(studio):
# 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_disabled()
page.locator("#render-preview").click()
expect(page.locator("#comparison-status")).to_contain_text("Ready", timeout=30000)
assert len(media_requests) == 2
@@ -140,39 +139,39 @@ def test_gui_browser_real_workflow(studio):
toml_path = screenshots / "roundtrip-preset.toml"
downloaded.value.save_as(toml_path)
assert "trim_silence = true" in toml_path.read_text()
- assert studio["token"] not in toml_path.read_text()
- page.locator("#setting-denoise_strength").fill("0.4")
+ # 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-denoise_strength")).to_have_value("0.85")
+ expect(page.locator("#setting-eq")).not_to_be_checked()
# The saved copy cannot change when controls are edited later.
- page.locator("#setting-denoise_strength").fill("0.5")
- saved = page.evaluate("JSON.parse(localStorage.getItem(Object.keys(localStorage).find(k => k.startsWith('producer-studio-v1:'))))")
- assert saved["comparisons"][0]["snapshot"]["settings"]["denoise_strength"] == .85
- assert saved["presets"][0]["settings"]["denoise_strength"] == .85
- assert studio["token"] not in json.dumps(saved)
+ 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-denoise_strength").fill("0.6")
+ 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-denoise_strength")).to_have_value("0.85")
+ 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_disabled()
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("delivery")
+ page.locator("#output-dir").fill(str(studio["root"] / "delivery"))
preview_requests = len(media_requests)
page.locator("#prepare-export").click()
expect(page.locator("#export-dialog")).to_be_visible()
@@ -195,7 +194,7 @@ def test_gui_browser_real_workflow(studio):
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 retains per-tab authorization and local snapshots, not misleading current audio.
+ # 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>")
@@ -206,18 +205,8 @@ def test_gui_browser_real_workflow(studio):
page.evaluate("window.scrollTo(0, 0)")
page.screenshot(path=str(screenshots / "mobile-entry.png"))
assert page.evaluate("document.documentElement.scrollWidth <= innerWidth")
- # A fresh tab without the private fragment cannot read API data.
- unauth = context.new_page()
- unauth.goto(studio["url"])
- expect(unauth.locator("#error-text")).to_contain_text("private URL")
- expect(unauth.locator("#workspace")).to_have_attribute("disabled", "")
- unauth.close()
- # Simulate server restart credential expiry and verify recovery guidance.
- page.route("**/api/jobs", lambda route: route.fulfill(status=401, json={"detail": "expired"}))
- expect(page.locator("#workspace")).to_have_attribute("disabled", "", timeout=10000)
- expect(page.locator("#connection")).to_contain_text("Session expired")
assert errors == []
- # Expected HTTP errors for stereo, invalid constraint and the expiry test are allowed.
+ # 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()
@@ -231,19 +220,20 @@ def test_gui_browser_long_overview(studio):
requests, errors = [], []
page.on("request", lambda request: requests.append(request.url))
page.on("pageerror", lambda exc: errors.append(str(exc)))
- page.goto(studio["url"] + "/#token=" + studio["token"])
- expect(page.locator("#workspace")).not_to_have_attribute("disabled", "")
+ page.goto(studio["url"])
# Four-hour metadata plus fixed-size peaks: browser never receives full recording samples.
- info = {"path": "Narration.wav", "input_id": "long-identity", "frames": 14400 * 8000,
- "duration": 14400, "sample_rate": 8000, "channels": 1, "subtype": "PCM_16"}
+ 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": {"path": "Narration.wav", "input_id": "long-identity"},
+ 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/input?*", lambda route: route.fulfill(json=info))
+ 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.get_by_role("button", name=re.compile(r"WAV\s*Narration.wav")).click()
+ 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")
diff --git a/lib/project/tests/test_gui_browser_races.py b/lib/project/tests/test_gui_browser_races.py
index fab60ba..5f2638b 100644
--- a/lib/project/tests/test_gui_browser_races.py
+++ b/lib/project/tests/test_gui_browser_races.py
@@ -15,8 +15,8 @@ def test_gui_browser_cancellation_is_not_pretend(studio):
with sync_playwright() as playwright:
browser = playwright.chromium.launch(headless=True, args=["--no-sandbox"])
page = browser.new_page()
- page.goto(studio["url"] + "/#token=" + studio["token"])
- page.get_by_role("button", name=re.compile(r"WAV\s*Narration.wav")).click()
+ page.goto(studio["url"])
+ page.locator("#pick-files").set_input_files(studio["source"])
expect(page.locator("#wave-status")).to_contain_text("2,048", timeout=10000)
studio["control"]["hold"] = True
page.locator("#render-preview").click()
@@ -48,8 +48,8 @@ def test_gui_browser_late_audio_is_aborted_and_urls_are_released(studio):
const create = URL.createObjectURL.bind(URL), revoke = URL.revokeObjectURL.bind(URL);
URL.createObjectURL = blob => { const url = create(blob); objectURLs.created.push(url); return url; };
URL.revokeObjectURL = url => { objectURLs.revoked.push(url); revoke(url); };""")
- page.goto(studio["url"] + "/#token=" + studio["token"])
- page.get_by_role("button", name=re.compile(r"WAV\s*Narration.wav")).click()
+ page.goto(studio["url"])
+ page.locator("#pick-files").set_input_files(studio["source"])
expect(page.locator("#wave-status")).to_contain_text("2,048", timeout=10000)
page.locator("#selection-start").fill("1")
page.locator("#selection-end").fill("4")
@@ -99,10 +99,9 @@ def test_gui_browser_nonzero_monitoring_gain_is_playback_only(studio):
AudioContext.prototype.createGain = function() {
const gain = createGain.call(this); monitorGains.push(gain); return gain;
};""")
- page.goto(studio["url"] + "/#token=" + studio["token"])
- page.get_by_role("button", name=re.compile(r"WAV\s*Narration.wav")).click()
+ page.goto(studio["url"])
+ page.locator("#pick-files").set_input_files(studio["source"])
expect(page.locator("#wave-status")).to_contain_text("2,048", timeout=10000)
- page.locator("#advanced-settings summary").click()
# Real FFmpeg leveling makes a measurable difference without any AI runtime.
page.locator("#setting-leveling").check()
page.locator("#setting-level_target_db").fill("-12")
@@ -137,9 +136,9 @@ def test_delayed_job_list_preserves_new_submission_intent(studio, kind):
with sync_playwright() as playwright:
browser = playwright.chromium.launch(headless=True, args=["--no-sandbox"])
page = browser.new_page()
- page.goto(studio["url"] + "/#token=" + studio["token"])
+ page.goto(studio["url"])
if kind == "preview":
- page.get_by_role("button", name=re.compile(r"WAV\s*Narration.wav")).click()
+ page.locator("#pick-files").set_input_files(studio["source"])
expect(page.locator("#wave-status")).to_contain_text("2,048", timeout=10000)
held = []
@@ -158,7 +157,7 @@ def test_delayed_job_list_preserves_new_submission_intent(studio, kind):
with page.expect_response(lambda response: response.request.method == "POST"
and response.url.endswith("/api/jobs")) as posted:
if kind == "scan":
- page.get_by_role("button", name=re.compile(r"WAV\s*Narration.wav")).click()
+ page.locator("#pick-files").set_input_files(studio["source"])
else:
page.locator("#render-preview").click()
assert posted.value.status == 202
@@ -186,8 +185,8 @@ def test_delayed_audio_resume_does_not_start_superseded_playback(studio, interru
window.releaseAudioResume = () => { resume.call(this).then(resolve); };
});
};""")
- page.goto(studio["url"] + "/#token=" + studio["token"])
- page.get_by_role("button", name=re.compile(r"WAV\s*Narration.wav")).click()
+ page.goto(studio["url"])
+ page.locator("#pick-files").set_input_files(studio["source"])
expect(page.locator("#wave-status")).to_contain_text("2,048", timeout=10000)
page.locator("#render-preview").click()
expect(page.locator("#comparison-status")).to_contain_text("Ready", timeout=30000)
@@ -200,8 +199,8 @@ def test_delayed_audio_resume_does_not_start_superseded_playback(studio, interru
page.locator("#setting-denoise_strength").fill("0.4")
else:
page.route("**/api/jobs", lambda route: route.fulfill(
- status=401, content_type="application/json", body='{"detail":"Expired"}'))
- expect(page.locator("#workspace")).to_have_attribute("disabled", "", timeout=10000)
+ status=500, content_type="application/json", body='{"detail":"down"}'))
+ expect(page.locator("#connection")).to_contain_text("interrupted", timeout=10000)
page.evaluate("releaseAudioResume()")
# Let both the resume and any erroneously scheduled play promise settle.
page.wait_for_timeout(300)
diff --git a/lib/project/tests/test_gui_jobs.py b/lib/project/tests/test_gui_jobs.py
index 6116a72..d44772f 100644
--- a/lib/project/tests/test_gui_jobs.py
+++ b/lib/project/tests/test_gui_jobs.py
@@ -7,8 +7,6 @@ import time
import pytest
from producer import gui_jobs, setup
-from producer.gui import rooted
-
def terminal(jobs, job_id):
deadline = time.monotonic() + 10
@@ -34,13 +32,13 @@ def test_single_worker_bounded_queue_and_running_process_cancellation(wav, tmp_p
"Waiting for worker", progress)
return {"bins": 2048}
monkeypatch.setattr(gui_jobs, "waveform", slow_scan)
- jobs = gui_jobs.Jobs(tmp_path, lambda value: rooted(tmp_path, value))
- snapshot = {"path": source.name, "input_id": gui_jobs.identity(source)}
+ jobs = gui_jobs.Jobs()
+ snapshot = {"upload_id": jobs.register_upload(source, source.name), "input_id": gui_jobs.identity(source)}
try:
first = jobs.submit("scan", snapshot)
assert started.wait(5)
# Public snapshots cannot mutate the immutable in-flight job.
- first["snapshot"]["path"] = "missing.wav"
+ first["snapshot"]["name"] = "renamed.wav"
queued = [jobs.submit("scan", snapshot) for _ in range(3)]
with pytest.raises(gui_jobs.Busy):
jobs.submit("scan", snapshot)
@@ -71,8 +69,8 @@ def test_history_and_result_cache_are_bounded_and_evicted_media_expires(wav, tmp
(work / "processed.wav").write_bytes(b"preview")
return {}
monkeypatch.setattr(gui_jobs, "render_preview", preview)
- jobs = gui_jobs.Jobs(tmp_path, lambda value: rooted(tmp_path, value))
- snapshot = {"path": source.name, "input_id": gui_jobs.identity(source)}
+ jobs = gui_jobs.Jobs()
+ snapshot = {"upload_id": jobs.register_upload(source, source.name), "input_id": gui_jobs.identity(source)}
try:
results = []
for index in range(10):
@@ -82,7 +80,7 @@ def test_history_and_result_cache_are_bounded_and_evicted_media_expires(wav, tmp
results.append(result["result"]["result_id"])
assert len(jobs.list()) == 5
assert len(jobs.cache) == 2
- assert len(list(jobs.work.iterdir())) == 2
+ assert len([item for item in jobs.work.iterdir() if item.name != "uploads"]) == 2
with pytest.raises(KeyError):
jobs.media(results[0], "original")
assert jobs.media(results[-1], "original") == b"preview"
@@ -93,7 +91,7 @@ def test_history_and_result_cache_are_bounded_and_evicted_media_expires(wav, tmp
with jobs.condition:
jobs._evict()
assert not jobs.cache
- assert not list(jobs.work.iterdir())
+ assert not [item for item in jobs.work.iterdir() if item.name != "uploads"]
finally:
jobs.close()
@@ -107,8 +105,8 @@ def test_queued_input_replacement_is_rejected_and_shutdown_cancels_scan(wav, tmp
progress("Scanning", 1, 10)
time.sleep(0.01)
monkeypatch.setattr(gui_jobs, "waveform", scan)
- jobs = gui_jobs.Jobs(tmp_path, lambda value: rooted(tmp_path, value))
- snapshot = {"path": source.name, "input_id": gui_jobs.identity(source)}
+ jobs = gui_jobs.Jobs()
+ snapshot = {"upload_id": jobs.register_upload(source, source.name), "input_id": gui_jobs.identity(source)}
running = jobs.submit("scan", snapshot)
assert started.wait(5)
queued = jobs.submit("scan", snapshot)
diff --git a/lib/project/tests/test_gui_launcher.py b/lib/project/tests/test_gui_launcher.py
index 8111c67..758d034 100644
--- a/lib/project/tests/test_gui_launcher.py
+++ b/lib/project/tests/test_gui_launcher.py
@@ -18,10 +18,10 @@ def test_cli_gui_dispatch_preserves_configuration_flags(tmp_path, monkeypatch):
config.write_text('profile = "narrator"\n')
received = []
monkeypatch.setattr(gui, "serve", lambda *args, **kwargs: received.append((args, kwargs)))
- monkeypatch.setattr(sys, "argv", ["producer", "gui", "--root", str(tmp_path), "--port", "9001",
+ monkeypatch.setattr(sys, "argv", ["producer", "gui", "--port", "9001",
"--config", str(config), "--profile", "radio", "--target-lufs", "-21", "--set", "target_lufs=-18"])
cli.main()
- assert received == [((tmp_path, 9001), {"config": config, "profile": "radio", "overrides": {"target_lufs": -18}})]
+ assert received == [((9001,), {"config": config, "profile": "radio", "overrides": {"target_lufs": -18}})]
def test_launcher_only_gui_installs_extra_and_preserves_config_injection(tmp_path):
@@ -58,7 +58,7 @@ def test_launcher_only_gui_installs_extra_and_preserves_config_injection(tmp_pat
calls = [json.loads(line) for line in log.read_text().splitlines()]
installs = [call for call in calls if call[0] == "pip"]
assert len(installs) == 1 and installs[0][-1].endswith("[bundled-ffmpeg]")
- args = launch("gui", "--root", str(checkout), "--port", "8766")
+ args = launch("gui", "--port", "8766")
assert args[3:5] == ["gui", f"--config={checkout / 'config.toml'}"]
calls = [json.loads(line) for line in log.read_text().splitlines()]
installs = [call for call in calls if call[0] == "pip"]
@@ -86,7 +86,7 @@ assert parser().parse_args(['config']).command == 'config'
assert result.returncode == 0, result.stderr
-def test_server_binds_only_loopback_and_prints_fragment_token(tmp_path, monkeypatch, capsys):
+def test_server_binds_only_loopback_and_prints_plain_url(tmp_path, monkeypatch, capsys):
pytest.importorskip("fastapi")
uvicorn = pytest.importorskip("uvicorn")
import socket
@@ -103,11 +103,11 @@ def test_server_binds_only_loopback_and_prints_fragment_token(tmp_path, monkeypa
def run(self, sockets):
seen.append(sockets[0].getsockname())
monkeypatch.setattr(uvicorn, "Server", Server)
- gui.serve(tmp_path, port)
+ gui.serve(port)
assert seen == [("127.0.0.1", port)]
output = capsys.readouterr().out
- assert f"http://127.0.0.1:{port}/#token=" in output
- assert len(output.split("#token=")[1].splitlines()[0]) >= 43
+ assert f"http://127.0.0.1:{port}/\n" in output
+ assert "#token" not in output
for invalid in (0, -1, 65536):
with pytest.raises(ValueError):
- gui.serve(tmp_path, invalid)
+ gui.serve(invalid)
diff --git a/lib/project/tests/test_pipeline.py b/lib/project/tests/test_pipeline.py
index fe04519..0bc2cef 100644
--- a/lib/project/tests/test_pipeline.py
+++ b/lib/project/tests/test_pipeline.py
@@ -64,13 +64,13 @@ def test_existing_report_log_is_replaced_without_overwrite(wav, tmp_path, bypass
@pytest.mark.parametrize("kind,message", [
("empty", "empty"), ("silence", "silent or too quiet"),
("quiet", "silent or too quiet"), ("surround", "Only mono or stereo"),
- ("nonfinite", "non-finite"), ("stereo", "downmix to mono"),
+ ("nonfinite", "non-finite"),
])
def test_unusable_inputs_leave_no_outputs(wav, tmp_path, bypass, progress, ffmpeg, kind, message):
samples = {
"empty": np.empty(0), "silence": np.zeros(48000),
"quiet": np.full(48000, 1e-6), "surround": np.ones((4800, 3)) * 0.1,
- "nonfinite": np.array([0.1, np.nan]), "stereo": np.ones((4800, 2)) * 0.1,
+ "nonfinite": np.array([0.1, np.nan]),
}[kind]
source = wav(samples)
target = tmp_path / "outputs" / "master.wav"
@@ -79,6 +79,51 @@ def test_unusable_inputs_leave_no_outputs(wav, tmp_path, bypass, progress, ffmpe
assert not target.parent.exists()
+def test_downmix_to_mono_strategies(wav, tmp_path, progress):
+ from producer.audio import downmix_to_mono
+ time = np.arange(48000 * 3) / 48000
+ voice = 0.1 * np.sin(2 * np.pi * 440 * time)
+ # Identical channels: a plain average is lossless and keeps full level.
+ mono = tmp_path / "same.wav"
+ method = downmix_to_mono(wav(np.column_stack([voice, voice])), mono, progress)
+ assert method == "the channels were averaged"
+ assert sf.info(mono).channels == 1
+ assert np.abs(sf.read(mono)[0] - voice).max() == pytest.approx(0, abs=1e-7)
+ # Inverted channels: the phase-cancel case the old error guarded against.
+ mono = tmp_path / "inverted.wav"
+ method = downmix_to_mono(wav(np.column_stack([voice, -voice])), mono, progress)
+ assert "inverted" in method
+ assert np.abs(sf.read(mono)[0] - voice).max() == pytest.approx(0, abs=1e-7)
+ # One silent channel: keep the live channel at unity gain, not -6 dB.
+ mono = tmp_path / "silent.wav"
+ method = downmix_to_mono(wav(np.column_stack([voice, np.zeros_like(voice)])), mono, progress)
+ assert "right channel is silent" in method
+ assert np.abs(sf.read(mono)[0] - voice).max() == pytest.approx(0, abs=1e-7)
+ method = downmix_to_mono(wav(np.column_stack([np.zeros_like(voice), voice])),
+ tmp_path / "silent-left.wav", progress)
+ assert "left channel is silent" in method
+ with pytest.raises(ValueError, match="non-finite"):
+ downmix_to_mono(wav(np.array([[0.1, np.nan]])), tmp_path / "bad.wav", progress)
+
+
+def test_stereo_input_is_downmixed_not_rejected(wav, tmp_path, bypass, progress, ffmpeg):
+ time = np.arange(48000 * 3) / 48000
+ voice = 0.1 * np.sin(2 * np.pi * 440 * time)
+ source = wav(np.column_stack([voice, -voice]))
+ original = source.read_bytes()
+ target = tmp_path / "master.wav"
+ report = pipeline.process(source, target, bypass, progress)
+ assert any("downmixed to mono" in warning and "inverted" in warning
+ for warning in report["warnings"])
+ stats = sf.info(target)
+ assert stats.channels == 1 and stats.frames == 144000
+ # A naive (L + R) / 2 mix of these channels is digital silence; the master
+ # must carry real signal instead.
+ assert report["master"]["sample_peak_dbfs"] > -30
+ assert source.read_bytes() == original
+ assert not list(target.parent.glob(".producer-*"))
+
+
def test_measure_reports_channels_peaks_clipping_and_progress(wav):
samples = np.tile([1.0, -0.25], (4800, 1))
events = []
generated by cgit v1.2.3 (git 2.25.1) at 2026-10-24 12:45:23 +0000