gahusb
755dea63f4
1. video.py _container_to_nas, orchestrator.py _local_path에서 path 변환 전 ?쿼리 strip — 이전 commit 20c5268의 cache-buster ?v=...가 Windows path로 그대로 전달되어 input_validation 실패하던 문제 픽스 2. cover.py _generate_with_dalle가 background_keyword를 prompt에 포함 — 사용자가 PipelineStartModal에서 '배경 키워드' 입력 시 처음부터 원하는 분위기 cover 생성
84 lines
3.4 KiB
Python
84 lines
3.4 KiB
Python
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
from app.pipeline.orchestrator import _resolve_input
|
|
|
|
|
|
def test_resolve_input_track():
|
|
pipeline = {"id": 1, "track_id": 13, "compile_job_id": None}
|
|
track = {
|
|
"id": 13, "title": "Lo-Fi Drive", "genre": "lo-fi",
|
|
"moods": ["chill"], "duration_sec": 176,
|
|
"file_path": "/app/data/x.mp3", "audio_url": "/media/music/x.mp3",
|
|
}
|
|
with patch("app.pipeline.orchestrator.db.get_track_by_id", return_value=track):
|
|
result = _resolve_input(pipeline)
|
|
assert result["audio_path"] == "/app/data/x.mp3"
|
|
assert result["duration_sec"] == 176
|
|
assert len(result["tracks"]) == 1
|
|
assert result["tracks"][0]["start_offset_sec"] == 0
|
|
assert result["title"] == "Lo-Fi Drive"
|
|
assert result["genre"] == "lo-fi"
|
|
|
|
|
|
def test_resolve_input_compile_job():
|
|
pipeline = {"id": 2, "track_id": None, "compile_job_id": 5}
|
|
job = {
|
|
"id": 5, "status": "succeeded", "title": "Chill Mix",
|
|
"audio_path": "/app/data/compiles/5.mp3",
|
|
"track_ids": [13, 14, 15],
|
|
"crossfade_sec": 3,
|
|
}
|
|
tracks = {
|
|
13: {"id": 13, "title": "T1", "duration_sec": 180},
|
|
14: {"id": 14, "title": "T2", "duration_sec": 200},
|
|
15: {"id": 15, "title": "T3", "duration_sec": 150},
|
|
}
|
|
with patch("app.pipeline.orchestrator.db.get_compile_job", return_value=job), \
|
|
patch("app.pipeline.orchestrator.db.get_track_by_id", side_effect=lambda i: tracks[i]):
|
|
result = _resolve_input(pipeline)
|
|
assert result["audio_path"] == "/app/data/compiles/5.mp3"
|
|
# 누적 = 180+200+150 - 2*3(crossfade pair gaps) = 524
|
|
assert result["duration_sec"] == 524
|
|
assert len(result["tracks"]) == 3
|
|
assert result["tracks"][0]["start_offset_sec"] == 0
|
|
assert result["tracks"][1]["start_offset_sec"] == 177 # 180 - 3
|
|
assert result["tracks"][2]["start_offset_sec"] == 374 # 177 + 200 - 3
|
|
assert result["title"] == "Chill Mix"
|
|
assert result["genre"] == "mix"
|
|
|
|
|
|
def test_resolve_input_compile_not_ready():
|
|
pipeline = {"id": 3, "track_id": None, "compile_job_id": 6}
|
|
job = {"id": 6, "status": "rendering"}
|
|
with patch("app.pipeline.orchestrator.db.get_compile_job", return_value=job):
|
|
with pytest.raises(ValueError, match="not ready"):
|
|
_resolve_input(pipeline)
|
|
|
|
|
|
def test_resolve_input_neither():
|
|
pipeline = {"id": 4, "track_id": None, "compile_job_id": None}
|
|
with pytest.raises(ValueError):
|
|
_resolve_input(pipeline)
|
|
|
|
|
|
def test_resolve_input_compile_job_done_status():
|
|
"""compile job status='done'도 accept (production convention)."""
|
|
pipeline = {"id": 5, "track_id": None, "compile_job_id": 7}
|
|
job = {
|
|
"id": 7, "status": "done", "title": "Done Mix",
|
|
"audio_path": "/app/data/compiles/7.mp3",
|
|
"track_ids": [1], "crossfade_sec": 0,
|
|
}
|
|
track = {"id": 1, "title": "T1", "duration_sec": 100}
|
|
with patch("app.pipeline.orchestrator.db.get_compile_job", return_value=job), \
|
|
patch("app.pipeline.orchestrator.db.get_track_by_id", return_value=track):
|
|
result = _resolve_input(pipeline)
|
|
assert result["audio_path"] == "/app/data/compiles/7.mp3"
|
|
assert result["title"] == "Done Mix"
|
|
|
|
|
|
def test_local_path_strips_cache_buster():
|
|
from app.pipeline.orchestrator import _local_path
|
|
# /media/videos/3/cover.jpg?v=... → /app/data/videos/3/cover.jpg
|
|
assert _local_path("/media/videos/3/cover.jpg?v=20260510065642") == "/app/data/videos/3/cover.jpg"
|