53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
import json
|
|
import os
|
|
import queue
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from agent_call.bailian import (
|
|
BailianTTS,
|
|
_BailianCosyVoiceCallback,
|
|
build_bailian_config,
|
|
)
|
|
|
|
|
|
class BailianAdapterTests(unittest.TestCase):
|
|
def test_qwen_realtime_url_is_derived_from_asr_endpoint(self) -> None:
|
|
adapter = BailianTTS("wss://example.test/api-ws/v1/inference/", "secret")
|
|
self.assertEqual(
|
|
adapter._url_for_qwen_realtime(),
|
|
"wss://example.test/api-ws/v1/realtime",
|
|
)
|
|
|
|
def test_config_selects_cosyvoice_for_custom_voice(self) -> None:
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"BAILIAN_TTS_VOICE": "cosyvoice-v3.5-plus-bailian-test",
|
|
"BAILIAN_TTS_MODEL": "",
|
|
},
|
|
clear=False,
|
|
):
|
|
config = build_bailian_config("agent_test", "回答收到。")
|
|
self.assertEqual(config["tts"]["model"], "cosyvoice-v3.5-plus")
|
|
self.assertEqual(config["asr"]["provider_ref"], "bailian")
|
|
self.assertEqual(config["asr"]["model"], "fun-asr-realtime")
|
|
|
|
def test_cosyvoice_callback_records_returned_model_when_present(self) -> None:
|
|
callback = _BailianCosyVoiceCallback(4)
|
|
callback.on_event(json.dumps({"header": {"model": "cosyvoice-test"}}))
|
|
self.assertEqual(callback.provider_model, "cosyvoice-test")
|
|
|
|
def test_cosyvoice_callback_is_bounded(self) -> None:
|
|
callback = _BailianCosyVoiceCallback(1)
|
|
callback.on_data(b"\x00\x00")
|
|
callback.on_complete()
|
|
self.assertTrue(callback.overflowed)
|
|
self.assertEqual(callback.events.get_nowait()["type"], "__audio__")
|
|
with self.assertRaises(queue.Empty):
|
|
callback.events.get_nowait()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|