""" 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""" @staticmethod 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}" @staticmethod def generate_session_id() -> str: """Generate a random session ID""" return str(uuid.uuid4()) @staticmethod def generate_message_id() -> str: """Generate a random message ID""" return str(uuid.uuid4()) @staticmethod 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 @staticmethod 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}" @staticmethod 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 @staticmethod 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) } @staticmethod 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)) } @staticmethod 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)) } @staticmethod 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 @staticmethod 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 @staticmethod 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""" @staticmethod 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() } @staticmethod 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() } @staticmethod 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() } @staticmethod 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"] )