Adolfo Reyna
143 lines
4.9 KiB
Python
143 lines
4.9 KiB
Python
import json
|
|
from pathlib import Path
|
|
from typer.testing import CliRunner
|
|
|
|
from reyna_cli.cli import app
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
def test_media_execution_local_tts_delegates(monkeypatch, tmp_path):
|
|
from reyna_cli import media_execution
|
|
|
|
target = tmp_path / "speech.wav"
|
|
|
|
def fake_synthesize(text, output, **kwargs):
|
|
output.write_bytes(b"RIFFfake")
|
|
return output
|
|
|
|
monkeypatch.setattr(media_execution, "synthesize_wav", fake_synthesize)
|
|
result = media_execution.generate_local_tts("hello", target, voice="Alex")
|
|
assert result["ok"] is True
|
|
assert result["filePath"] == str(target)
|
|
assert target.read_bytes() == b"RIFFfake"
|
|
|
|
|
|
def test_media_execution_kokoro_writes_returned_file(monkeypatch, tmp_path):
|
|
from reyna_cli import media_execution
|
|
|
|
source = tmp_path / "source.wav"
|
|
source.write_bytes(b"wav")
|
|
target = tmp_path / "target.wav"
|
|
monkeypatch.setattr(media_execution, "_json_request", lambda *args, **kwargs: {"filePath": str(source), "voice": "af_heart"})
|
|
result = media_execution.generate_kokoro("hello", target, url="http://127.0.0.1:7332")
|
|
assert result["ok"] is True
|
|
assert target.read_bytes() == b"wav"
|
|
|
|
|
|
def test_media_execution_gemini_requires_key(monkeypatch, tmp_path):
|
|
from reyna_cli import media_execution
|
|
|
|
monkeypatch.delenv("GEMINI_API_KEY", raising=False)
|
|
try:
|
|
media_execution.generate_gemini_image("a tree", tmp_path / "tree.png")
|
|
except RuntimeError as exc:
|
|
assert "GEMINI_API_KEY" in str(exc)
|
|
else:
|
|
raise AssertionError("missing Gemini key must fail before network")
|
|
|
|
|
|
def test_media_execution_gemini_decodes_image(monkeypatch, tmp_path):
|
|
from reyna_cli import media_execution
|
|
|
|
monkeypatch.setenv("GEMINI_API_KEY", "test-only")
|
|
monkeypatch.setattr(media_execution, "_json_request", lambda *args, **kwargs: {"id": "i1", "output_image": {"data": "aGVsbG8=", "mime_type": "image/png"}})
|
|
target = tmp_path / "tree.png"
|
|
result = media_execution.generate_gemini_image("a tree", target)
|
|
assert result["path"] == str(target)
|
|
assert target.read_bytes() == b"hello"
|
|
|
|
|
|
def test_cli_exposes_execution_commands():
|
|
for args in [
|
|
["local-services", "speech", "--help"],
|
|
["local-services", "kokoro", "--help"],
|
|
["local-services", "voicebox", "--help"],
|
|
["local-services", "image", "--help"],
|
|
]:
|
|
result = runner.invoke(app, args)
|
|
assert result.exit_code == 0, result.stdout
|
|
assert "generate" in result.stdout
|
|
|
|
|
|
def test_cli_image_rejects_unknown_engine():
|
|
result = runner.invoke(app, ["local-services", "image", "generate", "tree", "--engine", "unknown", "--json"])
|
|
assert result.exit_code == 1
|
|
assert "codex or gemini" in result.stdout
|
|
assert json.loads(result.stdout)["ok"] is False
|
|
|
|
|
|
def test_speech_file_execution_uses_cached_binary(monkeypatch, tmp_path):
|
|
from reyna_cli import speech_execution
|
|
|
|
audio = tmp_path / "sample.wav"
|
|
audio.write_bytes(b"wav")
|
|
binary = tmp_path / "transcriber"
|
|
binary.write_bytes(b"binary")
|
|
|
|
monkeypatch.setattr(speech_execution, "_ensure_binary", lambda: binary)
|
|
monkeypatch.setattr(
|
|
speech_execution.subprocess,
|
|
"run",
|
|
lambda *args, **kwargs: type("Result", (), {"returncode": 0, "stdout": '{"ok":true,"transcript":"hello"}', "stderr": ""})(),
|
|
)
|
|
result = speech_execution.transcribe_file(audio, locale="en-US")
|
|
assert result["transcript"] == "hello"
|
|
assert result["source"] == "reyna_cli_direct"
|
|
|
|
|
|
def test_speech_file_execution_rejects_missing_audio(tmp_path):
|
|
from reyna_cli.speech_execution import SpeechExecutionError, transcribe_file
|
|
|
|
try:
|
|
transcribe_file(tmp_path / "missing.wav")
|
|
except SpeechExecutionError as exc:
|
|
assert "not found" in str(exc)
|
|
else:
|
|
raise AssertionError("missing audio must fail before compilation")
|
|
|
|
|
|
def test_live_session_frames_audio_and_reads_event(monkeypatch, tmp_path):
|
|
from reyna_cli import speech_live
|
|
|
|
class FakeStdin:
|
|
def __init__(self):
|
|
self.data = bytearray()
|
|
def write(self, value):
|
|
self.data.extend(value)
|
|
def flush(self):
|
|
pass
|
|
def close(self):
|
|
pass
|
|
|
|
class FakeProcess:
|
|
def __init__(self):
|
|
self.stdin = FakeStdin()
|
|
self.stdout = []
|
|
def poll(self):
|
|
return None
|
|
def wait(self, timeout=None):
|
|
return 0
|
|
def kill(self):
|
|
pass
|
|
|
|
fake = FakeProcess()
|
|
monkeypatch.setattr(speech_live, "_ensure_binary", lambda: tmp_path / "live")
|
|
monkeypatch.setattr(speech_live.subprocess, "Popen", lambda *args, **kwargs: fake)
|
|
session = speech_live.SpeechLiveSession("en-US")
|
|
session.events.put({"ok": True, "event": "final", "text": "hello"})
|
|
result = session.transcribe_chunk(b"wav")
|
|
assert result["text"] == "hello"
|
|
assert bytes(fake.stdin.data[:4]) == b"\x00\x00\x00\x03"
|
|
session.close()
|