| Server IP : 3.147.158.171 / Your IP : 216.73.216.216 Web Server : Apache/2.4.67 (Amazon Linux) OpenSSL/3.5.5 System : Linux ip-172-31-2-178.us-east-2.compute.internal 6.1.172-216.329.amzn2023.x86_64 #1 SMP PREEMPT_DYNAMIC Wed May 20 06:31:34 UTC 2026 x86_64 User : ec2-user ( 1000) PHP Version : 8.4.21 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : OFF Directory : /tsai/repo/api/tests/ |
Upload File : |
"""
Shared pytest fixtures for API tests.
This module provides:
- test_client: FastAPI TestClient instance
- Mock fixtures for OpenAI, Anthropic, and other external services
- Mock site configuration for testing
"""
import pytest
from unittest.mock import MagicMock, AsyncMock, patch
from fastapi.testclient import TestClient
from cachetools import TTLCache
from app.main import app
from app.appTypes import SiteConfig, Message, LLModel, LLMProvider, ModelInfo
@pytest.fixture
def test_client():
"""Create a FastAPI TestClient for testing endpoints."""
return TestClient(app)
@pytest.fixture
def mock_site_config():
"""Create a mock site configuration for testing."""
return SiteConfig(
cb_host="chat-test.example.test",
api_host="api.example.test",
wp_host="https://test.example.test",
wp_dir="/sites/test",
drupal_host="https://drupal.example.test",
drupal_api="/api/v1",
slug="test",
drupal_uid="test-user"
)
@pytest.fixture
def mock_message():
"""Create a basic mock message for testing."""
return Message(
userId=999,
body="Hello, this is a test message",
instructions="You are a helpful assistant.",
conversation=None,
model=LLModel(
provider=LLMProvider.OPEN_AI,
name="gpt-4o-mini",
temperature=0.7
)
)
@pytest.fixture
def mock_message_with_claude():
"""Create a mock message configured to use Claude."""
return Message(
userId=999,
body="Hello, this is a test message for Claude",
instructions="You are Claude, an AI assistant.",
conversation=None,
model=LLModel(
provider=LLMProvider.CLAUDE,
name="claude-sonnet-4-6",
temperature=0.7,
max_tokens=4096
)
)
@pytest.fixture
def mock_openai_models_response():
"""Mock response from OpenAI models.list() endpoint."""
mock_model = MagicMock()
mock_model.id = "gpt-4o"
mock_model.object = "model"
mock_model.created = 1699500000
mock_model.owned_by = "openai"
mock_response = MagicMock()
mock_response.data = [mock_model]
mock_response.model_dump = MagicMock(return_value={
"data": [{"id": "gpt-4o", "object": "model", "created": 1699500000, "owned_by": "openai"}]
})
return mock_response
@pytest.fixture
def mock_openai_stream_chunk():
"""Create a mock OpenAI streaming chunk."""
mock_choice = MagicMock()
mock_choice.delta.content = "Hello"
mock_choice.finish_reason = None
mock_chunk = MagicMock()
mock_chunk.choices = [mock_choice]
return mock_chunk
@pytest.fixture
def mock_openai_stream_end_chunk():
"""Create a mock OpenAI streaming end chunk."""
mock_choice = MagicMock()
mock_choice.delta.content = None
mock_choice.finish_reason = "stop"
mock_chunk = MagicMock()
mock_chunk.choices = [mock_choice]
return mock_chunk
@pytest.fixture
def mock_langchain_models():
"""Mock list of available LangChain models."""
return [
ModelInfo(
provider="OpenAI",
name="gpt-5.5",
description="GPT-5.5 flagship with vision",
max_tokens=8192,
supports_streaming=True,
supports_structured_output=True,
supports_vision=True
),
ModelInfo(
provider="OpenAI",
name="gpt-4o-mini",
description="Cheap, fast vision-capable model used for image analysis",
max_tokens=16384,
supports_streaming=True,
supports_structured_output=True,
supports_vision=True
),
ModelInfo(
provider="Claude",
name="claude-sonnet-4-6",
description="Claude 4.5 Sonnet - Best balance of intelligence, speed, and cost",
max_tokens=8192,
supports_streaming=True,
supports_structured_output=False,
supports_vision=True
)
]
@pytest.fixture
def mock_langchain_chat_model():
"""Create a mock LangChain chat model that supports async streaming."""
mock_model = MagicMock()
# Create an async generator for streaming
async def mock_astream(messages):
"""Mock async stream that yields content chunks."""
chunks = [
MagicMock(content="Hello"),
MagicMock(content=" "),
MagicMock(content="World"),
MagicMock(content="!")
]
for chunk in chunks:
yield chunk
mock_model.astream = mock_astream
return mock_model
@pytest.fixture
def mock_langchain_chat_model_non_streaming():
"""Create a mock LangChain chat model for non-streaming responses."""
mock_model = MagicMock()
async def mock_ainvoke(messages):
"""Mock async invoke that returns a complete response."""
response = MagicMock()
response.content = "This is a complete response from the model."
return response
mock_model.ainvoke = mock_ainvoke
return mock_model
@pytest.fixture
def clear_data_stores():
"""Clear the data stores before each test to ensure isolation."""
# Clear chat endpoint data store
from app.api.v1.endpoints import chat
chat.data_store.clear()
# Clear langchain endpoint data store
from app.api.v1.endpoints import responses_langchain
responses_langchain.data_store.clear()
yield
# Clean up after test
chat.data_store.clear()
responses_langchain.data_store.clear()
@pytest.fixture
def mock_site_config_service(mock_site_config):
"""Create a mock SiteConfigService that returns the mock site config."""
mock_service = MagicMock()
mock_service.get_site_by_cb_host = MagicMock(return_value=mock_site_config)
return mock_service
class MockOpenAIClient:
"""Mock OpenAI client for testing without making real API calls."""
def __init__(self):
self.models = MagicMock()
self.chat = MagicMock()
# Mock models.list()
mock_model = MagicMock()
mock_model.id = "gpt-4o"
mock_model.model_dump = MagicMock(return_value={
"id": "gpt-4o",
"object": "model",
"created": 1699500000,
"owned_by": "openai"
})
mock_models_response = MagicMock()
mock_models_response.data = [mock_model]
mock_models_response.model_dump = MagicMock(return_value={
"data": [mock_model.model_dump()]
})
self.models.list = MagicMock(return_value=mock_models_response)
# Mock chat.completions.create() for streaming
def create_stream(*args, **kwargs):
"""Generate mock streaming chunks."""
chunks = [
self._create_chunk("Hello"),
self._create_chunk(" "),
self._create_chunk("World"),
self._create_chunk("!", finish_reason="stop")
]
return iter(chunks)
self.chat.completions = MagicMock()
self.chat.completions.create = MagicMock(side_effect=create_stream)
def _create_chunk(self, content, finish_reason=None):
"""Helper to create a mock streaming chunk."""
mock_choice = MagicMock()
mock_choice.delta.content = content if finish_reason is None else None
mock_choice.finish_reason = finish_reason
mock_chunk = MagicMock()
mock_chunk.choices = [mock_choice]
return mock_chunk
@pytest.fixture
def mock_openai_client():
"""Provide a MockOpenAIClient instance."""
return MockOpenAIClient()