Spaces:
Running
Running
File size: 3,585 Bytes
0f24c2f 4369b06 0f24c2f 3b1b3cb 4369b06 0f24c2f 4369b06 3b1b3cb 0f24c2f 3b1b3cb 0f24c2f 4369b06 0f24c2f 4369b06 0f24c2f 4369b06 0f24c2f 3b1b3cb 0f24c2f 4369b06 0f24c2f 4369b06 0f24c2f 3b1b3cb 0f24c2f 4369b06 0f24c2f 4369b06 0f24c2f 4369b06 0f24c2f | 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 97 98 99 100 101 102 | import asyncio
from pathlib import Path
import pytest
from models import JobState
pytestmark = pytest.mark.asyncio
async def test_run_demucs_updates_state_on_success(tmp_path, monkeypatch):
"""Worker updates job from pending -> processing -> done (Demucs mocked)."""
from processor import _jobs, _run_demucs, create_job
job_id = "test-success-job"
input_path = tmp_path / "input.mp3"
output_dir = tmp_path / "outputs" / job_id
output_dir.mkdir(parents=True, exist_ok=True)
input_path.write_bytes(b"\xff\xfb" + b"\x00" * 100)
(output_dir / "original.mp3").write_bytes(b"\xff\xfb" + b"\x00" * 100)
class FakeProc:
def __init__(self, args):
self.args = args
self.returncode = 0
async def communicate(self):
if self.args[0] == "ffmpeg":
Path(self.args[-1]).parent.mkdir(parents=True, exist_ok=True)
Path(self.args[-1]).write_bytes(b"\xff\xfb" + b"\x00" * 50)
return b"", b""
# Simulate Demucs writing stems into the expected folder layout.
out_idx = self.args.index("-o")
demucs_work_dir = Path(self.args[out_idx + 1])
input_stem = Path(self.args[-1]).stem
base = demucs_work_dir / "htdemucs" / input_stem
base.mkdir(parents=True, exist_ok=True)
for stem_name in ("vocals", "drums", "bass", "other"):
(base / f"{stem_name}.mp3").write_bytes(b"\xff\xfb" + b"\x00" * 40)
return b"", b""
async def fake_create_subprocess(*args, **kwargs):
return FakeProc(args)
monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create_subprocess)
monkeypatch.setattr(
"processor.analyze_audio_file",
lambda _: {
"bpm": 128.4,
"bpm_confidence": 0.91,
"musical_key": "A",
"key_scale": "minor",
"key_confidence": 0.88,
},
)
create_job(job_id)
await _run_demucs(job_id, str(input_path), str(output_dir))
assert _jobs[job_id].state == JobState.done
assert _jobs[job_id].progress == 100
assert _jobs[job_id].stems is not None
assert set(_jobs[job_id].stems.keys()) == {"instrumental", "vocals", "drums", "bass", "other"}
for stem in _jobs[job_id].stems.values():
assert (output_dir / stem).exists()
assert _jobs[job_id].bpm == pytest.approx(128.4)
assert _jobs[job_id].bpm_confidence == pytest.approx(0.91)
assert _jobs[job_id].musical_key == "A"
assert _jobs[job_id].key_scale == "minor"
assert _jobs[job_id].key_confidence == pytest.approx(0.88)
async def test_run_demucs_marks_error_on_failure(tmp_path, monkeypatch):
"""Worker marks job as error when Demucs subprocess fails."""
from processor import _jobs, _run_demucs, create_job
job_id = "test-fail-job"
input_path = tmp_path / "input.mp3"
output_dir = tmp_path / "outputs" / job_id
input_path.write_bytes(b"\xff\xfb" + b"\x00" * 100)
class FakeFailProc:
returncode = 1
async def communicate(self):
return b"", b"Demucs error: model not found"
async def fake_create_subprocess(*args, **kwargs):
return FakeFailProc()
monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create_subprocess)
create_job(job_id)
with pytest.raises(RuntimeError, match="Demucs failed"):
await _run_demucs(job_id, str(input_path), str(output_dir))
assert _jobs[job_id].state == JobState.error
assert _jobs[job_id].error is not None
|