| from __future__ import annotations |
|
|
| import importlib |
| import os |
| import sys |
| import tempfile |
| import unittest |
| from pathlib import Path |
| from unittest.mock import patch |
|
|
| from fastapi.testclient import TestClient |
| import numpy as np |
|
|
|
|
| BACKEND_DIR = Path(__file__).resolve().parents[1] |
| if str(BACKEND_DIR) not in sys.path: |
| sys.path.insert(0, str(BACKEND_DIR)) |
|
|
|
|
| class AudioStudioApiTests(unittest.TestCase): |
| def _load_client(self, runtime_mode: str) -> TestClient: |
| self.temp_dir = tempfile.TemporaryDirectory() |
| os.environ["OMNIVOICE_RUNTIME_MODE"] = runtime_mode |
| os.environ["OMNIVOICE_DATA_ROOT"] = self.temp_dir.name |
|
|
| sys.modules.pop("main", None) |
| main = importlib.import_module("main") |
| main = importlib.reload(main) |
| self.addCleanup(self.temp_dir.cleanup) |
| return TestClient(main.app) |
|
|
| def tearDown(self) -> None: |
| os.environ.pop("OMNIVOICE_RUNTIME_MODE", None) |
| os.environ.pop("OMNIVOICE_DATA_ROOT", None) |
| os.environ.pop("OMNIVOICE_ENABLED_FEATURES", None) |
|
|
| def test_audio_projects_require_capability_in_production(self) -> None: |
| client = self._load_client(runtime_mode="production") |
|
|
| response = client.post("/api/audio/projects", json={"name": "Forbidden"}) |
|
|
| self.assertEqual(response.status_code, 403) |
|
|
| def test_audio_project_create_and_list_in_self_hosted(self) -> None: |
| client = self._load_client(runtime_mode="self_hosted") |
|
|
| create_response = client.post("/api/audio/projects", json={"name": "Narration Draft"}) |
|
|
| self.assertEqual(create_response.status_code, 200) |
| created = create_response.json() |
| self.assertEqual(created["name"], "Narration Draft") |
| self.assertGreaterEqual(len(created["lines"]), 1) |
| self.assertIn("speaker_profiles", created) |
|
|
| list_response = client.get("/api/audio/projects") |
|
|
| self.assertEqual(list_response.status_code, 200) |
| listed = list_response.json() |
| self.assertEqual(len(listed), 1) |
| self.assertEqual(listed[0]["id"], created["id"]) |
| self.assertEqual(listed[0]["name"], "Narration Draft") |
|
|
| def test_audio_render_and_mixdown_create_files(self) -> None: |
| client = self._load_client(runtime_mode="self_hosted") |
|
|
| create_response = client.post("/api/audio/projects", json={"name": "Render Pipeline"}) |
| project = create_response.json() |
|
|
| class FakeAudioGenerator: |
| runtime_backend = "mock" |
| runtime_device = "cpu" |
|
|
| def generate(self, **_: object) -> list[np.ndarray]: |
| return [np.zeros(24000, dtype=np.float32)] |
|
|
| with patch("audio_projects.get_audio_generator", return_value=FakeAudioGenerator()): |
| for line in project["lines"]: |
| render_response = client.post( |
| f"/api/audio/projects/{project['id']}/lines/{line['id']}/render", |
| json={}, |
| ) |
| self.assertEqual(render_response.status_code, 200) |
| rendered = render_response.json() |
| self.assertEqual(rendered["render_filename"], f"{line['id']}.wav") |
| self.assertGreater(rendered["render_duration_seconds"], 0) |
| self.assertEqual(rendered["render_backend"], "mock") |
| self.assertEqual(rendered["render_device"], "cpu") |
|
|
| mixdown_response = client.post( |
| f"/api/audio/projects/{project['id']}/mixdown", |
| json={"format": "wav"}, |
| ) |
| self.assertEqual(mixdown_response.status_code, 200) |
| mixdown = mixdown_response.json() |
| self.assertEqual(mixdown["format"], "wav") |
|
|
| file_response = client.get( |
| f"/api/audio/projects/{project['id']}/files/mixdowns/{mixdown['filename']}" |
| ) |
| self.assertEqual(file_response.status_code, 200) |
|
|
