legco_ai_assistant/backend/app/test/test_phase1_llm_client.py

63 lines
2.1 KiB
Python

"""Phase 1 tests: LLM client.
Covers:
- Async HTTP-based LLM client for Qwen LLM
- Provider switching via Settings
- Error handling for API failures
- Mocked responses in test mode
"""
import asyncio
import pytest
import httpx
from unittest.mock import AsyncMock
from app.services.llm_client import LLMClient, LLMClientError
from app.core.config import get_settings
class TestLLMClient:
"""LLM client tests (external calls mocked)."""
@pytest.mark.asyncio
async def test_llm_call_success(self, monkeypatch):
"""Should return content from mocked LLM API."""
settings = get_settings()
client = LLMClient(settings)
# Mock the underlying HTTP response
class _Resp:
status_code = 200
def json(self):
return {
"choices": [{"message": {"content": "mock response"}}]
}
def raise_for_status(self):
pass
async def _mock_post(*args, **kwargs): # type: ignore
return _Resp()
# Patch AsyncClient.post
if hasattr(client, "_client") and client._client is not None:
client._client.post = _mock_post # type: ignore
result = await client.complete(prompt="test prompt", temperature=0.7)
assert isinstance(result, str)
assert "mock" in result
def test_llm_provider_switching(self):
settings = get_settings()
# Ensure base URL comes from settings via client; the client stores base_url
client = LLMClient(settings)
assert settings.llm_base_url.rstrip("/") in client.base_url
@pytest.mark.asyncio
async def test_llm_api_error_handling(self, monkeypatch):
settings = get_settings()
client = LLMClient(settings)
async def _mock_post(*args, **kwargs): # type: ignore
raise httpx.HTTPStatusError("err", request=None, response=None) # type: ignore
client._client.post = _mock_post # type: ignore
with pytest.raises(LLMClientError):
await client.complete(prompt="test", temperature=0.7)