From d1550d5a854ca1617068a67d368e28bfedb00b5d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 18 Mar 2025 15:52:37 -0700 Subject: [PATCH] tests: remove TestFrameProcessor, reimplement with run_test() --- src/pipecat/tests/utils.py | 42 +++--- src/pipecat/utils/test_frame_processor.py | 48 ------- tests/integration/integration_azure_llm.py | 33 ----- tests/integration/integration_ollama_llm.py | 28 ---- tests/integration/integration_openai_llm.py | 128 ------------------ ...st_integration_unified_function_calling.py | 27 ++-- 6 files changed, 37 insertions(+), 269 deletions(-) delete mode 100644 src/pipecat/utils/test_frame_processor.py delete mode 100644 tests/integration/integration_azure_llm.py delete mode 100644 tests/integration/integration_ollama_llm.py delete mode 100644 tests/integration/integration_openai_llm.py diff --git a/src/pipecat/tests/utils.py b/src/pipecat/tests/utils.py index d68c87d88..e2368ba09 100644 --- a/src/pipecat/tests/utils.py +++ b/src/pipecat/tests/utils.py @@ -6,7 +6,7 @@ import asyncio from dataclasses import dataclass -from typing import Any, Awaitable, Callable, Dict, Sequence, Tuple +from typing import Any, Awaitable, Callable, Dict, Optional, Sequence, Tuple from pipecat.frames.frames import ( EndFrame, @@ -80,8 +80,8 @@ async def run_test( processor: FrameProcessor, *, frames_to_send: Sequence[Frame], - expected_down_frames: Sequence[type], - expected_up_frames: Sequence[type] = [], + expected_down_frames: Optional[Sequence[type]] = None, + expected_up_frames: Optional[Sequence[type]] = None, ignore_start: bool = True, start_metadata: Dict[str, Any] = {}, send_end_frame: bool = True, @@ -126,33 +126,35 @@ async def run_test( # Down frames # received_down_frames: Sequence[Frame] = [] - while not received_down.empty(): - frame = await received_down.get() - if not isinstance(frame, EndFrame) or not send_end_frame: - received_down_frames.append(frame) + if expected_down_frames is not None: + while not received_down.empty(): + frame = await received_down.get() + if not isinstance(frame, EndFrame) or not send_end_frame: + received_down_frames.append(frame) - print("received DOWN frames =", received_down_frames) - print("expected DOWN frames =", expected_down_frames) + print("received DOWN frames =", received_down_frames) + print("expected DOWN frames =", expected_down_frames) - assert len(received_down_frames) == len(expected_down_frames) + assert len(received_down_frames) == len(expected_down_frames) - for real, expected in zip(received_down_frames, expected_down_frames): - assert isinstance(real, expected) + for real, expected in zip(received_down_frames, expected_down_frames): + assert isinstance(real, expected) # # Up frames # received_up_frames: Sequence[Frame] = [] - while not received_up.empty(): - frame = await received_up.get() - received_up_frames.append(frame) + if expected_up_frames is not None: + while not received_up.empty(): + frame = await received_up.get() + received_up_frames.append(frame) - print("received UP frames =", received_up_frames) - print("expected UP frames =", expected_up_frames) + print("received UP frames =", received_up_frames) + print("expected UP frames =", expected_up_frames) - assert len(received_up_frames) == len(expected_up_frames) + assert len(received_up_frames) == len(expected_up_frames) - for real, expected in zip(received_up_frames, expected_up_frames): - assert isinstance(real, expected) + for real, expected in zip(received_up_frames, expected_up_frames): + assert isinstance(real, expected) return (received_down_frames, received_up_frames) diff --git a/src/pipecat/utils/test_frame_processor.py b/src/pipecat/utils/test_frame_processor.py deleted file mode 100644 index b35864497..000000000 --- a/src/pipecat/utils/test_frame_processor.py +++ /dev/null @@ -1,48 +0,0 @@ -from typing import List - -from pipecat.processors.frame_processor import FrameProcessor - - -class TestException(Exception): - pass - - -class TestFrameProcessor(FrameProcessor): - __test__ = False # Prevents pytest from collecting this class as a test - - def __init__(self, test_frames): - self.test_frames = test_frames - self._list_counter = 0 - super().__init__() - - async def process_frame(self, frame, direction): - await super().process_frame(frame, direction) - - if not self.test_frames[ - 0 - ]: # then we've run out of required frames but the generator is still going? - raise TestException(f"Oops, got an extra frame, {frame}") - if isinstance(self.test_frames[0], List): - # We need to consume frames until we see the next frame type after this - next_frame = self.test_frames[1] - if isinstance(frame, next_frame): - # we're done iterating the list I guess - print(f"TestFrameProcessor got expected list exit frame: {frame}") - # pop twice to get rid of the list, as well as the next frame - self.test_frames.pop(0) - self.test_frames.pop(0) - self.list_counter = 0 - else: - fl = self.test_frames[0] - fl_el = fl[self._list_counter % len(fl)] - if isinstance(frame, fl_el): - print(f"TestFrameProcessor got expected list frame: {frame}") - self._list_counter += 1 - else: - raise TestException(f"Inside a list, expected {fl_el} but got {frame}") - - else: - if not isinstance(frame, self.test_frames[0]): - raise TestException(f"Expected {self.test_frames[0]}, but got {frame}") - print(f"TestFrameProcessor got expected frame: {frame}") - self.test_frames.pop(0) diff --git a/tests/integration/integration_azure_llm.py b/tests/integration/integration_azure_llm.py deleted file mode 100644 index 8e49e9d04..000000000 --- a/tests/integration/integration_azure_llm.py +++ /dev/null @@ -1,33 +0,0 @@ -import asyncio -import os -import unittest - -from openai.types.chat import ( - ChatCompletionSystemMessageParam, -) - -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContext, - OpenAILLMContextFrame, -) -from pipecat.services.azure import AzureLLMService - -if __name__ == "__main__": - - @unittest.skip("Skip azure integration test") - async def test_chat(): - llm = AzureLLMService( - api_key=os.getenv("AZURE_CHATGPT_API_KEY"), - endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), - model=os.getenv("AZURE_CHATGPT_MODEL"), - ) - context = OpenAILLMContext() - message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam( - content="Please tell the world hello.", name="system", role="system" - ) - context.add_message(message) - frame = OpenAILLMContextFrame(context) - async for s in llm.process_frame(frame): - print(s) - - asyncio.run(test_chat()) diff --git a/tests/integration/integration_ollama_llm.py b/tests/integration/integration_ollama_llm.py deleted file mode 100644 index 085500cb8..000000000 --- a/tests/integration/integration_ollama_llm.py +++ /dev/null @@ -1,28 +0,0 @@ -import asyncio -import unittest - -from openai.types.chat import ( - ChatCompletionSystemMessageParam, -) - -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContext, - OpenAILLMContextFrame, -) -from pipecat.services.ollama import OLLamaLLMService - -if __name__ == "__main__": - - @unittest.skip("Skip azure integration test") - async def test_chat(): - llm = OLLamaLLMService() - context = OpenAILLMContext() - message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam( - content="Please tell the world hello.", name="system", role="system" - ) - context.add_message(message) - frame = OpenAILLMContextFrame(context) - async for s in llm.process_frame(frame): - print(s) - - asyncio.run(test_chat()) diff --git a/tests/integration/integration_openai_llm.py b/tests/integration/integration_openai_llm.py deleted file mode 100644 index 2c84c8882..000000000 --- a/tests/integration/integration_openai_llm.py +++ /dev/null @@ -1,128 +0,0 @@ -import asyncio -import json -import os -from typing import List - -from openai.types.chat import ( - ChatCompletionSystemMessageParam, - ChatCompletionToolParam, - ChatCompletionUserMessageParam, -) - -from pipecat.frames.frames import LLMFullResponseEndFrame, LLMFullResponseStartFrame, TextFrame -from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame, OpenAILLMService -from pipecat.utils.test_frame_processor import TestFrameProcessor - -tools = [ - ChatCompletionToolParam( - type="function", - function={ - "name": "get_current_weather", - "description": "Get the current weather", - "parameters": { - "type": "object", - "properties": { - "location": { - "type": "string", - "description": "The city and state, e.g. San Francisco, CA", - }, - "format": { - "type": "string", - "enum": ["celsius", "fahrenheit"], - "description": "The temperature unit to use. Infer this from the users location.", - }, - }, - "required": ["location", "format"], - }, - }, - ) -] - -if __name__ == "__main__": - - async def test_simple_functions(): - async def get_weather_from_api(llm, args): - return json.dumps({"conditions": "nice", "temperature": "75"}) - - api_key = os.getenv("OPENAI_API_KEY") - - llm = OpenAILLMService( - api_key=api_key or "", - model="gpt-4-1106-preview", - ) - - llm.register_function("get_current_weather", get_weather_from_api) - t = TestFrameProcessor([LLMFullResponseStartFrame, TextFrame, LLMFullResponseEndFrame]) - llm.link(t) - - context = OpenAILLMContext(tools=tools) - system_message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam( - content="Ask the user to ask for a weather report", name="system", role="system" - ) - user_message: ChatCompletionUserMessageParam = ChatCompletionUserMessageParam( - content="Could you tell me the weather for Boulder, Colorado", - name="user", - role="user", - ) - context.add_message(system_message) - context.add_message(user_message) - frame = OpenAILLMContextFrame(context) - await llm.process_frame(frame, FrameDirection.DOWNSTREAM) - - async def test_advanced_functions(): - async def get_weather_from_api(llm, args): - return [ - { - "role": "system", - "content": "The user has asked for live weather. Respond by telling them we don't currently support live weather for that area, but it's coming soon.", - } - ] - - api_key = os.getenv("OPENAI_API_KEY") - - llm = OpenAILLMService( - api_key=api_key or "", - model="gpt-4-1106-preview", - ) - - llm.register_function("get_current_weather", get_weather_from_api) - t = TestFrameProcessor([LLMFullResponseStartFrame, TextFrame, LLMFullResponseEndFrame]) - llm.link(t) - - context = OpenAILLMContext(tools=tools) - system_message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam( - content="Ask the user to ask for a weather report", name="system", role="system" - ) - user_message: ChatCompletionUserMessageParam = ChatCompletionUserMessageParam( - content="Could you tell me the weather for Boulder, Colorado", - name="user", - role="user", - ) - context.add_message(system_message) - context.add_message(user_message) - frame = OpenAILLMContextFrame(context) - await llm.process_frame(frame, FrameDirection.DOWNSTREAM) - - async def test_chat(): - api_key = os.getenv("OPENAI_API_KEY") - t = TestFrameProcessor([LLMFullResponseStartFrame, TextFrame, LLMFullResponseEndFrame]) - llm = OpenAILLMService( - api_key=api_key or "", - model="gpt-4o", - ) - llm.link(t) - context = OpenAILLMContext() - message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam( - content="Please tell the world hello.", name="system", role="system" - ) - context.add_message(message) - frame = OpenAILLMContextFrame(context) - await llm.process_frame(frame, FrameDirection.DOWNSTREAM) - - async def run_tests(): - await test_simple_functions() - await test_advanced_functions() - await test_chat() - - asyncio.run(run_tests()) diff --git a/tests/integration/test_integration_unified_function_calling.py b/tests/integration/test_integration_unified_function_calling.py index 88407d703..0696c531a 100644 --- a/tests/integration/test_integration_unified_function_calling.py +++ b/tests/integration/test_integration_unified_function_calling.py @@ -1,3 +1,9 @@ +# +# Copyright (c) 2024-2025 Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + import os from unittest.mock import AsyncMock @@ -6,17 +12,12 @@ from dotenv import load_dotenv from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema -from pipecat.frames.frames import ( - LLMFullResponseEndFrame, - LLMFullResponseStartFrame, - LLMTextFrame, -) -from pipecat.processors.frame_processor import FrameDirection +from pipecat.pipeline.pipeline import Pipeline from pipecat.services.ai_services import LLMService from pipecat.services.anthropic import AnthropicLLMService from pipecat.services.google import GoogleLLMService from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame, OpenAILLMService -from pipecat.utils.test_frame_processor import TestFrameProcessor +from pipecat.tests.utils import run_test load_dotenv(override=True) @@ -47,8 +48,6 @@ async def _test_llm_function_calling(llm: LLMService): mock_fetch_weather = AsyncMock() llm.register_function(None, mock_fetch_weather) - t = TestFrameProcessor([LLMFullResponseStartFrame, LLMTextFrame, LLMFullResponseEndFrame]) - llm.link(t) messages = [ { @@ -61,10 +60,14 @@ async def _test_llm_function_calling(llm: LLMService): # This is done by default inside the create_context_aggregator context.set_llm_adapter(llm.get_llm_adapter()) - frame = OpenAILLMContextFrame(context) + pipeline = Pipeline([llm]) - # This will fail if an exception is raised - await llm.process_frame(frame, FrameDirection.DOWNSTREAM) + frames_to_send = [OpenAILLMContextFrame(context)] + await run_test( + pipeline, + frames_to_send=frames_to_send, + expected_down_frames=None, + ) # Assert that the mock function was called mock_fetch_weather.assert_called_once()