"""
Unit tests for boring.llm.sdk module.
"""
import os
from unittest.mock import MagicMock, patch
import pytest
from boring.llm.sdk import GeminiClient
@pytest.fixture
def temp_project(tmp_path):
project = tmp_path / "project"
project.mkdir()
return project
class TestGeminiClient:
"""Tests for GeminiClient class."""
def test_init_with_api_key(self, temp_project):
"""Test initialization with API key."""
with (
patch("boring.llm.sdk.GENAI_AVAILABLE", True),
patch("boring.llm.sdk.genai") as mock_genai,
patch("boring.llm.sdk.get_boring_tools", return_value=[]),
patch("boring.llm.sdk.log_status"),
):
mock_client = MagicMock()
mock_genai.Client.return_value = mock_client
client = GeminiClient(api_key="test-key", log_dir=temp_project / "logs")
assert client.api_key == "test-key"
assert client.backend == "sdk"
assert client.client == mock_client
def test_init_without_api_key_cli_available(self, temp_project):
"""Test initialization without API key but CLI available."""
with (
patch("boring.llm.sdk.settings") as mock_settings,
patch("boring.llm.sdk.GENAI_AVAILABLE", True),
patch("boring.cli_client.check_cli_available", return_value=True),
patch("boring.cli_client.check_cli_authenticated", return_value=(True, "OK")),
patch("boring.cli_client.GeminiCLIAdapter") as mock_adapter,
patch("boring.llm.sdk.get_boring_tools", return_value=[]),
patch("boring.llm.sdk.log_status"),
patch("boring.llm.sdk.os.environ", {"PATH": os.environ.get("PATH", "")}),
patch("boring.cli.cli_client.shutil.which", return_value="/mock/gemini"),
):
mock_settings.GOOGLE_API_KEY = None
mock_settings.OFFLINE_MODE = False
mock_adapter_instance = MagicMock()
mock_adapter.return_value = mock_adapter_instance
client = GeminiClient(log_dir=temp_project / "logs")
assert client.backend == "cli"
assert client.cli_adapter == mock_adapter_instance
def test_init_genai_not_available(self, temp_project):
"""Test initialization when genai not available."""
with patch("boring.llm.sdk.GENAI_AVAILABLE", False):
with pytest.raises(ImportError, match="google-genai"):
GeminiClient(api_key="test-key", log_dir=temp_project / "logs")
def test_init_no_api_key_no_cli(self, temp_project):
"""Test initialization when no API key and no CLI."""
with (
patch("boring.llm.sdk.GENAI_AVAILABLE", True),
patch("boring.cli_client.check_cli_available", return_value=False),
patch("boring.llm.sdk.settings") as mock_settings,
):
mock_settings.GOOGLE_API_KEY = None
with pytest.raises(ValueError, match="GOOGLE_API_KEY"):
GeminiClient(log_dir=temp_project / "logs")
def test_generate_sdk_success(self, temp_project):
"""Test successful generation with SDK."""
with (
patch("boring.llm.sdk.GENAI_AVAILABLE", True),
patch("boring.llm.sdk.genai") as mock_genai,
patch("boring.llm.sdk.types") as mock_types,
patch("boring.llm.sdk.get_boring_tools", return_value=[]),
patch("boring.llm.sdk.log_status"),
patch("boring.llm.sdk.settings") as mock_settings,
):
# Disable semantic cache for this test
mock_settings.SEMANTIC_CACHE_ENABLED = False
# Mock types.Content and types.Part
mock_content = MagicMock()
mock_types.Content.return_value = mock_content
mock_types.Part.return_value = MagicMock()
mock_client = MagicMock()
mock_response = MagicMock()
mock_response.text = "Generated response"
mock_client.models.generate_content.return_value = mock_response
mock_genai.Client.return_value = mock_client
client = GeminiClient(api_key="test-key", log_dir=temp_project / "logs")
result, success = client.generate("Test prompt")
assert success is True
assert result == "Generated response"
@pytest.mark.skip(reason="Flaky mock integration")
def test_generate_cli_backend(self, temp_project):
"""Test generation with CLI backend."""
with (
patch("boring.llm.sdk.GENAI_AVAILABLE", True),
patch("boring.cli_client.check_cli_available", return_value=True),
patch("boring.cli_client.GeminiCLIAdapter") as mock_adapter_class,
patch("boring.llm.sdk.get_boring_tools", return_value=[]),
patch("boring.llm.sdk.log_status"),
):
mock_adapter = MagicMock()
mock_adapter.generate.return_value = ("Response", True)
mock_adapter_class.return_value = mock_adapter
client = GeminiClient(log_dir=temp_project / "logs")
result, success = client.generate("Test prompt")
assert success is True
assert result == "Response"
def test_generate_model_not_found_fallback(self, temp_project):
"""Test generation with model not found fallback."""
with (
patch("boring.llm.sdk.GENAI_AVAILABLE", True),
patch("boring.llm.sdk.genai") as mock_genai,
patch("boring.llm.sdk.types"),
patch("boring.llm.sdk.get_boring_tools", return_value=[]),
patch("boring.llm.sdk.log_status"),
patch("boring.llm.sdk._logger"),
patch("boring.llm.sdk.settings") as mock_settings,
):
mock_settings.SEMANTIC_CACHE_ENABLED = False
mock_client = MagicMock()
# First call fails with 404, second succeeds
mock_response = MagicMock()
mock_response.text = "Fallback response"
mock_client.models.generate_content.side_effect = [
Exception("404 Model not found"),
mock_response,
]
mock_genai.Client.return_value = mock_client
client = GeminiClient(
api_key="test-key", model_name="invalid-model", log_dir=temp_project / "logs"
)
result, success = client.generate("Test prompt")
assert success is True
assert "Fallback response" in result
def test_generate_rate_limit_error(self, temp_project):
"""Test generation with rate limit error."""
with (
patch("boring.llm.sdk.GENAI_AVAILABLE", True),
patch("boring.llm.sdk.genai") as mock_genai,
patch("boring.llm.sdk.types"),
patch("boring.llm.sdk.get_boring_tools", return_value=[]),
patch("boring.llm.sdk.log_status"),
patch("boring.llm.sdk.settings") as mock_settings,
):
mock_settings.SEMANTIC_CACHE_ENABLED = False
mock_client = MagicMock()
mock_client.models.generate_content.side_effect = Exception("429 Rate limit exceeded")
mock_genai.Client.return_value = mock_client
client = GeminiClient(api_key="test-key", log_dir=temp_project / "logs")
# Mock _get_semantic_cache locally to avoid import issues
with patch.object(client, "_get_semantic_cache", return_value=None):
result, success = client.generate("Test prompt")
assert success is False
assert "RATE_LIMIT_ERROR" in result
def test_generate_timeout_error(self, temp_project):
"""Test generation with timeout error."""
with (
patch("boring.llm.sdk.GENAI_AVAILABLE", True),
patch("boring.llm.sdk.genai") as mock_genai,
patch("boring.llm.sdk.types"),
patch("boring.llm.sdk.get_boring_tools", return_value=[]),
patch("boring.llm.sdk.log_status"),
patch("boring.llm.sdk.settings") as mock_settings,
):
mock_settings.SEMANTIC_CACHE_ENABLED = False
mock_client = MagicMock()
mock_client.models.generate_content.side_effect = Exception("Deadline exceeded")
mock_genai.Client.return_value = mock_client
client = GeminiClient(api_key="test-key", log_dir=temp_project / "logs")
# Mock _get_semantic_cache locally
with patch.object(client, "_get_semantic_cache", return_value=None):
result, success = client.generate("Test prompt")
assert success is False
assert "TIMEOUT_ERROR" in result
def test_generate_empty_response(self, temp_project):
"""Test generation with empty response."""
with (
patch("boring.llm.sdk.GENAI_AVAILABLE", True),
patch("boring.llm.sdk.genai") as mock_genai,
patch("boring.llm.sdk.types"),
patch("boring.llm.sdk.get_boring_tools", return_value=[]),
patch("boring.llm.sdk.log_status"),
patch("boring.llm.sdk.settings") as mock_settings,
):
mock_settings.SEMANTIC_CACHE_ENABLED = False
mock_client = MagicMock()
mock_response = MagicMock()
mock_response.text = None
mock_client.models.generate_content.return_value = mock_response
mock_genai.Client.return_value = mock_client
client = GeminiClient(api_key="test-key", log_dir=temp_project / "logs")
# Mock _get_semantic_cache locally
with patch.object(client, "_get_semantic_cache", return_value=None):
result, success = client.generate("Test prompt")
assert success is False
assert result == ""
def test_generate_with_retry_success(self, temp_project):
"""Test generate_with_retry with success."""
with (
patch("boring.llm.sdk.GENAI_AVAILABLE", True),
patch("boring.llm.sdk.genai") as mock_genai,
patch("boring.llm.sdk.settings") as mock_settings,
patch("boring.llm.sdk.types"),
patch("boring.llm.sdk.get_boring_tools", return_value=[]),
patch("boring.llm.sdk.log_status"),
):
mock_settings.SEMANTIC_CACHE_ENABLED = False
mock_client = MagicMock()
mock_response = MagicMock()
mock_response.text = "Response"
mock_client.models.generate_content.return_value = mock_response
mock_genai.Client.return_value = mock_client
client = GeminiClient(api_key="test-key", log_dir=temp_project / "logs")
# Mock _get_semantic_cache locally
with patch.object(client, "_get_semantic_cache", return_value=None):
result, success = client.generate_with_retry("Test prompt")
assert success is True
assert "Response" in result
def test_generate_with_retry_rate_limit(self, temp_project):
"""Test generate_with_retry with rate limit retry."""
with (
patch("boring.llm.sdk.GENAI_AVAILABLE", True),
patch("boring.llm.sdk.genai") as mock_genai,
patch("boring.llm.sdk.types"),
patch("boring.llm.sdk.get_boring_tools", return_value=[]),
patch("boring.llm.sdk.log_status"),
patch("time.sleep"),
patch("boring.llm.sdk.settings") as mock_settings,
):
mock_settings.SEMANTIC_CACHE_ENABLED = False
mock_client = MagicMock()
# First call fails with rate limit, second succeeds
mock_response = MagicMock()
mock_response.text = "Success after retry"
mock_client.models.generate_content.side_effect = [
Exception("429 Rate limit"),
mock_response,
]
mock_genai.Client.return_value = mock_client
client = GeminiClient(api_key="test-key", log_dir=temp_project / "logs")
# Mock _get_semantic_cache locally
with patch.object(client, "_get_semantic_cache", return_value=None):
result, success = client.generate_with_retry("Test prompt", max_retries=3)
assert success is True
def test_generate_with_tools_success(self, temp_project):
"""Test generate_with_tools successfully."""
with (
patch("boring.llm.sdk.GENAI_AVAILABLE", True),
patch("boring.llm.sdk.genai") as mock_genai,
patch("boring.llm.sdk.types"),
patch("boring.llm.sdk.get_boring_tools", return_value=[{"name": "test_tool"}]),
patch("boring.llm.sdk.log_status"),
):
mock_client = MagicMock()
mock_response = MagicMock()
mock_part = MagicMock()
mock_part.text = "Response"
mock_part.function_call = None
mock_candidate = MagicMock()
mock_candidate.content.parts = [mock_part]
mock_response.candidates = [mock_candidate]
mock_client.models.generate_content.return_value = mock_response
mock_genai.Client.return_value = mock_client
client = GeminiClient(api_key="test-key", log_dir=temp_project / "logs")
with patch.object(client, "_get_semantic_cache", return_value=None):
text, calls, success = client.generate_with_tools("Test prompt")
assert success is True
assert "Response" in text
@pytest.mark.skip(reason="Flaky mock integration")
def test_generate_with_tools_cli_backend(self, temp_project):
"""Test generate_with_tools with CLI backend."""
with (
patch("boring.llm.sdk.GENAI_AVAILABLE", True),
patch("boring.cli_client.check_cli_available", return_value=True),
patch("boring.cli_client.GeminiCLIAdapter") as mock_adapter_class,
patch("boring.llm.sdk.get_boring_tools", return_value=[]),
patch("boring.llm.sdk.log_status"),
):
mock_adapter = MagicMock()
from types import SimpleNamespace
mock_response = SimpleNamespace()
mock_response.text = "Response"
mock_response.function_calls = [{"name": "test", "args": {}}]
mock_response.success = True
mock_adapter.generate_with_tools.return_value = mock_response
mock_adapter_class.return_value = mock_adapter
client = GeminiClient(log_dir=temp_project / "logs")
text, calls, success = client.generate_with_tools("Test prompt")
print(f"DEBUG: Adapter: {client.cli_adapter}")
print(f"DEBUG: Calls: {calls}")
print(f"DEBUG: Success: {success}")
print(f"DEBUG: Mock Adapter Return: {mock_adapter.generate_with_tools.return_value}")
print(f"DEBUG: Mock Response Func Calls: {mock_response.function_calls}")
assert success is True
assert len(calls) > 0
