"""Tests for audio preprocessing / enhancement module (audio_enhancer.py). Validates: 1. Disabled enhancer -> zero overhead pass-through of raw bytes. 2. Successful enhancement -> receives enhanced audio, saves with _enhanced suffix, emits audio.enhance.result with latency_ms/path/size. 3. Failure/timeout -> graceful fallback to raw audio bytes, ok=False event logged. 4. Zero transcoding by default (raw bytes sent without ffmpeg). 5. Timeout configuration adheres to 60.0s. """ import asyncio import os import sys from pathlib import Path from unittest.mock import AsyncMock, patch import httpx sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from app.audio_enhancer import enhance_voice from app.config import Settings from app import observability as obs _passed = _failed = 0 def check(name: str, ok: bool, detail: str = "") -> None: global _passed, _failed ok = bool(ok) mark = "PASS" if ok else "FAIL" print(f"[{mark}] {name}" + (f" — {detail}" if detail and not ok else "")) _passed += ok _failed += (not ok) async def test_disabled_enhancer(): cfg = Settings(AUDIO_ENHANCE_ENABLED=False, AUDIO_ENHANCE_BASE_URL="http://mock/enhance") raw_audio = b"fake-raw-voice-bytes" out_audio, meta = await enhance_voice(cfg, raw_audio, mime_type="audio/ogg") check("disabled: raw bytes returned directly", out_audio == raw_audio) check("disabled: metadata indicates skipped", meta.get("skipped") is True) async def test_timeout_setting(): cfg = Settings(AUDIO_ENHANCE_TIMEOUT_SECONDS=60.0) check("config: timeout defaults to 60.0s", cfg.AUDIO_ENHANCE_TIMEOUT_SECONDS == 60.0) async def test_successful_enhancement(): cfg = Settings( AUDIO_ENHANCE_ENABLED=True, AUDIO_ENHANCE_BASE_URL="http://mock-enhancer.internal/v1/enhance", AUDIO_ENHANCE_MODEL="voice-cleaner-v1", AUDIO_ENHANCE_TIMEOUT_SECONDS=60.0, ) raw_audio = b"noisy-voice-bytes" enhanced_audio = b"crystal-clear-voice-bytes" trace_id = "test-trace-enh-001" message_id = 12345 events_captured = [] async def fake_event(t_id, stage, ok=True, latency_ms=None, model=None, payload=None, **kwargs): events_captured.append({"stage": stage, "ok": ok, "latency_ms": latency_ms, "payload": payload}) # Mock httpx response mock_resp = httpx.Response( status_code=200, content=enhanced_audio, headers={"content-type": "audio/ogg"}, request=httpx.Request("POST", cfg.AUDIO_ENHANCE_BASE_URL), ) with patch.object(obs, "event", side_effect=fake_event): with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post: mock_post.return_value = mock_resp out_bytes, meta = await enhance_voice( cfg, raw_audio, mime_type="audio/ogg", filename="voice.ogg", trace_id=trace_id, message_id=message_id, ) check("success: enhanced bytes returned", out_bytes == enhanced_audio) check("success: meta indicates enhanced", meta.get("enhanced") is True) check("success: enhanced path exists in meta", bool(meta.get("path"))) check("success: path has _enhanced suffix", "_enhanced" in (meta.get("path") or "")) # Verify physical file existence if meta.get("path"): saved_file = Path(meta["path"]) check("success: enhanced file written to disk", saved_file.exists() and saved_file.read_bytes() == enhanced_audio) # Verify observability event enh_events = [e for e in events_captured if e["stage"] == "audio.enhance.result"] check("obs: audio.enhance.result event emitted", len(enh_events) == 1) if enh_events: ev = enh_events[0] check("obs: event ok=True", ev["ok"] is True) check("obs: payload contains size & latency_ms", "size" in ev["payload"] and ev.get("latency_ms") is not None) check("obs: payload recorded correct size", ev["payload"]["size"] == len(enhanced_audio)) async def test_graceful_fallback_on_error(): cfg = Settings( AUDIO_ENHANCE_ENABLED=True, AUDIO_ENHANCE_BASE_URL="http://mock-enhancer.internal/v1/enhance", AUDIO_ENHANCE_TIMEOUT_SECONDS=60.0, ) raw_audio = b"original-unprocessed-audio" trace_id = "test-trace-enh-002" message_id = 999 events_captured = [] async def fake_event(t_id, stage, ok=True, latency_ms=None, model=None, payload=None, **kwargs): events_captured.append({"stage": stage, "ok": ok, "latency_ms": latency_ms, "payload": payload}) # Mock HTTP 500 error mock_resp = httpx.Response( status_code=500, text="Internal Model Error", request=httpx.Request("POST", cfg.AUDIO_ENHANCE_BASE_URL), ) with patch.object(obs, "event", side_effect=fake_event): with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post: mock_post.return_value = mock_resp out_bytes, meta = await enhance_voice( cfg, raw_audio, mime_type="audio/ogg", trace_id=trace_id, message_id=message_id, ) check("fallback: returns raw original bytes on 500", out_bytes == raw_audio) check("fallback: meta indicates fallback", meta.get("fallback") is True) enh_events = [e for e in events_captured if e["stage"] == "audio.enhance.result"] check("obs: fallback event ok=False", len(enh_events) == 1 and enh_events[0]["ok"] is False) if enh_events: check("obs: error detail captured", "HTTP 500" in enh_events[0]["payload"].get("error", "")) async def test_zero_transcoding_by_default(): cfg = Settings( AUDIO_ENHANCE_ENABLED=True, AUDIO_ENHANCE_BASE_URL="http://mock-enhancer.internal/v1/enhance", AUDIO_ENHANCE_TARGET_FORMAT="", # empty -> zero transcoding ) raw_audio = b"raw-bale-ogg-bytes" captured_payload = None async def fake_post(url, files=None, data=None, headers=None, **kwargs): nonlocal captured_payload captured_payload = files["file"][1] return httpx.Response(status_code=200, content=b"clean", request=httpx.Request("POST", url)) with patch("httpx.AsyncClient.post", side_effect=fake_post): await enhance_voice(cfg, raw_audio, mime_type="audio/ogg") check("transcoding: raw bytes sent directly without modification", captured_payload == raw_audio) async def main(): print("Running audio_enhancer unit tests...") await test_disabled_enhancer() await test_timeout_setting() await test_successful_enhancement() await test_graceful_fallback_on_error() await test_zero_transcoding_by_default() print(f"\nResult: {_passed} passed, {_failed} failed") return 1 if _failed else 0 if __name__ == "__main__": sys.exit(asyncio.run(main()))