From 177cb2ca8b72a1a3f8cdf55f5e8d189978c0ab2f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 21 Jan 2025 09:57:54 -0800 Subject: [PATCH] tests: initial pipeline and parallelpipeline tests --- tests/test_pipeline.py | 132 ++++++++++------------------------------- 1 file changed, 30 insertions(+), 102 deletions(-) diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 0acdb034d..dddd1c0de 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -4,119 +4,47 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import asyncio import unittest -from unittest.mock import Mock -from pipecat.frames.frames import EndFrame, TextFrame +from pipecat.frames.frames import TextFrame +from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.pipeline import Pipeline -from pipecat.processors.aggregators.sentence import SentenceAggregator -from pipecat.processors.frame_processor import FrameProcessor -from pipecat.processors.text_transformer import StatelessTextTransformer +from pipecat.processors.filters.identity_filter import IdentityFilter +from tests.utils import run_test -class TestDailyPipeline(unittest.IsolatedAsyncioTestCase): - @unittest.skip("FIXME: This test is failing") - async def test_pipeline_simple(self): - aggregator = SentenceAggregator() +class TestPipeline(unittest.IsolatedAsyncioTestCase): + async def test_pipeline_single(self): + pipeline = Pipeline([IdentityFilter()]) - outgoing_queue = asyncio.Queue() - incoming_queue = asyncio.Queue() - pipeline = Pipeline([aggregator], incoming_queue, outgoing_queue) + frames_to_send = [TextFrame(text="Hello from Pipecat!")] + expected_returned_frames = [TextFrame] + await run_test(pipeline, frames_to_send, expected_returned_frames) - await incoming_queue.put(TextFrame("Hello, ")) - await incoming_queue.put(TextFrame("world.")) - await incoming_queue.put(EndFrame()) + async def test_pipeline_multiple(self): + identity1 = IdentityFilter() + identity2 = IdentityFilter() + identity3 = IdentityFilter() - await pipeline.run_pipeline() + pipeline = Pipeline([identity1, identity2, identity3]) - self.assertEqual(await outgoing_queue.get(), TextFrame("Hello, world.")) - self.assertIsInstance(await outgoing_queue.get(), EndFrame) - - @unittest.skip("FIXME: This test is failing") - async def test_pipeline_multiple_stages(self): - sentence_aggregator = SentenceAggregator() - to_upper = StatelessTextTransformer(lambda x: x.upper()) - add_space = StatelessTextTransformer(lambda x: x + " ") - - outgoing_queue = asyncio.Queue() - incoming_queue = asyncio.Queue() - pipeline = Pipeline( - [add_space, sentence_aggregator, to_upper], incoming_queue, outgoing_queue - ) - - sentence = "Hello, world. It's me, a pipeline." - for c in sentence: - await incoming_queue.put(TextFrame(c)) - await incoming_queue.put(EndFrame()) - - await pipeline.run_pipeline() - - self.assertEqual(await outgoing_queue.get(), TextFrame("H E L L O , W O R L D .")) - self.assertEqual( - await outgoing_queue.get(), - TextFrame(" I T ' S M E , A P I P E L I N E ."), - ) - # leftover little bit because of the spacing - self.assertEqual( - await outgoing_queue.get(), - TextFrame(" "), - ) - self.assertIsInstance(await outgoing_queue.get(), EndFrame) + frames_to_send = [TextFrame(text="Hello from Pipecat!")] + expected_returned_frames = [TextFrame] + await run_test(pipeline, frames_to_send, expected_returned_frames) -class TestLogFrame(unittest.TestCase): - class MockProcessor(FrameProcessor): - def __init__(self, name): - self.name = name +class TestParallelPipeline(unittest.IsolatedAsyncioTestCase): + async def test_parallel_single(self): + pipeline = ParallelPipeline([IdentityFilter()]) - def __str__(self): - return self.name + frames_to_send = [TextFrame(text="Hello from Pipecat!")] + expected_returned_frames = [TextFrame] + await run_test(pipeline, frames_to_send, expected_returned_frames) - def setUp(self): - self.processor1 = self.MockProcessor("processor1") - self.processor2 = self.MockProcessor("processor2") - self.pipeline = Pipeline(processors=[self.processor1, self.processor2]) - self.pipeline._name = "MyClass" - self.pipeline._logger = Mock() + async def test_parallel_multiple(self): + """Should only passthrough one instance of TextFrame.""" + pipeline = ParallelPipeline([IdentityFilter()], [IdentityFilter()]) - @unittest.skip("FIXME: This test is failing") - def test_log_frame_from_source(self): - frame = Mock(__class__=Mock(__name__="MyFrame")) - self.pipeline._log_frame(frame, depth=1) - self.pipeline._logger.debug.assert_called_once_with( - "MyClass source -> MyFrame -> processor1" - ) - - @unittest.skip("FIXME: This test is failing") - def test_log_frame_to_sink(self): - frame = Mock(__class__=Mock(__name__="MyFrame")) - self.pipeline._log_frame(frame, depth=3) - self.pipeline._logger.debug.assert_called_once_with( - "MyClass processor2 -> MyFrame -> sink" - ) - - @unittest.skip("FIXME: This test is failing") - def test_log_frame_repeated_log(self): - frame = Mock(__class__=Mock(__name__="MyFrame")) - self.pipeline._log_frame(frame, depth=2) - self.pipeline._logger.debug.assert_called_once_with( - "MyClass processor1 -> MyFrame -> processor2" - ) - self.pipeline._log_frame(frame, depth=2) - self.pipeline._logger.debug.assert_called_with("MyClass ... repeated") - - @unittest.skip("FIXME: This test is failing") - def test_log_frame_reset_repeated_log(self): - frame1 = Mock(__class__=Mock(__name__="MyFrame1")) - frame2 = Mock(__class__=Mock(__name__="MyFrame2")) - self.pipeline._log_frame(frame1, depth=2) - self.pipeline._logger.debug.assert_called_once_with( - "MyClass processor1 -> MyFrame1 -> processor2" - ) - self.pipeline._log_frame(frame1, depth=2) - self.pipeline._logger.debug.assert_called_with("MyClass ... repeated") - self.pipeline._log_frame(frame2, depth=2) - self.pipeline._logger.debug.assert_called_with( - "MyClass processor1 -> MyFrame2 -> processor2" - ) + frames_to_send = [TextFrame(text="Hello from Pipecat!")] + expected_returned_frames = [TextFrame] + await run_test(pipeline, frames_to_send, expected_returned_frames)