tests: move test_ai_services to test_utils_string
This commit is contained in:
@@ -1,43 +0,0 @@
|
||||
#
|
||||
# Copyright (c) 2024-2025 Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import asyncio
|
||||
import unittest
|
||||
|
||||
import openai
|
||||
import pyaudio
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from pipecat.frames.frames import AudioRawFrame, ErrorFrame
|
||||
from pipecat.services.openai import OpenAITTSService
|
||||
|
||||
load_dotenv()
|
||||
|
||||
|
||||
class TestWhisperOpenAIService(unittest.IsolatedAsyncioTestCase):
|
||||
@unittest.skip("FIXME: This test is failing")
|
||||
async def test_whisper_tts(self):
|
||||
pa = pyaudio.PyAudio()
|
||||
stream = pa.open(format=pyaudio.paInt16, channels=1, rate=24_000, output=True)
|
||||
|
||||
tts = OpenAITTSService(voice="nova")
|
||||
|
||||
async for frame in tts.run_tts("Hello, there. Nice to meet you, seems to work well"):
|
||||
self.assertIsInstance(frame, AudioRawFrame)
|
||||
stream.write(frame.audio)
|
||||
|
||||
await asyncio.sleep(0.5)
|
||||
stream.stop_stream()
|
||||
pa.terminate()
|
||||
|
||||
tts = OpenAITTSService(voice="invalid_voice")
|
||||
with self.assertRaises(openai.BadRequestError):
|
||||
async for frame in tts.run_tts("wont work"):
|
||||
self.assertIsInstance(frame, ErrorFrame)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -5,30 +5,11 @@
|
||||
#
|
||||
|
||||
import unittest
|
||||
from typing import AsyncGenerator
|
||||
|
||||
from pipecat.frames.frames import EndFrame, Frame, TextFrame
|
||||
from pipecat.services.ai_services import AIService, match_endofsentence
|
||||
from pipecat.utils.string import match_endofsentence
|
||||
|
||||
|
||||
class SimpleAIService(AIService):
|
||||
async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
|
||||
yield frame
|
||||
|
||||
|
||||
class TestBaseAIService(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_simple_processing(self):
|
||||
service = SimpleAIService()
|
||||
|
||||
input_frames = [TextFrame("hello"), EndFrame()]
|
||||
|
||||
output_frames = []
|
||||
for input_frame in input_frames:
|
||||
async for output_frame in service.process_frame(input_frame):
|
||||
output_frames.append(output_frame)
|
||||
|
||||
self.assertEqual(input_frames, output_frames)
|
||||
|
||||
class TestUtilsString(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_endofsentence(self):
|
||||
assert match_endofsentence("This is a sentence.")
|
||||
assert match_endofsentence("This is a sentence! ")
|
||||
@@ -57,7 +38,3 @@ class TestBaseAIService(unittest.IsolatedAsyncioTestCase):
|
||||
for i in chinese_sentences:
|
||||
assert match_endofsentence(i)
|
||||
assert not match_endofsentence("你好,")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user