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

tests(google): mock the new pipecat.version()

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-12-08 11:44:45 -08:00
parent b0f63c3785
commit d289b38ba7
5 changed files with 23 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/pipecat/services/google/gemini_live/llm.py
View File

@@ -24,7 +24,6 @@ from loguru import logger
from PIL import Image
from pydantic import BaseModel, Field
from pipecat.services.google.utils import update_google_client_http_options
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
from pipecat.frames.frames import (
@@ -69,6 +68,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame, LLMSearchResult
from pipecat.services.google.utils import update_google_client_http_options
from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
from pipecat.services.openai.llm import (
OpenAIAssistantContextAggregator,

5
src/pipecat/services/google/image.py
View File

@@ -22,9 +22,8 @@ from loguru import logger
from PIL import Image
from pydantic import BaseModel, Field
from pipecat.services.google.utils import update_google_client_http_options
from pipecat.frames.frames import ErrorFrame, Frame, URLImageRawFrame
from pipecat.services.google.utils import update_google_client_http_options
from pipecat.services.image_service import ImageGenService
try:
@@ -78,7 +77,7 @@ class GoogleImageGenService(ImageGenService):
# Add client header
http_options = update_google_client_http_options(http_options)
self._client = genai.Client(api_key=api_key, http_options=http_options)
self.set_model_name(self._params.model)

2
src/pipecat/services/google/llm.py
View File

@@ -22,7 +22,6 @@ from loguru import logger
from PIL import Image
from pydantic import BaseModel, Field
from pipecat.services.google.utils import update_google_client_http_options
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter, GeminiLLMInvocationParams
from pipecat.frames.frames import (
AudioRawFrame,
@@ -51,6 +50,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.google.frames import LLMSearchResponseFrame
from pipecat.services.google.utils import update_google_client_http_options
from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
from pipecat.services.openai.llm import (
OpenAIAssistantContextAggregator,

6
src/pipecat/services/google/utils.py
View File

@@ -4,9 +4,11 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Utility functions for Google services."""
from typing import Any, Dict, Optional, Union
from pipecat import __version__ as pipecat_version
from pipecat import version as pipecat_version
def update_google_client_http_options(http_options: Optional[Union[Dict[str, Any], Any]]) -> Any:
@@ -19,7 +21,7 @@ def update_google_client_http_options(http_options: Optional[Union[Dict[str, Any
Returns:
The updated http_options.
"""
client_header = {"x-goog-api-client": f"pipecat/{pipecat_version}"}
client_header = {"x-goog-api-client": f"pipecat/{pipecat_version()}"}
if http_options is None:
http_options = {"headers": client_header}

35
tests/test_google_utils.py
View File

@@ -1,38 +1,33 @@
import importlib.util
import sys
#
# Copyright (c) 2024-2025 Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import unittest
from unittest.mock import MagicMock
# Mock Pipecat package
sys.modules["pipecat"] = MagicMock()
sys.modules["pipecat"].__version__ = "0.0.0-test"
import pipecat.services.google.utils
from pipecat.services.google.utils import update_google_client_http_options
# Load the module directly from source
spec = importlib.util.spec_from_file_location(
"pipecat.services.google.utils", "src/pipecat/services/google/utils.py"
)
utils_module = importlib.util.module_from_spec(spec)
sys.modules["pipecat.services.google.utils"] = utils_module
spec.loader.exec_module(utils_module)
MOCKED_VERSION = "0.0.0-test"
update_google_client_http_options = utils_module.update_google_client_http_options
pipecat_version = "0.0.0-test"
pipecat.services.google.utils.pipecat_version = lambda: MOCKED_VERSION
class TestGoogleUtils(unittest.TestCase):
def test_update_google_client_http_options_none(self):
options = update_google_client_http_options(None)
self.assertEqual(options, {"headers": {"x-goog-api-client": f"pipecat/{pipecat_version}"}})
self.assertEqual(options, {"headers": {"x-goog-api-client": f"pipecat/{MOCKED_VERSION}"}})
def test_update_google_client_http_options_dict_empty(self):
options = update_google_client_http_options({})
self.assertEqual(options, {"headers": {"x-goog-api-client": f"pipecat/{pipecat_version}"}})
self.assertEqual(options, {"headers": {"x-goog-api-client": f"pipecat/{MOCKED_VERSION}"}})
def test_update_google_client_http_options_dict_existing_headers(self):
initial_options = {"headers": {"Authorization": "Bearer token"}}
options = update_google_client_http_options(initial_options)
self.assertEqual(options["headers"]["Authorization"], "Bearer token")
self.assertEqual(options["headers"]["x-goog-api-client"], f"pipecat/{pipecat_version}")
self.assertEqual(options["headers"]["x-goog-api-client"], f"pipecat/{MOCKED_VERSION}")
def test_update_google_client_http_options_object(self):
class HttpOptions:
@@ -42,7 +37,7 @@ class TestGoogleUtils(unittest.TestCase):
http_options = HttpOptions()
updated_options = update_google_client_http_options(http_options)
self.assertEqual(
updated_options.headers, {"x-goog-api-client": f"pipecat/{pipecat_version}"}
updated_options.headers, {"x-goog-api-client": f"pipecat/{MOCKED_VERSION}"}
)
def test_update_google_client_http_options_object_existing_headers(self):
@@ -53,7 +48,7 @@ class TestGoogleUtils(unittest.TestCase):
http_options = HttpOptions()
updated_options = update_google_client_http_options(http_options)
self.assertEqual(updated_options.headers["Authorization"], "Bearer token")
self.assertEqual(updated_options.headers["x-goog-api-client"], f"pipecat/{pipecat_version}")
self.assertEqual(updated_options.headers["x-goog-api-client"], f"pipecat/{MOCKED_VERSION}")
if __name__ == "__main__":