Spaces:
Sleeping
Sleeping
File size: 3,357 Bytes
fa35534 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | #!/usr/bin/env python3
"""Smoke-test card selection semantics, selected-only export, draw-next, and immediate clip edits."""
from __future__ import annotations
import io
import json
import sys
import time
import zipfile
from pathlib import Path
from urllib.parse import quote
import soundfile as sf
from fastapi.testclient import TestClient
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from app import app # noqa: E402
from synth_generator import generate_test_song # noqa: E402
def wait_for_job(client: TestClient, job_id: str) -> dict:
for _ in range(120):
payload = client.get(f"/api/jobs/{job_id}").json()
if payload["status"] in {"complete", "error"}:
return payload
time.sleep(0.1)
raise TimeoutError(job_id)
def main() -> int:
song = generate_test_song(pattern_name="funk", bars=1, bpm=124, add_bass=False)
buf = io.BytesIO()
sf.write(buf, song.drums_only, song.sr, format="WAV")
buf.seek(0)
client = TestClient(app)
response = client.post(
"/api/jobs",
files={"file": ("cards.wav", buf, "audio/wav")},
data={"params": json.dumps({"stem": "all", "clustering_mode": "online_preview", "target_min": 3, "target_max": 10})},
)
response.raise_for_status()
job_id = response.json()["id"]
job = wait_for_job(client, job_id)
assert job["status"] == "complete", job.get("error")
samples = job["result"]["samples"]
assert samples, "expected at least one sample"
labels = [sample["label"] for sample in samples[: min(2, len(samples))]]
selected = client.post(f"/api/jobs/{job_id}/export-selected", json={"labels": labels})
selected.raise_for_status()
selected_payload = selected.json()["export"]
assert selected_payload["kind"] == "selected-sample-export"
assert selected_payload["selected_labels"] == sorted(labels)
archive_response = client.get(selected_payload["file_urls"]["archive"])
archive_response.raise_for_status()
with zipfile.ZipFile(io.BytesIO(archive_response.content)) as zf:
names = zf.namelist()
assert any(name.endswith(".wav") for name in names), names
label = labels[0]
draw = client.post(f"/api/jobs/{job_id}/samples/{quote(label, safe='')}/draw", json={})
draw.raise_for_status()
drawn = draw.json()["sample"]
assert drawn["label"] == label
assert drawn["url"]
drawn_audio = client.get(drawn["url"])
drawn_audio.raise_for_status()
assert drawn_audio.content[:4] == b"RIFF"
edit = client.post(f"/api/jobs/{job_id}/samples/{quote(label, safe='')}/edit", json={"start_offset_ms": 5, "tail_offset_ms": 30})
edit.raise_for_status()
edited = edit.json()["sample"]
assert edited["label"] == label
assert "overrides/hits" in edited["file"], edited
edited_audio = client.get(edited["url"])
edited_audio.raise_for_status()
assert edited_audio.content[:4] == b"RIFF"
print(json.dumps({
"status": "ok",
"job_id": job_id,
"selected_labels": labels,
"drawn_representative_hit_index": drawn["representative_hit_index"],
"edited_file": edited["file"],
"archive": selected_payload["files"]["archive"],
}, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
|