diff options
Diffstat (limited to 'lib/project/tests/test_gui.py')
| -rw-r--r-- | lib/project/tests/test_gui.py | 162 |
1 files changed, 92 insertions, 70 deletions
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): |