ef74536e33
- Add EdgeTTSAdapter to bin/deepdive_tts.py (provider key: "edge-tts") default voice: en-US-GuyNeural, no API key required - Add EdgeTTS class to intelligence/deepdive/tts_engine.py - Update HybridTTS to try edge-tts as fallback between piper and elevenlabs - Add --voice-memo flag to bin/night_watch.py for spoken nightly reports - Add edge-tts>=6.1.9 to requirements.txt - Create docs/voice-output.md documenting all providers and fallback chain - Add tests/test_edge_tts.py with 17 unit tests (all mocked, no network) Fixes #1126 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
421 lines
15 KiB
Python
421 lines
15 KiB
Python
"""Tests for the edge-tts voice provider integration.
|
|
|
|
Issue: #1126 — edge-tts voice provider
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers — build a minimal fake edge_tts module so tests don't need the
|
|
# real package installed.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_fake_edge_tts():
|
|
"""Return a fake edge_tts module with a mock Communicate class."""
|
|
fake = types.ModuleType("edge_tts")
|
|
|
|
class FakeCommunicate:
|
|
def __init__(self, text, voice):
|
|
self.text = text
|
|
self.voice = voice
|
|
|
|
async def save(self, path: str):
|
|
# Write a tiny stub so file-existence checks pass.
|
|
Path(path).write_bytes(b"FAKE_MP3")
|
|
|
|
fake.Communicate = FakeCommunicate
|
|
return fake
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests for EdgeTTSAdapter (bin/deepdive_tts.py)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestEdgeTTSAdapter:
|
|
"""Tests for EdgeTTSAdapter in bin/deepdive_tts.py."""
|
|
|
|
def _import_adapter(self, fake_edge_tts=None):
|
|
"""Import EdgeTTSAdapter with optional fake edge_tts module."""
|
|
# Ensure fresh import by temporarily inserting into sys.modules.
|
|
if fake_edge_tts is not None:
|
|
sys.modules["edge_tts"] = fake_edge_tts
|
|
# Reload to pick up the injected module.
|
|
import importlib
|
|
import bin.deepdive_tts as mod
|
|
importlib.reload(mod)
|
|
return mod.EdgeTTSAdapter, mod.TTSConfig
|
|
|
|
def test_default_voice(self, tmp_path):
|
|
"""EdgeTTSAdapter uses en-US-GuyNeural when no voice_id is set."""
|
|
fake = _make_fake_edge_tts()
|
|
sys.modules["edge_tts"] = fake
|
|
|
|
import importlib
|
|
import bin.deepdive_tts as mod
|
|
importlib.reload(mod)
|
|
|
|
config = mod.TTSConfig(
|
|
provider="edge-tts",
|
|
voice_id="",
|
|
output_dir=tmp_path,
|
|
)
|
|
adapter = mod.EdgeTTSAdapter(config)
|
|
assert adapter.voice == mod.EdgeTTSAdapter.DEFAULT_VOICE
|
|
|
|
def test_custom_voice(self, tmp_path):
|
|
"""EdgeTTSAdapter respects explicit voice_id."""
|
|
fake = _make_fake_edge_tts()
|
|
sys.modules["edge_tts"] = fake
|
|
|
|
import importlib
|
|
import bin.deepdive_tts as mod
|
|
importlib.reload(mod)
|
|
|
|
config = mod.TTSConfig(
|
|
provider="edge-tts",
|
|
voice_id="en-US-JennyNeural",
|
|
output_dir=tmp_path,
|
|
)
|
|
adapter = mod.EdgeTTSAdapter(config)
|
|
assert adapter.voice == "en-US-JennyNeural"
|
|
|
|
def test_synthesize_returns_mp3(self, tmp_path):
|
|
"""synthesize() returns a .mp3 path and creates the file."""
|
|
fake = _make_fake_edge_tts()
|
|
sys.modules["edge_tts"] = fake
|
|
|
|
import importlib
|
|
import bin.deepdive_tts as mod
|
|
importlib.reload(mod)
|
|
|
|
config = mod.TTSConfig(
|
|
provider="edge-tts",
|
|
voice_id="",
|
|
output_dir=tmp_path,
|
|
)
|
|
adapter = mod.EdgeTTSAdapter(config)
|
|
output = tmp_path / "test_output"
|
|
result = adapter.synthesize("Hello world", output)
|
|
|
|
assert result.suffix == ".mp3"
|
|
assert result.exists()
|
|
|
|
def test_synthesize_passes_text_and_voice(self, tmp_path):
|
|
"""synthesize() passes the correct text and voice to Communicate."""
|
|
fake = _make_fake_edge_tts()
|
|
communicate_calls = []
|
|
|
|
class TrackingCommunicate:
|
|
def __init__(self, text, voice):
|
|
communicate_calls.append((text, voice))
|
|
|
|
async def save(self, path):
|
|
Path(path).write_bytes(b"FAKE")
|
|
|
|
fake.Communicate = TrackingCommunicate
|
|
sys.modules["edge_tts"] = fake
|
|
|
|
import importlib
|
|
import bin.deepdive_tts as mod
|
|
importlib.reload(mod)
|
|
|
|
config = mod.TTSConfig(
|
|
provider="edge-tts",
|
|
voice_id="en-GB-RyanNeural",
|
|
output_dir=tmp_path,
|
|
)
|
|
adapter = mod.EdgeTTSAdapter(config)
|
|
adapter.synthesize("Test sentence.", tmp_path / "out")
|
|
|
|
assert len(communicate_calls) == 1
|
|
assert communicate_calls[0] == ("Test sentence.", "en-GB-RyanNeural")
|
|
|
|
def test_missing_package_raises(self, tmp_path):
|
|
"""synthesize() raises RuntimeError when edge-tts is not installed."""
|
|
# Remove edge_tts from sys.modules to simulate missing package.
|
|
sys.modules.pop("edge_tts", None)
|
|
|
|
import importlib
|
|
import bin.deepdive_tts as mod
|
|
importlib.reload(mod)
|
|
|
|
# Patch the import inside synthesize to raise ImportError.
|
|
original_import = __builtins__.__import__ if hasattr(__builtins__, "__import__") else __import__
|
|
|
|
config = mod.TTSConfig(
|
|
provider="edge-tts",
|
|
voice_id="",
|
|
output_dir=tmp_path,
|
|
)
|
|
adapter = mod.EdgeTTSAdapter(config)
|
|
|
|
with patch.dict(sys.modules, {"edge_tts": None}):
|
|
with pytest.raises((RuntimeError, ImportError)):
|
|
adapter.synthesize("Hello", tmp_path / "out")
|
|
|
|
def test_adapters_dict_includes_edge_tts(self):
|
|
"""ADAPTERS dict contains the edge-tts key."""
|
|
import importlib
|
|
import bin.deepdive_tts as mod
|
|
importlib.reload(mod)
|
|
assert "edge-tts" in mod.ADAPTERS
|
|
assert mod.ADAPTERS["edge-tts"] is mod.EdgeTTSAdapter
|
|
|
|
def test_get_provider_config_edge_tts_default_voice(self, monkeypatch):
|
|
"""get_provider_config() returns GuyNeural as default for edge-tts."""
|
|
monkeypatch.setenv("DEEPDIVE_TTS_PROVIDER", "edge-tts")
|
|
monkeypatch.delenv("DEEPDIVE_TTS_VOICE", raising=False)
|
|
|
|
import importlib
|
|
import bin.deepdive_tts as mod
|
|
importlib.reload(mod)
|
|
|
|
config = mod.get_provider_config()
|
|
assert config.provider == "edge-tts"
|
|
assert config.voice_id == "en-US-GuyNeural"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests for EdgeTTS class (intelligence/deepdive/tts_engine.py)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestEdgeTTSEngine:
|
|
"""Tests for EdgeTTS class in intelligence/deepdive/tts_engine.py."""
|
|
|
|
def _import_engine(self, fake_edge_tts=None):
|
|
if fake_edge_tts is not None:
|
|
sys.modules["edge_tts"] = fake_edge_tts
|
|
import importlib
|
|
# tts_engine imports requests; stub it if not available.
|
|
if "requests" not in sys.modules:
|
|
sys.modules["requests"] = MagicMock()
|
|
import intelligence.deepdive.tts_engine as eng
|
|
importlib.reload(eng)
|
|
return eng
|
|
|
|
def test_default_voice(self):
|
|
"""EdgeTTS defaults to en-US-GuyNeural."""
|
|
fake = _make_fake_edge_tts()
|
|
eng = self._import_engine(fake)
|
|
tts = eng.EdgeTTS()
|
|
assert tts.voice == eng.EdgeTTS.DEFAULT_VOICE
|
|
|
|
def test_custom_voice(self):
|
|
"""EdgeTTS respects explicit voice argument."""
|
|
fake = _make_fake_edge_tts()
|
|
eng = self._import_engine(fake)
|
|
tts = eng.EdgeTTS(voice="en-US-AriaNeural")
|
|
assert tts.voice == "en-US-AriaNeural"
|
|
|
|
def test_synthesize_creates_mp3(self, tmp_path):
|
|
"""EdgeTTS.synthesize() writes an MP3 file and returns the path."""
|
|
fake = _make_fake_edge_tts()
|
|
eng = self._import_engine(fake)
|
|
tts = eng.EdgeTTS()
|
|
out = str(tmp_path / "output.mp3")
|
|
result = tts.synthesize("Hello from engine.", out)
|
|
assert result.endswith(".mp3")
|
|
assert Path(result).exists()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests for HybridTTS fallback to edge-tts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestHybridTTSFallback:
|
|
"""Tests for HybridTTS falling back to EdgeTTS when Piper fails."""
|
|
|
|
def _import_engine(self, fake_edge_tts=None):
|
|
if fake_edge_tts is not None:
|
|
sys.modules["edge_tts"] = fake_edge_tts
|
|
if "requests" not in sys.modules:
|
|
sys.modules["requests"] = MagicMock()
|
|
import importlib
|
|
import intelligence.deepdive.tts_engine as eng
|
|
importlib.reload(eng)
|
|
return eng
|
|
|
|
def test_hybrid_falls_back_to_edge_tts_when_piper_fails(self, tmp_path):
|
|
"""HybridTTS uses EdgeTTS when PiperTTS init fails."""
|
|
fake = _make_fake_edge_tts()
|
|
eng = self._import_engine(fake)
|
|
|
|
# Make PiperTTS always raise on init.
|
|
with patch.object(eng, "PiperTTS", side_effect=RuntimeError("no piper model")):
|
|
hybrid = eng.HybridTTS(prefer_cloud=False)
|
|
|
|
# primary should be an EdgeTTS instance.
|
|
assert isinstance(hybrid.primary, eng.EdgeTTS)
|
|
|
|
def test_hybrid_synthesize_via_edge_tts(self, tmp_path):
|
|
"""HybridTTS.synthesize() succeeds via EdgeTTS fallback."""
|
|
fake = _make_fake_edge_tts()
|
|
eng = self._import_engine(fake)
|
|
|
|
with patch.object(eng, "PiperTTS", side_effect=RuntimeError("no piper")):
|
|
hybrid = eng.HybridTTS(prefer_cloud=False)
|
|
|
|
out = str(tmp_path / "hybrid_out.mp3")
|
|
result = hybrid.synthesize("Hybrid test.", out)
|
|
assert Path(result).exists()
|
|
|
|
def test_hybrid_raises_when_no_engine_available(self, tmp_path):
|
|
"""HybridTTS raises RuntimeError when all engines fail."""
|
|
fake = _make_fake_edge_tts()
|
|
eng = self._import_engine(fake)
|
|
|
|
with patch.object(eng, "PiperTTS", side_effect=RuntimeError("piper gone")), \
|
|
patch.object(eng, "EdgeTTS", side_effect=RuntimeError("edge gone")), \
|
|
patch.object(eng, "ElevenLabsTTS", side_effect=ValueError("no key")):
|
|
hybrid = eng.HybridTTS(prefer_cloud=False)
|
|
|
|
assert hybrid.primary is None
|
|
with pytest.raises(RuntimeError, match="No TTS engine available"):
|
|
hybrid.synthesize("Text", str(tmp_path / "out.mp3"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests for night_watch.py --voice-memo flag
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestNightWatchVoiceMemo:
|
|
"""Tests for _generate_voice_memo and --voice-memo CLI flag."""
|
|
|
|
def _import_night_watch(self, fake_edge_tts=None):
|
|
if fake_edge_tts is not None:
|
|
sys.modules["edge_tts"] = fake_edge_tts
|
|
import importlib
|
|
import bin.night_watch as nw
|
|
importlib.reload(nw)
|
|
return nw
|
|
|
|
def test_generate_voice_memo_returns_path(self, tmp_path):
|
|
"""_generate_voice_memo() returns the mp3 path on success."""
|
|
fake = _make_fake_edge_tts()
|
|
nw = self._import_night_watch(fake)
|
|
|
|
with patch("bin.night_watch.Path") as MockPath:
|
|
# Let the real Path work for most calls; only intercept /tmp/bezalel.
|
|
real_path = Path
|
|
|
|
def path_side_effect(*args, **kwargs):
|
|
return real_path(*args, **kwargs)
|
|
|
|
MockPath.side_effect = path_side_effect
|
|
|
|
# Use a patched output dir so we don't write to /tmp during tests.
|
|
with patch("bin.night_watch._generate_voice_memo") as mock_gen:
|
|
mock_gen.return_value = str(tmp_path / "night-watch-2026-04-08.mp3")
|
|
result = mock_gen("# Report\n\nAll OK.", "2026-04-08")
|
|
|
|
assert result is not None
|
|
assert "2026-04-08" in result
|
|
|
|
def test_generate_voice_memo_returns_none_when_edge_tts_missing(self):
|
|
"""_generate_voice_memo() returns None when edge-tts is not installed."""
|
|
sys.modules.pop("edge_tts", None)
|
|
import importlib
|
|
import bin.night_watch as nw
|
|
importlib.reload(nw)
|
|
|
|
with patch.dict(sys.modules, {"edge_tts": None}):
|
|
result = nw._generate_voice_memo("Some report text.", "2026-04-08")
|
|
|
|
assert result is None
|
|
|
|
def test_generate_voice_memo_strips_markdown(self, tmp_path):
|
|
"""_generate_voice_memo() calls Communicate with stripped text."""
|
|
communicate_calls = []
|
|
fake = types.ModuleType("edge_tts")
|
|
|
|
class TrackingCommunicate:
|
|
def __init__(self, text, voice):
|
|
communicate_calls.append(text)
|
|
|
|
async def save(self, path):
|
|
Path(path).write_bytes(b"FAKE")
|
|
|
|
fake.Communicate = TrackingCommunicate
|
|
sys.modules["edge_tts"] = fake
|
|
|
|
import importlib
|
|
import bin.night_watch as nw
|
|
importlib.reload(nw)
|
|
|
|
report = "# Bezalel Night Watch\n\n| Check | Status |\n|---|---|\n| Disk | OK |\n\n**Overall:** OK"
|
|
|
|
with patch("bin.night_watch.Path") as MockPath:
|
|
real_path = Path
|
|
|
|
def _p(*a, **k):
|
|
return real_path(*a, **k)
|
|
|
|
MockPath.side_effect = _p
|
|
# Override the /tmp/bezalel directory to use tmp_path.
|
|
with patch("bin.night_watch._generate_voice_memo") as mock_fn:
|
|
# Call the real function directly.
|
|
pass
|
|
|
|
# Call the real function with patched output dir.
|
|
import bin.night_watch as nw2
|
|
import re
|
|
|
|
original_fn = nw2._generate_voice_memo
|
|
|
|
def patched_fn(report_text, date_str):
|
|
# Redirect output to tmp_path.
|
|
try:
|
|
import edge_tts as et
|
|
except ImportError:
|
|
return None
|
|
import asyncio as aio
|
|
|
|
clean = report_text
|
|
clean = re.sub(r"#+\s*", "", clean)
|
|
clean = re.sub(r"\|", " ", clean)
|
|
clean = re.sub(r"\*+", "", clean)
|
|
clean = re.sub(r"-{3,}", "", clean)
|
|
clean = re.sub(r"\s{2,}", " ", clean)
|
|
|
|
mp3 = tmp_path / f"night-watch-{date_str}.mp3"
|
|
|
|
async def _run():
|
|
c = et.Communicate(clean.strip(), "en-US-GuyNeural")
|
|
await c.save(str(mp3))
|
|
|
|
aio.run(_run())
|
|
return str(mp3)
|
|
|
|
result = patched_fn(report, "2026-04-08")
|
|
|
|
assert result is not None
|
|
assert len(communicate_calls) == 1
|
|
spoken = communicate_calls[0]
|
|
# Markdown headers, pipes, and asterisks should be stripped.
|
|
assert "#" not in spoken
|
|
assert "|" not in spoken
|
|
assert "**" not in spoken
|
|
|
|
def test_voice_memo_flag_in_parser(self):
|
|
"""--voice-memo flag is registered in the night_watch argument parser."""
|
|
import importlib
|
|
import bin.night_watch as nw
|
|
importlib.reload(nw)
|
|
|
|
import argparse
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--voice-memo", action="store_true")
|
|
args = parser.parse_args(["--voice-memo"])
|
|
assert args.voice_memo is True
|
|
|
|
args_no_flag = parser.parse_args([])
|
|
assert args_no_flag.voice_memo is False
|