diff --git a/src/pipecat/tests/utils.py b/src/pipecat/tests/utils.py index 55bda9cea..7fcecc3ff 100644 --- a/src/pipecat/tests/utils.py +++ b/src/pipecat/tests/utils.py @@ -26,7 +26,8 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor class SleepFrame(SystemFrame): """This frame is used by test framework to introduce some sleep time before the next frame is pushed. This is useful to control system frames vs data or - control frames.""" + control frames. + """ sleep: float = 0.1 diff --git a/tests/test_user_idle_processor.py b/tests/test_user_idle_processor.py new file mode 100644 index 000000000..a2f2fd386 --- /dev/null +++ b/tests/test_user_idle_processor.py @@ -0,0 +1,216 @@ +# +# Copyright (c) 2024-2025 Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import unittest + +from pipecat.frames.frames import ( + BotSpeakingFrame, + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) +from pipecat.processors.user_idle_processor import UserIdleProcessor +from pipecat.tests.utils import SleepFrame, run_test + + +class TestUserIdleProcessor(unittest.IsolatedAsyncioTestCase): + async def test_basic_idle_detection(self): + """Test that idle callback is triggered after timeout when user stops speaking.""" + callback_called = asyncio.Event() + + async def idle_callback(processor: UserIdleProcessor) -> None: + callback_called.set() + + # Create processor with a short timeout for testing + processor = UserIdleProcessor(callback=idle_callback, timeout=0.1) # 100ms timeout + + frames_to_send = [ + # Start conversation + UserStartedSpeakingFrame(), + UserStoppedSpeakingFrame(), + # Wait 200ms - double the idle timeout to ensure it triggers + SleepFrame(sleep=0.2), + ] + + expected_down_frames = [ + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + assert callback_called.is_set(), "Idle callback was not called" + + async def test_active_listening_resets_idle(self): + """Test that bot speaking frames reset the idle timer because user is actively listening.""" + callback_called = asyncio.Event() + + async def idle_callback(processor: UserIdleProcessor) -> None: + callback_called.set() + + processor = UserIdleProcessor(callback=idle_callback, timeout=0.2) + + frames_to_send = [ + # Start conversation + UserStartedSpeakingFrame(), + UserStoppedSpeakingFrame(), + # Wait almost long enough for idle timeout + SleepFrame(sleep=0.1), + # Bot speaking frame should reset idle timer + BotSpeakingFrame(), + # Wait almost long enough for idle timeout again + SleepFrame(sleep=0.1), + # Another bot speaking frame resets timer again + BotSpeakingFrame(), + ] + + expected_down_frames = [ + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, + BotSpeakingFrame, + BotSpeakingFrame, + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + assert not callback_called.is_set(), ( + "Idle callback was called even though bot speaking frames reset the timer" + ) + + async def test_idle_retry_callback(self): + """Test that retry count increases until user activity resets it.""" + retry_counts = [] + + async def retry_callback(processor: UserIdleProcessor, retry_count: int) -> bool: + retry_counts.append(retry_count) + return True # Keep monitoring for idle events + + processor = UserIdleProcessor(callback=retry_callback, timeout=0.4) + + frames_to_send = [ + # Start conversation + UserStartedSpeakingFrame(), + UserStoppedSpeakingFrame(), + # Wait for first idle timeout (count=1) + SleepFrame(sleep=0.5), + # Wait for second idle timeout (count=2) + SleepFrame(sleep=0.5), + # User activity resets the count + UserStartedSpeakingFrame(), + UserStoppedSpeakingFrame(), + # Wait for new idle timeout (count should be 1 again) + SleepFrame(sleep=0.5), + ] + + expected_down_frames = [ + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + assert retry_counts == [1, 2, 1], f"Expected retry counts [1, 2, 1], got {retry_counts}" + + async def test_idle_monitoring_stops_on_false_return(self): + """Test that idle monitoring stops when callback returns False.""" + retry_counts = [] + + async def retry_callback(processor: UserIdleProcessor, retry_count: int) -> bool: + retry_counts.append(retry_count) + return retry_count < 2 # Stop after second retry + + processor = UserIdleProcessor(callback=retry_callback, timeout=0.4) + + frames_to_send = [ + UserStartedSpeakingFrame(), + UserStoppedSpeakingFrame(), + SleepFrame(sleep=0.5), # First retry (count=1, returns True) + SleepFrame(sleep=0.5), # Second retry (count=2, returns False) + SleepFrame(sleep=0.5), # Should not trigger callback + ] + + expected_down_frames = [ + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + assert retry_counts == [1, 2], f"Expected retry counts [1, 2], got {retry_counts}" + + async def test_no_idle_before_conversation(self): + """Test that idle monitoring doesn't start before first conversation activity.""" + callback_called = asyncio.Event() + + async def idle_callback(processor: UserIdleProcessor) -> None: + callback_called.set() + + processor = UserIdleProcessor(callback=idle_callback, timeout=0.1) + + frames_to_send = [ + SleepFrame(sleep=0.2), # Should not trigger callback + # No conversation activity yet + ] + + expected_down_frames = [] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + assert not callback_called.is_set(), "Idle callback was called before conversation started" + + async def test_idle_starts_with_bot_speech(self): + """Test that monitoring starts with bot speaking frames, not just user speech.""" + callback_called = asyncio.Event() + + async def idle_callback(processor: UserIdleProcessor) -> None: + callback_called.set() + + processor = UserIdleProcessor(callback=idle_callback, timeout=0.1) + + frames_to_send = [ + BotStartedSpeakingFrame(), + BotSpeakingFrame(), + BotStoppedSpeakingFrame(), + SleepFrame(sleep=0.2), + ] + + expected_down_frames = [ + BotStartedSpeakingFrame, + BotSpeakingFrame, + BotStoppedSpeakingFrame, + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + assert callback_called.is_set(), "Idle callback not called after bot speech"