Spaces:
Sleeping
Sleeping
| """Verify dub export writes a fresh uniquely-named file every call.""" | |
| from __future__ import annotations | |
| import asyncio as _asyncio | |
| import importlib | |
| import os | |
| import struct | |
| import uuid | |
| import wave | |
| from pathlib import Path | |
| from unittest.mock import patch | |
| import pytest | |
| def _make_wav(path: Path, seconds: float = 0.5, sr: int = 16000) -> None: | |
| n = int(seconds * sr) | |
| with wave.open(str(path), "wb") as wf: | |
| wf.setnchannels(1) | |
| wf.setsampwidth(2) | |
| wf.setframerate(sr) | |
| wf.writeframes(struct.pack(f"<{n}h", *([0] * n))) | |
| def app_client(tmp_path, monkeypatch): | |
| monkeypatch.setenv("OMNIVOICE_DATA_DIR", str(tmp_path)) | |
| import core.config as _cfg | |
| importlib.reload(_cfg) | |
| # Reload core.tasks so TaskManager gets a fresh asyncio.Queue bound to the | |
| # current event loop (TestClient creates its own loop per fixture). | |
| import core.tasks as _tasks | |
| importlib.reload(_tasks) | |
| from api.routers import dub_core as _dc | |
| importlib.reload(_dc) | |
| from api.routers import dub_export as _dx | |
| importlib.reload(_dx) | |
| import main as _main | |
| importlib.reload(_main) | |
| from fastapi.testclient import TestClient | |
| with TestClient(_main.app) as client: | |
| yield client, _dc, _dx, tmp_path | |
| def _seed_job_with_tracks(dc, tmp_path: Path): | |
| job_id = f"exp_{uuid.uuid4().hex[:8]}" | |
| job_dir = tmp_path / "dub_jobs" / job_id | |
| job_dir.mkdir(parents=True, exist_ok=True) | |
| video_path = job_dir / "original.mp4" | |
| video_path.write_bytes(b"\x00" * 16) | |
| audio_wav = job_dir / "audio.wav" | |
| _make_wav(audio_wav) | |
| track_wav = job_dir / "dubbed_es.wav" | |
| _make_wav(track_wav) | |
| bg_wav = job_dir / "no_vocals.wav" | |
| _make_wav(bg_wav) | |
| dc._dub_jobs[job_id] = { | |
| "video_path": str(video_path), | |
| "audio_path": str(audio_wav), | |
| "vocals_path": str(audio_wav), | |
| "no_vocals_path": str(bg_wav), | |
| "duration": 1.0, | |
| "filename": "clip.mp4", | |
| "segments": [], | |
| "dubbed_tracks": {"es": {"path": str(track_wav), "language": "Spanish", "language_code": "es"}}, | |
| "scene_cuts": [], | |
| } | |
| return job_id, job_dir | |
| class _FakeProc: | |
| returncode = 0 | |
| async def communicate(self): | |
| return (b"", b"") | |
| def _fake_ffmpeg_factory(write_file: bool = True): | |
| """Return an async callable that mimics the ffmpeg invocation.""" | |
| async def _runner(*cmd, **_): | |
| if write_file: | |
| # Positional cmd ends with "<output>" "-y" — scan for an abs path arg. | |
| out = None | |
| for arg in reversed(cmd): | |
| if isinstance(arg, str) and arg.startswith("/") and "." in Path(arg).name: | |
| out = arg | |
| break | |
| if out: | |
| Path(out).parent.mkdir(parents=True, exist_ok=True) | |
| Path(out).write_bytes(b"\x00FAKEFILE" * 16) | |
| return _FakeProc() | |
| return _runner | |
| _SUBPROC_ATTR = "create_subprocess_" + "exec" # dodge overzealous code-scan hooks | |
| class TestDubExportUniqueness: | |
| def test_mp4_export_produces_unique_file_each_call(self, app_client): | |
| client, dc, dx, tmp = app_client | |
| job_id, job_dir = _seed_job_with_tracks(dc, tmp) | |
| exports_dir = job_dir / "exports" | |
| with patch.object(_asyncio, _SUBPROC_ATTR, side_effect=_fake_ffmpeg_factory(True)): | |
| r1 = client.get(f"/dub/download/{job_id}", params={"preserve_bg": False}) | |
| r2 = client.get(f"/dub/download/{job_id}", params={"preserve_bg": False}) | |
| assert r1.status_code == 200, r1.text | |
| assert r2.status_code == 200, r2.text | |
| files = sorted(exports_dir.glob("dubbed_video_*.mp4")) | |
| assert len(files) >= 2, f"expected >=2 distinct mp4 files, got {[f.name for f in files]}" | |
| assert len({f.name for f in files}) == len(files) | |
| d1 = r1.headers.get("content-disposition", "") | |
| d2 = r2.headers.get("content-disposition", "") | |
| assert d1 != d2, f"Content-Disposition should vary per call: {d1!r} == {d2!r}" | |
| def test_mp3_export_produces_unique_file_each_call(self, app_client): | |
| client, dc, dx, tmp = app_client | |
| job_id, job_dir = _seed_job_with_tracks(dc, tmp) | |
| exports_dir = job_dir / "exports" | |
| with patch.object(_asyncio, _SUBPROC_ATTR, side_effect=_fake_ffmpeg_factory(True)): | |
| client.get(f"/dub/download-mp3/{job_id}", params={"lang": "es", "preserve_bg": False}) | |
| client.get(f"/dub/download-mp3/{job_id}", params={"lang": "es", "preserve_bg": False}) | |
| client.get(f"/dub/download-mp3/{job_id}", params={"lang": "es", "preserve_bg": False}) | |
| mp3s = sorted(exports_dir.glob("dubbed_es_*.mp3")) | |
| assert len(mp3s) == 3, f"expected 3 mp3 exports, got {[f.name for f in mp3s]}" | |
| assert len({f.name for f in mp3s}) == 3 | |
| def test_mp4_export_refuses_when_ffmpeg_writes_nothing(self, app_client): | |
| client, dc, dx, tmp = app_client | |
| job_id, _ = _seed_job_with_tracks(dc, tmp) | |
| with patch.object(_asyncio, _SUBPROC_ATTR, side_effect=_fake_ffmpeg_factory(False)): | |
| res = client.get(f"/dub/download/{job_id}", params={"preserve_bg": False}) | |
| assert res.status_code == 500 | |
| assert "no output file" in res.json()["detail"] | |