82 lines
3.1 KiB
Python
82 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
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_wait_until_silent_tracks_transport_boundary(self):
|
|
self.assertFalse(self.coordinator.speaking)
|
|
await self.coordinator.wait_until_silent()
|
|
|
|
await self.coordinator.observe(BotStartedSpeakingFrame())
|
|
self.assertTrue(self.coordinator.speaking)
|
|
wait_task = asyncio.create_task(self.coordinator.wait_until_silent())
|
|
await asyncio.sleep(0)
|
|
self.assertFalse(wait_task.done())
|
|
|
|
await self.coordinator.observe(BotStoppedSpeakingFrame())
|
|
await wait_task
|
|
self.assertFalse(self.coordinator.speaking)
|
|
|
|
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"])
|
|
|
|
async def test_workflow_end_waits_for_every_queued_fixed_speech(self):
|
|
first_completion = self.coordinator.track_speech()
|
|
second_completion = self.coordinator.track_speech()
|
|
self.coordinator.begin("workflow_completed")
|
|
await self.coordinator.arm_after_tracked_speech()
|
|
|
|
await self.coordinator.observe(BotStartedSpeakingFrame())
|
|
await self.coordinator.observe(BotStoppedSpeakingFrame())
|
|
self.assertTrue(first_completion.done())
|
|
self.assertFalse(second_completion.done())
|
|
self.assertEqual(self.reasons, [])
|
|
|
|
await self.coordinator.observe(BotStartedSpeakingFrame())
|
|
await self.coordinator.observe(BotStoppedSpeakingFrame())
|
|
self.assertTrue(second_completion.done())
|
|
self.assertEqual(self.reasons, ["workflow_completed"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|