Spaces:
Sleeping
Sleeping
| """ | |
| Mock data generators for the Atlas AI Chat API test suite | |
| """ | |
| from datetime import datetime, timedelta | |
| import string | |
| from typing import Any, Dict, List, Optional | |
| import uuid | |
| import random | |
| class MockDataGenerator: | |
| """Generator for mock test data""" | |
| def generate_user_id(prefix: str = "test_user") -> str: | |
| """Generate a random test user ID""" | |
| suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=8)) | |
| return f"{prefix}_{suffix}" | |
| def generate_session_id() -> str: | |
| """Generate a random session ID""" | |
| return str(uuid.uuid4()) | |
| def generate_message_id() -> str: | |
| """Generate a random message ID""" | |
| return str(uuid.uuid4()) | |
| def generate_chat_request(user_id: Optional[str] = None) -> Dict[str, Any]: | |
| """Generate a mock chat request""" | |
| prompts = [ | |
| "What is the weather like today?", | |
| "Tell me about artificial intelligence", | |
| "How do I cook pasta?", | |
| "What are the benefits of exercise?", | |
| "Explain quantum computing", | |
| "What is the capital of France?", | |
| "How does photosynthesis work?", | |
| "What are the latest technology trends?" | |
| ] | |
| request = { | |
| "prompt": random.choice(prompts), | |
| "max_new_tokens": random.randint(50, 500), | |
| "use_search": random.choice([True, False]), | |
| "temperature": round(random.uniform(0.1, 1.0), 1) | |
| } | |
| if user_id is not None: | |
| request["user_id"] = user_id | |
| return request | |
| def generate_ai_response() -> str: | |
| """Generate a mock AI response""" | |
| responses = [ | |
| "This is a helpful AI response about your question.", | |
| "Based on the information available, here's what I can tell you...", | |
| "That's an interesting question! Let me explain...", | |
| "Here's a comprehensive answer to your query...", | |
| "I'd be happy to help you understand this topic better.", | |
| "This is a detailed response with relevant information.", | |
| "Let me provide you with accurate information about this.", | |
| "Here's what you need to know about this subject." | |
| ] | |
| base_response = random.choice(responses) | |
| # Add some random additional content | |
| additional_content = " ".join([ | |
| "Additional information here.", | |
| "More details about the topic.", | |
| "Further explanation follows.", | |
| "This provides comprehensive coverage." | |
| ][:random.randint(1, 4)]) | |
| return f"{base_response} {additional_content}" | |
| def generate_search_results(count: int = 5) -> List[Dict[str, str]]: | |
| """Generate mock search results""" | |
| domains = ["example.com", "test.org", "sample.net", "demo.edu", "info.gov"] | |
| topics = ["technology", "science", "health", "education", "business"] | |
| results = [] | |
| for i in range(count): | |
| domain = random.choice(domains) | |
| topic = random.choice(topics) | |
| result = { | |
| "title": f"Test Result {i+1}: {topic.title()} Information", | |
| "url": f"https://{domain}/{topic}/article-{i+1}", | |
| "snippet": f"This is a mock search result snippet about {topic}. It contains relevant information for testing purposes." | |
| } | |
| results.append(result) | |
| return results | |
| def generate_session_data(user_id: Optional[str] = None) -> Dict[str, Any]: | |
| """Generate mock session data""" | |
| user_agents = [ | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", | |
| "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36", | |
| "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36", | |
| "TestAgent/1.0", | |
| "TestClient/2.0" | |
| ] | |
| return { | |
| "session_id": MockDataGenerator.generate_session_id(), | |
| "user_id": user_id, | |
| "user_agent": random.choice(user_agents), | |
| "start_time": datetime.utcnow() - timedelta(minutes=random.randint(1, 60)), | |
| "status": random.choice(["active", "ended"]), | |
| "message_count": random.randint(1, 10) | |
| } | |
| def generate_message_data(session_id: str, user_id: Optional[str] = None) -> Dict[str, Any]: | |
| """Generate mock message data""" | |
| return { | |
| "message_id": MockDataGenerator.generate_message_id(), | |
| "session_id": session_id, | |
| "user_id": user_id, | |
| "prompt_length": random.randint(10, 200), | |
| "response_length": random.randint(50, 500), | |
| "response_time_ms": random.randint(500, 5000), | |
| "used_search": random.choice([True, False]), | |
| "max_tokens": random.randint(100, 1000), | |
| "temperature": round(random.uniform(0.1, 1.0), 1), | |
| "success": random.choice([True, True, True, False]), # 75% success rate | |
| "timestamp": datetime.utcnow() - timedelta(minutes=random.randint(1, 30)) | |
| } | |
| def generate_search_analytics_data(message_id: str, user_id: Optional[str] = None) -> Dict[str, Any]: | |
| """Generate mock search analytics data""" | |
| search_engines = ["brave", "duckduckgo"] | |
| used_engines = random.sample(search_engines, random.randint(1, 2)) | |
| return { | |
| "message_id": message_id, | |
| "user_id": user_id, | |
| "search_query": "test search query", | |
| "search_terms": ["test", "search", "query"], | |
| "brave_results": random.randint(0, 10) if "brave" in used_engines else 0, | |
| "duckduckgo_results": random.randint(0, 10) if "duckduckgo" in used_engines else 0, | |
| "total_unique_results": random.randint(5, 15), | |
| "brave_response_time_ms": random.randint(500, 2000) if "brave" in used_engines else 0, | |
| "duckduckgo_response_time_ms": random.randint(500, 2000) if "duckduckgo" in used_engines else 0, | |
| "search_engines_used": used_engines, | |
| "search_success": random.choice([True, True, False]), # 67% success rate | |
| "fallback_used": random.choice([True, False]), | |
| "timestamp": datetime.utcnow() - timedelta(minutes=random.randint(1, 30)) | |
| } | |
| def generate_bulk_test_data( | |
| num_users: int = 5, | |
| sessions_per_user: int = 2, | |
| messages_per_session: int = 3, | |
| include_anonymous: bool = True | |
| ) -> Dict[str, List[Dict[str, Any]]]: | |
| """Generate bulk test data for performance testing""" | |
| data = { | |
| "users": [], | |
| "sessions": [], | |
| "messages": [], | |
| "search_analytics": [] | |
| } | |
| # Generate authenticated users | |
| for i in range(num_users): | |
| user_id = MockDataGenerator.generate_user_id(f"bulk_user_{i}") | |
| data["users"].append(user_id) | |
| for j in range(sessions_per_user): | |
| session_data = MockDataGenerator.generate_session_data(user_id) | |
| data["sessions"].append(session_data) | |
| for k in range(messages_per_session): | |
| message_data = MockDataGenerator.generate_message_data( | |
| session_data["session_id"], user_id | |
| ) | |
| data["messages"].append(message_data) | |
| # 50% chance of search analytics | |
| if random.choice([True, False]): | |
| search_data = MockDataGenerator.generate_search_analytics_data( | |
| message_data["message_id"], user_id | |
| ) | |
| data["search_analytics"].append(search_data) | |
| # Generate anonymous users if requested | |
| if include_anonymous: | |
| anonymous_sessions = num_users // 2 # Half as many anonymous sessions | |
| for i in range(anonymous_sessions): | |
| session_data = MockDataGenerator.generate_session_data(None) | |
| data["sessions"].append(session_data) | |
| for j in range(messages_per_session): | |
| message_data = MockDataGenerator.generate_message_data( | |
| session_data["session_id"], None | |
| ) | |
| data["messages"].append(message_data) | |
| # 30% chance of search analytics for anonymous users | |
| if random.random() < 0.3: | |
| search_data = MockDataGenerator.generate_search_analytics_data( | |
| message_data["message_id"], None | |
| ) | |
| data["search_analytics"].append(search_data) | |
| return data | |
| def generate_time_series_data( | |
| user_id: Optional[str] = None, | |
| hours_back: int = 24, | |
| messages_per_hour: int = 2 | |
| ) -> List[Dict[str, Any]]: | |
| """Generate time series message data for analytics testing""" | |
| messages = [] | |
| base_time = datetime.utcnow() | |
| for hour in range(hours_back): | |
| timestamp = base_time - timedelta(hours=hour) | |
| for i in range(random.randint(0, messages_per_hour * 2)): # Vary the count | |
| message_data = MockDataGenerator.generate_message_data( | |
| MockDataGenerator.generate_session_id(), user_id | |
| ) | |
| message_data["timestamp"] = timestamp + timedelta( | |
| minutes=random.randint(0, 59), | |
| seconds=random.randint(0, 59) | |
| ) | |
| messages.append(message_data) | |
| return messages | |
| def generate_performance_test_scenarios() -> List[Dict[str, Any]]: | |
| """Generate scenarios for performance testing""" | |
| scenarios = [ | |
| { | |
| "name": "light_load", | |
| "concurrent_users": 5, | |
| "requests_per_user": 3, | |
| "delay_between_requests": 1.0 | |
| }, | |
| { | |
| "name": "medium_load", | |
| "concurrent_users": 10, | |
| "requests_per_user": 5, | |
| "delay_between_requests": 0.5 | |
| }, | |
| { | |
| "name": "heavy_load", | |
| "concurrent_users": 20, | |
| "requests_per_user": 10, | |
| "delay_between_requests": 0.1 | |
| }, | |
| { | |
| "name": "burst_load", | |
| "concurrent_users": 50, | |
| "requests_per_user": 2, | |
| "delay_between_requests": 0.0 | |
| } | |
| ] | |
| return scenarios | |
| class MockAPIResponses: | |
| """Mock API responses for testing""" | |
| def successful_chat_response(session_id: str = None) -> Dict[str, Any]: | |
| """Generate a successful chat API response""" | |
| if session_id is None: | |
| session_id = MockDataGenerator.generate_session_id() | |
| return { | |
| "response": MockDataGenerator.generate_ai_response(), | |
| "session_id": session_id, | |
| "timestamp": datetime.utcnow().isoformat() | |
| } | |
| def error_chat_response(error_message: str = "Invalid request") -> Dict[str, Any]: | |
| """Generate an error chat API response""" | |
| return { | |
| "detail": error_message, | |
| "timestamp": datetime.utcnow().isoformat() | |
| } | |
| def analytics_stats_response() -> Dict[str, Any]: | |
| """Generate a mock analytics stats response""" | |
| return { | |
| "total_sessions": random.randint(100, 1000), | |
| "total_messages": random.randint(500, 5000), | |
| "active_sessions": random.randint(10, 100), | |
| "messages_today": random.randint(50, 500), | |
| "search_usage_percentage": round(random.uniform(20, 80), 1), | |
| "average_response_time_ms": random.randint(800, 2000), | |
| "success_rate_percentage": round(random.uniform(85, 99), 1), | |
| "last_updated": datetime.utcnow().isoformat() | |
| } | |
| def user_analytics_response(user_id: str) -> Dict[str, Any]: | |
| """Generate a mock user analytics response""" | |
| return { | |
| "user_id": user_id, | |
| "total_sessions": random.randint(1, 50), | |
| "active_sessions": random.randint(0, 5), | |
| "total_messages": random.randint(5, 200), | |
| "messages_with_search": random.randint(1, 100), | |
| "search_usage_percentage": round(random.uniform(10, 90), 1), | |
| "avg_response_time_ms": random.randint(800, 2000), | |
| "avg_messages_per_session": round(random.uniform(2, 10), 1), | |
| "daily_activity_last_30_days": [ | |
| { | |
| "date": (datetime.utcnow() - timedelta(days=i)).strftime("%Y-%m-%d"), | |
| "message_count": random.randint(0, 20) | |
| } | |
| for i in range(30) | |
| ], | |
| "last_updated": datetime.utcnow().isoformat() | |
| } | |
| # Convenience functions for common mock data patterns | |
| def create_mock_authenticated_user() -> str: | |
| """Create a mock authenticated user ID""" | |
| return MockDataGenerator.generate_user_id("mock_auth_user") | |
| def create_mock_chat_session(user_id: Optional[str] = None) -> Dict[str, Any]: | |
| """Create a complete mock chat session with messages""" | |
| session_data = MockDataGenerator.generate_session_data(user_id) | |
| messages = [] | |
| for i in range(random.randint(1, 5)): | |
| message_data = MockDataGenerator.generate_message_data( | |
| session_data["session_id"], user_id | |
| ) | |
| messages.append(message_data) | |
| return { | |
| "session": session_data, | |
| "messages": messages | |
| } | |
| def create_mock_analytics_dataset(size: str = "small") -> Dict[str, Any]: | |
| """Create a mock analytics dataset of specified size""" | |
| size_configs = { | |
| "small": {"users": 5, "sessions": 2, "messages": 3}, | |
| "medium": {"users": 20, "sessions": 3, "messages": 5}, | |
| "large": {"users": 100, "sessions": 5, "messages": 10} | |
| } | |
| config = size_configs.get(size, size_configs["small"]) | |
| return MockDataGenerator.generate_bulk_test_data( | |
| num_users=config["users"], | |
| sessions_per_user=config["sessions"], | |
| messages_per_session=config["messages"] | |
| ) |