pyr0ball
a6f60c9e07
Full FastAPI backend for the AI music continuation editor:
Services
- chain.py: chain + node CRUD, commit/discard, recursive CTE spine query
- musicgen.py: MusicGenClient with cf-orch allocation + mock mode (CF_MUSICGEN_MOCK=1)
- stems.py: Demucs 4-stem separation subprocess wrapper + mock mode
- export.py: ffmpeg concat demuxer to stitch committed spine into WAV/MP3
API endpoints
- chains: CRUD, multipart audio upload (WAV/MP3/FLAC/OGG/M4A/AIFF)
- nodes: branch creation (202 + BackgroundTasks), commit, discard, audio stream
- gpu: cf-orch capacity status; session allocation stubbed pending cf-orch#43
- stems: Paid-tier stem separation (Demucs, gated via tiers.py)
- export: POST /{chain_id}/export → FileResponse download
- events: SSE stream (node-status events) per chain via asyncio Queue pub/sub
Infrastructure
- lifespan: reads SPARROW_DB_PATH/DATA_DIR at startup (not import time)
- events_store: subscribe/unsubscribe/broadcast pattern for SSE
- CORS: open in dev, SPARROW_CORS_ORIGINS in production
- Background generation opens its own DB connection (WAL-safe)
Tests: 30/30 passing across service units and API integration
102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
# tests/test_api_nodes.py — integration tests for branch/commit/discard endpoints
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
|
|
|
|
def _create_chain_with_root(client, tmp_path=None):
|
|
"""Helper: create a chain and upload a root node."""
|
|
resp = client.post("/api/chains/", json={"name": "test chain"})
|
|
chain_id = resp.json()["id"]
|
|
|
|
resp = client.post(
|
|
f"/api/chains/{chain_id}/upload",
|
|
files={"file": ("root.wav", io.BytesIO(b"RIFF" + b"\x00" * 40), "audio/wav")},
|
|
)
|
|
node = resp.json()
|
|
return chain_id, node["id"]
|
|
|
|
|
|
def test_create_branch_returns_202(client, tmp_path):
|
|
chain_id, root_id = _create_chain_with_root(client)
|
|
|
|
resp = client.post(
|
|
f"/api/nodes/{root_id}/branch",
|
|
json={"prompt": "upbeat jazz", "duration_s": 10.0},
|
|
)
|
|
assert resp.status_code == 202
|
|
node = resp.json()
|
|
assert node["status"] == "pending"
|
|
assert node["parent_id"] == root_id
|
|
assert node["prompt"] == "upbeat jazz"
|
|
|
|
|
|
def test_branch_parent_not_found(client):
|
|
resp = client.post(
|
|
"/api/nodes/nonexistent/branch",
|
|
json={"prompt": "test"},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_commit_requires_ready_status(client, conn):
|
|
"""
|
|
Directly insert a pending node into the shared DB (bypasses BackgroundTasks
|
|
so we control the status without triggering mock generation).
|
|
"""
|
|
from app.services import chain as svc
|
|
|
|
chain_id, root_id = _create_chain_with_root(client)
|
|
# Create branch via service (no background task), stays "pending"
|
|
branch = svc.create_branch_node(
|
|
conn, root_id, chain_id, "test", None, None, None, 3.0, 10.0
|
|
)
|
|
resp = client.post(f"/api/nodes/{branch['id']}/commit")
|
|
assert resp.status_code == 409
|
|
|
|
|
|
def test_delete_root_node_refused(client):
|
|
chain_id, root_id = _create_chain_with_root(client)
|
|
resp = client.delete(f"/api/nodes/{root_id}")
|
|
assert resp.status_code == 409
|
|
|
|
|
|
def test_delete_node_not_found(client):
|
|
resp = client.delete("/api/nodes/doesnotexist")
|
|
assert resp.status_code == 409
|
|
|
|
|
|
def test_audio_not_found_for_pending_node(client, conn):
|
|
"""Pending node (no audio_path yet) returns 404 on the audio endpoint."""
|
|
from app.services import chain as svc
|
|
|
|
chain_id, root_id = _create_chain_with_root(client)
|
|
branch = svc.create_branch_node(
|
|
conn, root_id, chain_id, "test", None, None, None, 3.0, 10.0
|
|
)
|
|
resp = client.get(f"/api/nodes/{branch['id']}/audio")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_mock_generation_completes(client, conn):
|
|
"""
|
|
With CF_MUSICGEN_MOCK=1, TestClient runs BackgroundTasks synchronously,
|
|
so the node transitions to 'ready' before client.post() returns.
|
|
We verify the completed state via the shared conn fixture.
|
|
"""
|
|
from app.services import chain as svc
|
|
|
|
chain_id, root_id = _create_chain_with_root(client)
|
|
resp = client.post(
|
|
f"/api/nodes/{root_id}/branch",
|
|
json={"prompt": "test mock", "duration_s": 5.0},
|
|
)
|
|
assert resp.status_code == 202
|
|
branch_id = resp.json()["id"]
|
|
|
|
# Background task completed synchronously — node is ready
|
|
node = svc.get_node(conn, branch_id)
|
|
assert node is not None
|
|
assert node["status"] == "ready", (
|
|
f"Expected ready, got: {node['status']}, error: {node.get('error_msg')}"
|
|
)
|