e78dc4088a
- Add begin_response and finish_after_current_speech methods to CallEndCoordinator for better management of speech events. - Update PromptBrain to utilize new methods, ensuring proper handling of generated closing speech and tool-only calls. - Enhance tests to verify the correct behavior of speech tracking and response handling in various scenarios, including waiting for audio to finish before ending calls. - Introduce a new test suite for CallEndCoordinator to validate the interaction with speech frames.
50 lines
1.7 KiB
Python
50 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from pipecat.frames.frames import BotStartedSpeakingFrame, BotStoppedSpeakingFrame
|
|
from services.pipecat.call_lifecycle import CallEndCoordinator
|
|
|
|
|
|
class CallEndCoordinatorTest(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
self.reasons: list[str] = []
|
|
|
|
async def queue_end(reason: str) -> None:
|
|
self.reasons.append(reason)
|
|
|
|
self.coordinator = CallEndCoordinator(queue_end)
|
|
|
|
async def test_generated_closing_text_waits_for_audio_stop(self):
|
|
self.coordinator.begin_response()
|
|
self.coordinator.begin("prompt_end_call")
|
|
await self.coordinator.finish_after_current_speech(has_text=True)
|
|
|
|
self.assertEqual(self.reasons, [])
|
|
await self.coordinator.observe(BotStartedSpeakingFrame())
|
|
self.assertEqual(self.reasons, [])
|
|
await self.coordinator.observe(BotStoppedSpeakingFrame())
|
|
self.assertEqual(self.reasons, ["prompt_end_call"])
|
|
|
|
async def test_already_played_generated_text_finishes_immediately(self):
|
|
self.coordinator.begin_response()
|
|
await self.coordinator.observe(BotStartedSpeakingFrame())
|
|
await self.coordinator.observe(BotStoppedSpeakingFrame())
|
|
self.coordinator.begin("prompt_end_call")
|
|
|
|
await self.coordinator.finish_after_current_speech(has_text=True)
|
|
|
|
self.assertEqual(self.reasons, ["prompt_end_call"])
|
|
|
|
async def test_tool_only_end_call_finishes_without_waiting(self):
|
|
self.coordinator.begin_response()
|
|
self.coordinator.begin("tool_only")
|
|
|
|
await self.coordinator.finish_after_current_speech(has_text=False)
|
|
|
|
self.assertEqual(self.reasons, ["tool_only"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|