wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tests: make sure QueuedFrameProcessor push frames

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-01-30 13:48:44 -08:00
parent b65f32e8e1
commit d95819746a
2 changed files with 15 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
src/pipecat/tests/utils.py
View File

@@ -17,6 +17,7 @@ from pipecat.frames.frames import (
) )
from pipecat.observers.base_observer import BaseObserver from pipecat.observers.base_observer import BaseObserver
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.asyncio import TaskManager
@dataclass @dataclass
@@ -54,9 +55,12 @@ class QueuedFrameProcessor(FrameProcessor):
async def process_frame(self, frame: Frame, direction: FrameDirection): async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction) await super().process_frame(frame, direction)
if self._ignore_start and isinstance(frame, StartFrame): if self._ignore_start and isinstance(frame, StartFrame):
return await self.push_frame(frame, direction)
await self._queue.put(frame) else:
await self._queue.put(frame)
await self.push_frame(frame, direction)
async def run_test( async def run_test(
@@ -67,13 +71,15 @@ async def run_test(
) -> Tuple[Sequence[Frame], Sequence[Frame]]: ) -> Tuple[Sequence[Frame], Sequence[Frame]]:
received_up = asyncio.Queue() received_up = asyncio.Queue()
received_down = asyncio.Queue() received_down = asyncio.Queue()
up_processor = QueuedFrameProcessor(received_up) source = QueuedFrameProcessor(received_up)
down_processor = QueuedFrameProcessor(received_down) sink = QueuedFrameProcessor(received_down)
up_processor.link(processor) source.link(processor)
processor.link(down_processor) processor.link(sink)
await processor.queue_frame(StartFrame(clock=SystemClock())) task_manager = TaskManager()
task_manager.set_event_loop(asyncio.get_event_loop())
await source.queue_frame(StartFrame(clock=SystemClock(), task_manager=task_manager))
for frame in frames_to_send: for frame in frames_to_send:
await processor.process_frame(frame, FrameDirection.DOWNSTREAM) await processor.process_frame(frame, FrameDirection.DOWNSTREAM)

2
tests/test_pipeline.py
View File

@@ -57,6 +57,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
async def test_task_single(self): async def test_task_single(self):
pipeline = Pipeline([IdentityFilter()]) pipeline = Pipeline([IdentityFilter()])
task = PipelineTask(pipeline) task = PipelineTask(pipeline)
task.set_event_loop(asyncio.get_event_loop())
await task.queue_frame(TextFrame(text="Hello!")) await task.queue_frame(TextFrame(text="Hello!"))
await task.queue_frames([TextFrame(text="Bye!"), EndFrame()]) await task.queue_frames([TextFrame(text="Bye!"), EndFrame()])
@@ -81,6 +82,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
enable_heartbeats=True, heartbeats_period_secs=0.2, observers=[heartbeats_observer] enable_heartbeats=True, heartbeats_period_secs=0.2, observers=[heartbeats_observer]
), ),
) )
task.set_event_loop(asyncio.get_event_loop())
expected_heartbeats = 1.0 / 0.2 expected_heartbeats = 1.0 / 0.2