tests: add unit test to check for idle timeout on swallowed frames

tests/test_pipeline.py

@@ -24,6 +24,7 @@ from pipecat.pipeline.base_task import PipelineTaskParams
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.filters.frame_filter import FrameFilter
 from pipecat.processors.filters.identity_filter import IdentityFilter
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.tests.utils import HeartbeatsObserver, run_test
@@ -383,6 +384,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
         idle_timeout_secs = 0.2
         sleep_time_secs = idle_timeout_secs / 2
 
+        # Use the identify filter so the frames just reach the end of the pipeline.
         identity = IdentityFilter()
         pipeline = Pipeline([identity])
         task = PipelineTask(
@@ -392,6 +394,12 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
         )
 
         async def delayed_frames():
+            """Sending multiple text frames.
+
+            The total amount of elapsed time in this function should be greater
+            than the task idle timeout. If an idle timeout event is triggered it
+            means we haven't detected that the TextFrames have been pushed.
+            """
             await asyncio.sleep(sleep_time_secs)
             await task.queue_frame(TextFrame("Hello Pipecat!"))
             await asyncio.sleep(sleep_time_secs)
@@ -415,6 +423,51 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
         # Wait for the pending tasks to complete.
         await asyncio.gather(*pending)
 
+    async def test_idle_task_swallowed_frames(self):
+        idle_timeout_secs = 0.2
+        sleep_time_secs = idle_timeout_secs / 2
+
+        # Block all frames (except system frames). Here, we are testing that
+        # generated frames don't trigger an idle timeout (they don't need to
+        # reach the end of the pipeline).
+        filter = FrameFilter(types=())
+        pipeline = Pipeline([filter])
+        task = PipelineTask(
+            pipeline,
+            idle_timeout_secs=idle_timeout_secs,
+            idle_timeout_frames=(TextFrame,),
+        )
+
+        start_time = time.time()
+
+        async def delayed_frames():
+            """Sending multiple text frames.
+
+            The total amount of elapsed time in this function should be greater
+            than the task idle timeout. If an idle timeout event is triggered it
+            means we haven't detected that the TextFrames have been pushed.
+            """
+            await asyncio.sleep(sleep_time_secs)
+            await task.queue_frame(TextFrame("Hello Pipecat!"))
+            await asyncio.sleep(sleep_time_secs)
+            await task.queue_frame(TextFrame("Hello Pipecat!"))
+            await asyncio.sleep(sleep_time_secs)
+            await task.queue_frame(TextFrame("Hello Pipecat!"))
+
+        tasks = [
+            asyncio.create_task(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))),
+            asyncio.create_task(delayed_frames()),
+        ]
+
+        _, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
+
+        diff_time = time.time() - start_time
+
+        self.assertGreater(diff_time, sleep_time_secs * 3)
+
+        # Wait for the pending tasks to complete.
+        await asyncio.gather(*pending)
+
     async def test_task_cancel_timeout(self):
         class CancelFilter(FrameProcessor):
             def __init__(self, **kwargs):