""" Integration tests for chat API with user authentication Consolidated from: - test_chat_integration_user_auth.py - test_chat_analytics.py (integration portions) - Parts of test_user_authentication_comprehensive.py (integration portions) """ import asyncio import time import pytest from tests.utilities import HTTPHelpers, MockHelpers, TestHelpers, ValidationHelpers, authenticated_chat_request, empty_user_ids, invalid_user_ids, sample_chat_request, skip_if_no_server, valid_user_ids HTTPHelpers, TestHelpers, MockHelpers, ValidationHelpers, skip_if_no_server, authenticated_chat_request, sample_chat_request, valid_user_ids, invalid_user_ids, empty_user_ids ) class TestChatRequestValidation: """Test chat request validation with user_id""" @pytest.mark.asyncio @skip_if_no_server() async def test_valid_user_id_formats(self, valid_user_ids): """Test chat requests with various valid user_id formats""" for user_id in valid_user_ids[:5]: # Test first 5 to avoid too many requests chat_data = { "prompt": f"Test message for user {user_id}", "max_new_tokens": 50, "use_search": False, "temperature": 0.7, "user_id": user_id } response = await HTTPHelpers.make_chat_request(chat_data) if response is None: pytest.skip("Server not available") session_id = HTTPHelpers.assert_successful_chat_response(response) assert session_id is not None @pytest.mark.asyncio @skip_if_no_server() async def test_invalid_user_id_formats(self, invalid_user_ids): """Test chat requests with invalid user_id formats""" for user_id in invalid_user_ids[:10]: # Test first 10 to avoid too many requests chat_data = { "prompt": f"Test message for invalid user {user_id}", "max_new_tokens": 50, "use_search": False, "temperature": 0.7, "user_id": user_id } response = await HTTPHelpers.make_chat_request(chat_data) if response is None: pytest.skip("Server not available") HTTPHelpers.assert_failed_chat_response(response, expected_status=400) @pytest.mark.asyncio @skip_if_no_server() async def test_empty_user_id_handling(self, empty_user_ids): """Test that empty user_id is treated as anonymous""" for empty_user_id in empty_user_ids: chat_data = { "prompt": "Test message with empty user_id", "max_new_tokens": 50, "use_search": False, "temperature": 0.7, "user_id": empty_user_id } response = await HTTPHelpers.make_chat_request(chat_data) if response is None: pytest.skip("Server not available") session_id = HTTPHelpers.assert_successful_chat_response(response) assert session_id is not None @pytest.mark.asyncio @skip_if_no_server() async def test_missing_user_id_field(self, sample_chat_request): """Test that missing user_id field works (backward compatibility)""" response = await HTTPHelpers.make_chat_request(sample_chat_request) if response is None: pytest.skip("Server not available") session_id = HTTPHelpers.assert_successful_chat_response(response) assert session_id is not None class TestChatRequestFlow: """Test complete chat request flow with user authentication""" @pytest.mark.asyncio @skip_if_no_server() async def test_authenticated_user_session_flow(self): """Test complete flow for authenticated user""" user_id = "test_flow_user" # First request - creates new session chat_data1 = { "prompt": "First message from authenticated user", "max_new_tokens": 50, "use_search": False, "temperature": 0.7, "user_id": user_id } response1 = await HTTPHelpers.make_chat_request(chat_data1) if response1 is None: pytest.skip("Server not available") session_id = HTTPHelpers.assert_successful_chat_response(response1) # Second request - uses existing session chat_data2 = { "prompt": "Second message from same user", "max_new_tokens": 50, "use_search": True, # Enable search this time "temperature": 0.7, "user_id": user_id } response2 = await HTTPHelpers.make_chat_request(chat_data2, session_id=session_id) if response2 is None: pytest.skip("Server not available") session_id2 = HTTPHelpers.assert_successful_chat_response(response2) # Should return same session ID assert session_id2 == session_id # Wait for data to be written await TestHelpers.wait_for_data_persistence(2.0) # Verify data was stored correctly session_valid = await TestHelpers.verify_session_data( session_id, user_id, expected_messages=2 ) assert session_valid @pytest.mark.asyncio @skip_if_no_server() async def test_anonymous_user_session_flow(self): """Test complete flow for anonymous user""" # First request - anonymous user chat_data1 = { "prompt": "First message from anonymous user", "max_new_tokens": 50, "use_search": False, "temperature": 0.7 # No user_id field } response1 = await HTTPHelpers.make_chat_request(chat_data1) if response1 is None: pytest.skip("Server not available") session_id = HTTPHelpers.assert_successful_chat_response(response1) # Second request - same anonymous user chat_data2 = { "prompt": "Second message from anonymous user", "max_new_tokens": 50, "use_search": True, "temperature": 0.7 # No user_id field } response2 = await HTTPHelpers.make_chat_request(chat_data2, session_id=session_id) if response2 is None: pytest.skip("Server not available") session_id2 = HTTPHelpers.assert_successful_chat_response(response2) assert session_id2 == session_id # Wait for data to be written await TestHelpers.wait_for_data_persistence(2.0) # Verify data was stored correctly (user_id should be None) session_valid = await TestHelpers.verify_session_data( session_id, None, expected_messages=2 ) assert session_valid @pytest.mark.asyncio @skip_if_no_server() async def test_mixed_user_sessions(self): """Test that different users get different sessions""" user_id1 = "test_user_1" user_id2 = "test_user_2" # Request from user 1 chat_data1 = { "prompt": "Message from user 1", "max_new_tokens": 50, "use_search": False, "temperature": 0.7, "user_id": user_id1 } response1 = await HTTPHelpers.make_chat_request(chat_data1) if response1 is None: pytest.skip("Server not available") session_id1 = HTTPHelpers.assert_successful_chat_response(response1) # Request from user 2 chat_data2 = { "prompt": "Message from user 2", "max_new_tokens": 50, "use_search": False, "temperature": 0.7, "user_id": user_id2 } response2 = await HTTPHelpers.make_chat_request(chat_data2) if response2 is None: pytest.skip("Server not available") session_id2 = HTTPHelpers.assert_successful_chat_response(response2) # Sessions should be different assert session_id1 != session_id2 @pytest.mark.asyncio @skip_if_no_server() async def test_session_continuation_with_user_id(self): """Test that sessions can be continued with proper user_id""" user_id = "test_continuation_user" # First request creates session chat_data1 = { "prompt": "First message in session", "max_new_tokens": 50, "use_search": False, "temperature": 0.7, "user_id": user_id } response1 = await HTTPHelpers.make_chat_request(chat_data1) if response1 is None: pytest.skip("Server not available") session_id = HTTPHelpers.assert_successful_chat_response(response1) # Second request continues same session with same user_id chat_data2 = { "prompt": "Second message in same session", "max_new_tokens": 50, "use_search": True, "temperature": 0.7, "user_id": user_id } response2 = await HTTPHelpers.make_chat_request(chat_data2, session_id=session_id) if response2 is None: pytest.skip("Server not available") session_id2 = HTTPHelpers.assert_successful_chat_response(response2) # Should be same session assert session_id2 == session_id class TestChatAnalyticsIntegration: """Test that chat requests properly trigger analytics collection""" @pytest.mark.asyncio @skip_if_no_server() async def test_chat_request_creates_analytics_data(self): """Test that chat requests create analytics data""" user_id = "analytics_integration_user" chat_data = { "prompt": "Test message for analytics collection", "max_new_tokens": 100, "use_search": True, "temperature": 0.7, "user_id": user_id } response = await HTTPHelpers.make_chat_request(chat_data) if response is None: pytest.skip("Server not available") session_id = HTTPHelpers.assert_successful_chat_response(response) # Wait for analytics data to be written await TestHelpers.wait_for_data_persistence(3.0) # Verify session was created session_valid = await TestHelpers.verify_session_data( session_id, user_id, expected_messages=1 ) assert session_valid # Check that we can count the user's data session_count = await TestHelpers.count_documents_by_user_id("sessions", user_id) message_count = await TestHelpers.count_documents_by_user_id("messages", user_id) assert session_count >= 1 assert message_count >= 1 @pytest.mark.asyncio @skip_if_no_server() async def test_anonymous_chat_request_creates_analytics_data(self): """Test that anonymous chat requests create analytics data""" chat_data = { "prompt": "Test anonymous message for analytics", "max_new_tokens": 100, "use_search": False, "temperature": 0.7 } response = await HTTPHelpers.make_chat_request(chat_data) if response is None: pytest.skip("Server not available") session_id = HTTPHelpers.assert_successful_chat_response(response) # Wait for analytics data to be written await TestHelpers.wait_for_data_persistence(3.0) # Verify session was created with null user_id session_valid = await TestHelpers.verify_session_data( session_id, None, expected_messages=1 ) assert session_valid # Check that we can count anonymous data anon_session_count = await TestHelpers.count_documents_by_user_id("sessions", None) anon_message_count = await TestHelpers.count_documents_by_user_id("messages", None) assert anon_session_count >= 1 assert anon_message_count >= 1 @pytest.mark.asyncio @skip_if_no_server() async def test_search_analytics_collection(self): """Test that search analytics are collected when search is used""" user_id = "search_analytics_user" chat_data = { "prompt": "What is the weather like today?", "max_new_tokens": 100, "use_search": True, # Enable search "temperature": 0.7, "user_id": user_id } response = await HTTPHelpers.make_chat_request(chat_data) if response is None: pytest.skip("Server not available") session_id = HTTPHelpers.assert_successful_chat_response(response) # Wait for analytics data to be written await TestHelpers.wait_for_data_persistence(5.0) # Search might take longer # Verify session and message were created session_valid = await TestHelpers.verify_session_data( session_id, user_id, expected_messages=1 ) assert session_valid # Check for search analytics (if search was actually performed) search_count = await TestHelpers.count_documents_by_user_id("search_analytics", user_id) # Note: Search analytics might be 0 if search was not actually performed # This is acceptable as it depends on the search implementation assert search_count >= 0 class TestChatRequestPerformance: """Test performance of chat requests with user authentication""" @pytest.mark.asyncio @skip_if_no_server() async def test_authenticated_request_performance(self): """Test performance of authenticated chat requests""" user_id = "perf_test_user" # Warm up request warmup_data = { "prompt": "Warmup message", "max_new_tokens": 50, "use_search": False, "temperature": 0.7, "user_id": user_id } warmup_response = await HTTPHelpers.make_chat_request(warmup_data) if warmup_response is None: pytest.skip("Server not available") # Performance test num_requests = 3 # Keep small for CI total_time = 0 for i in range(num_requests): chat_data = { "prompt": f"Performance test message {i}", "max_new_tokens": 50, "use_search": False, "temperature": 0.7, "user_id": user_id } start_time = time.time() response = await HTTPHelpers.make_chat_request(chat_data) end_time = time.time() if response is None: pytest.skip("Server not available") HTTPHelpers.assert_successful_chat_response(response) request_time = end_time - start_time total_time += request_time avg_time = total_time / num_requests # Performance assertion (requests should be reasonably fast) assert avg_time < 30.0, f"Requests too slow: {avg_time:.2f}s average" @pytest.mark.asyncio @skip_if_no_server() async def test_anonymous_vs_authenticated_performance(self): """Compare performance between anonymous and authenticated requests""" # Test anonymous requests anonymous_times = [] for i in range(2): # Keep small for CI chat_data = { "prompt": f"Anonymous performance test {i}", "max_new_tokens": 50, "use_search": False, "temperature": 0.7 } start_time = time.time() response = await HTTPHelpers.make_chat_request(chat_data) end_time = time.time() if response is None: pytest.skip("Server not available") HTTPHelpers.assert_successful_chat_response(response) anonymous_times.append(end_time - start_time) # Test authenticated requests authenticated_times = [] for i in range(2): # Keep small for CI chat_data = { "prompt": f"Authenticated performance test {i}", "max_new_tokens": 50, "use_search": False, "temperature": 0.7, "user_id": "perf_auth_user" } start_time = time.time() response = await HTTPHelpers.make_chat_request(chat_data) end_time = time.time() if response is None: pytest.skip("Server not available") HTTPHelpers.assert_successful_chat_response(response) authenticated_times.append(end_time - start_time) avg_anonymous = sum(anonymous_times) / len(anonymous_times) avg_authenticated = sum(authenticated_times) / len(authenticated_times) # Performance should be similar (user authentication shouldn't add significant overhead) time_difference = abs(avg_authenticated - avg_anonymous) assert time_difference < 10.0, f"Too much performance difference: {time_difference:.2f}s" class TestChatRequestErrorHandling: """Test error handling in chat requests""" @pytest.mark.asyncio @skip_if_no_server() async def test_malformed_request_handling(self): """Test handling of malformed requests""" malformed_requests = [ {}, # Empty request {"prompt": ""}, # Empty prompt {"user_id": "valid_user"}, # Missing prompt {"prompt": "test", "user_id": "invalid@user"}, # Invalid user_id ] for malformed_data in malformed_requests: response = await HTTPHelpers.make_chat_request(malformed_data) if response is None: pytest.skip("Server not available") # Should return error status assert response.status_code >= 400 @pytest.mark.asyncio @skip_if_no_server() async def test_request_with_invalid_session_id(self): """Test request with invalid session ID""" chat_data = { "prompt": "Test message with invalid session", "max_new_tokens": 50, "use_search": False, "temperature": 0.7, "user_id": "test_user" } # Use invalid session ID invalid_session_id = "invalid-session-id-12345" response = await HTTPHelpers.make_chat_request(chat_data, session_id=invalid_session_id) if response is None: pytest.skip("Server not available") # Should still work (create new session or handle gracefully) # The exact behavior depends on implementation assert response.status_code in [200, 400, 404] class TestChatRequestWithMocks: """Test chat requests with mocked dependencies""" @pytest.mark.asyncio async def test_chat_request_with_mocked_ai(self): """Test chat request with mocked AI response""" with MockHelpers.mock_gemini_inference("Mocked AI response"), \ MockHelpers.mock_web_search([]): chat_data = { "prompt": "Test prompt for mocked AI", "max_new_tokens": 100, "use_search": False, "temperature": 0.7, "user_id": "mock_test_user" } response = await HTTPHelpers.make_chat_request(chat_data) if response is None: pytest.skip("Server not available") HTTPHelpers.assert_successful_chat_response(response) # Verify the mocked response data = response.json() assert "Mocked AI response" in data["response"] @pytest.mark.asyncio async def test_chat_request_with_mocked_search(self): """Test chat request with mocked search results""" mock_results = [ {"title": "Test Result", "url": "https://example.com", "snippet": "Test snippet"} ] with MockHelpers.mock_gemini_inference("AI response with search"), \ MockHelpers.mock_web_search(mock_results): chat_data = { "prompt": "Test prompt with search", "max_new_tokens": 100, "use_search": True, "temperature": 0.7, "user_id": "search_mock_user" } response = await HTTPHelpers.make_chat_request(chat_data) if response is None: pytest.skip("Server not available") HTTPHelpers.assert_successful_chat_response(response) if __name__ == "__main__": # Run tests manually for debugging async def run_basic_tests(): test_validation = TestChatRequestValidation() # Note: These would need fixtures to run manually print("✅ Chat request validation tests defined") test_flow = TestChatRequestFlow() print("✅ Chat request flow tests defined") test_analytics = TestChatAnalyticsIntegration() print("✅ Chat analytics integration tests defined") asyncio.run(run_basic_tests())