#!/usr/bin/env python3 """ Backward compatibility tests for user authentication feature This test file ensures that existing anonymous user workflows continue to work exactly as they did before the user authentication feature was added. """ import sys import os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # Load environment variables try: from dotenv import load_dotenv load_dotenv() except ImportError: pass # dotenv not available, continue without it import asyncio import httpx import time from datetime import datetime from typing import Optional, Dict, Any from analytics.collectors import create_session, track_message, track_search from analytics.dashboard import get_basic_stats, get_hourly_message_stats, get_session_stats from analytics.database import get_sessions_collection, get_messages_collection, get_search_analytics_collection class TestAnonymousUserCompatibility: """Test that anonymous users work exactly as before""" async def test_create_session_without_user_id(self): """Test creating sessions without user_id parameter (old way)""" # Create session the old way (no user_id parameter) session = await create_session(user_agent="TestAgent") assert session.user_id is None assert session.user_agent == "TestAgent" assert session.session_id is not None assert session.status == "active" print("✅ Anonymous session creation works as before") async def test_create_session_with_none_user_id(self): """Test creating sessions with explicit None user_id""" # Create session with explicit None session = await create_session(user_agent="TestAgent", user_id=None) assert session.user_id is None assert session.user_agent == "TestAgent" assert session.session_id is not None assert session.status == "active" print("✅ Session creation with None user_id works") async def test_track_message_without_user_id(self): """Test tracking messages without user_id parameter (old way)""" # Create session first session = await create_session(user_agent="TestAgent") # Track message the old way (no user_id parameter) message = await track_message( session_id=session.session_id, prompt_length=50, response_length=100, response_time_ms=1000, used_search=True, max_tokens=500, temperature=0.7, success=True ) assert message is not None assert message.user_id is None assert message.session_id == session.session_id assert message.prompt_length == 50 assert message.response_length == 100 assert message.used_search is True print("✅ Anonymous message tracking works as before") async def test_track_message_with_none_user_id(self): """Test tracking messages with explicit None user_id""" # Create session first session = await create_session() # Track message with explicit None user_id message = await track_message( session_id=session.session_id, prompt_length=40, response_length=80, response_time_ms=800, used_search=False, user_id=None ) assert message is not None assert message.user_id is None assert message.session_id == session.session_id print("✅ Message tracking with None user_id works") async def test_track_search_without_user_id(self): """Test tracking search without user_id parameter (old way)""" # Create session and message first session = await create_session() message = await track_message( session_id=session.session_id, prompt_length=50, response_length=100, response_time_ms=1000 ) # Track search the old way (no user_id parameter) search = await track_search( message_id=message.message_id, search_query="test query", search_terms=["test", "query"], brave_results=5, duckduckgo_results=3, total_unique_results=7, brave_response_time_ms=1000, duckduckgo_response_time_ms=800, search_engines_used=["brave", "duckduckgo"], search_success=True, fallback_used=False ) assert search is not None assert search.user_id is None assert search.message_id == message.message_id assert search.search_query == "test query" print("✅ Anonymous search tracking works as before") async def test_track_search_with_none_user_id(self): """Test tracking search with explicit None user_id""" # Create session and message first session = await create_session() message = await track_message( session_id=session.session_id, prompt_length=50, response_length=100, response_time_ms=1000 ) # Track search with explicit None user_id search = await track_search( message_id=message.message_id, search_query="test query", search_terms=["test", "query"], user_id=None ) assert search is not None assert search.user_id is None print("✅ Search tracking with None user_id works") class TestAnonymousChatRequests: """Test that anonymous chat requests work as before""" async def test_chat_request_without_user_id_field(self): """Test chat request without user_id field (old API format)""" chat_data = { "prompt": "Test anonymous chat request", "max_new_tokens": 100, "use_search": False, "temperature": 0.7 # No user_id field - this is the old format } try: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( "http://localhost:7860/chat", json=chat_data, headers={"Content-Type": "application/json"} ) assert response.status_code == 200 result = response.json() assert "response" in result assert isinstance(result["response"], str) assert len(result["response"]) > 0 # Check session ID in headers session_id = response.headers.get('X-Session-ID') assert session_id is not None print("✅ Anonymous chat request (old format) works") return session_id except httpx.ConnectError: print("⚠️ Server not running - skipping chat request test") return None async def test_chat_request_with_null_user_id(self): """Test chat request with null user_id""" chat_data = { "prompt": "Test chat request with null user_id", "max_new_tokens": 100, "use_search": False, "temperature": 0.7, "user_id": None } try: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( "http://localhost:7860/chat", json=chat_data, headers={"Content-Type": "application/json"} ) assert response.status_code == 200 result = response.json() assert "response" in result print("✅ Chat request with null user_id works") except httpx.ConnectError: print("⚠️ Server not running - skipping chat request test") async def test_multiple_anonymous_requests(self): """Test multiple anonymous requests work as before""" try: async with httpx.AsyncClient(timeout=30.0) as client: # First anonymous request chat_data1 = { "prompt": "First anonymous message", "max_new_tokens": 50, "use_search": False, "temperature": 0.7 } response1 = await client.post( "http://localhost:7860/chat", json=chat_data1, headers={"Content-Type": "application/json"} ) assert response1.status_code == 200 session_id1 = response1.headers.get('X-Session-ID') # Second anonymous request (different session) chat_data2 = { "prompt": "Second anonymous message", "max_new_tokens": 50, "use_search": False, "temperature": 0.7 } response2 = await client.post( "http://localhost:7860/chat", json=chat_data2, headers={"Content-Type": "application/json"} ) assert response2.status_code == 200 session_id2 = response2.headers.get('X-Session-ID') # Should get different sessions (as before) assert session_id1 != session_id2 print("✅ Multiple anonymous requests work as before") except httpx.ConnectError: print("⚠️ Server not running - skipping multiple requests test") async def test_anonymous_session_continuation(self): """Test that anonymous sessions can be continued with session ID""" try: async with httpx.AsyncClient(timeout=30.0) as client: # First request creates session chat_data1 = { "prompt": "First message in session", "max_new_tokens": 50, "use_search": False, "temperature": 0.7 } response1 = await client.post( "http://localhost:7860/chat", json=chat_data1, headers={"Content-Type": "application/json"} ) assert response1.status_code == 200 session_id = response1.headers.get('X-Session-ID') # Second request continues same session chat_data2 = { "prompt": "Second message in same session", "max_new_tokens": 50, "use_search": True, "temperature": 0.7 } response2 = await client.post( "http://localhost:7860/chat", json=chat_data2, headers={ "Content-Type": "application/json", "X-Session-ID": session_id } ) assert response2.status_code == 200 session_id2 = response2.headers.get('X-Session-ID') # Should be same session assert session_id2 == session_id print("✅ Anonymous session continuation works as before") except httpx.ConnectError: print("⚠️ Server not running - skipping session continuation test") class TestAnalyticsFunctionCompatibility: """Test that analytics functions work with anonymous data""" async def test_basic_stats_with_anonymous_data(self): """Test that get_basic_stats works with anonymous data""" # Create some anonymous data session = await create_session() await track_message( session_id=session.session_id, prompt_length=50, response_length=100, response_time_ms=1000 ) # Test basic stats function stats = await get_basic_stats() assert isinstance(stats, dict) assert "total_sessions" in stats assert "total_messages" in stats assert "active_sessions" in stats assert stats["total_sessions"] >= 1 assert stats["total_messages"] >= 1 print("✅ Basic stats work with anonymous data") async def test_hourly_stats_with_anonymous_data(self): """Test that get_hourly_message_stats works with anonymous data""" # Create some anonymous data session = await create_session() await track_message( session_id=session.session_id, prompt_length=50, response_length=100, response_time_ms=1000 ) # Test hourly stats function hourly_stats = await get_hourly_message_stats(hours=24) assert isinstance(hourly_stats, list) # Should have 24 hours of data assert len(hourly_stats) == 24 for hour_data in hourly_stats: assert "hour" in hour_data assert "message_count" in hour_data assert "search_count" in hour_data assert "avg_response_time_ms" in hour_data print("✅ Hourly stats work with anonymous data") async def test_session_stats_with_anonymous_data(self): """Test that get_session_stats works with anonymous data""" # Create some anonymous data session = await create_session() await track_message( session_id=session.session_id, prompt_length=50, response_length=100, response_time_ms=1000 ) # Test session stats function session_stats = await get_session_stats() assert isinstance(session_stats, dict) assert "total_sessions" in session_stats assert "active_sessions" in session_stats assert "ended_sessions" in session_stats assert session_stats["total_sessions"] >= 1 print("✅ Session stats work with anonymous data") class TestDatabaseCompatibility: """Test that database operations work with anonymous data""" async def test_anonymous_data_storage(self): """Test that anonymous data is stored correctly in database""" # Create anonymous session and message session = await create_session(user_agent="TestAgent") message = await track_message( session_id=session.session_id, prompt_length=50, response_length=100, response_time_ms=1000 ) # Wait for data to be written await asyncio.sleep(1) # Check database storage sessions_collection = await get_sessions_collection() messages_collection = await get_messages_collection() if sessions_collection and messages_collection: # Check session document session_doc = await sessions_collection.find_one({"_id": session.session_id}) assert session_doc is not None assert session_doc.get("user_id") is None assert session_doc.get("user_agent") == "TestAgent" # Check message document message_doc = await messages_collection.find_one({"_id": message.message_id}) assert message_doc is not None assert message_doc.get("user_id") is None assert message_doc.get("session_id") == session.session_id print("✅ Anonymous data stored correctly in database") else: print("⚠️ Database not available - skipping storage test") async def test_anonymous_data_queries(self): """Test that queries work correctly with anonymous data""" # Create anonymous data session = await create_session() await track_message( session_id=session.session_id, prompt_length=50, response_length=100, response_time_ms=1000 ) # Wait for data to be written await asyncio.sleep(1) # Test queries sessions_collection = await get_sessions_collection() messages_collection = await get_messages_collection() if sessions_collection and messages_collection: # Query anonymous sessions anonymous_sessions = await sessions_collection.count_documents({"user_id": None}) assert anonymous_sessions >= 1 # Query anonymous messages anonymous_messages = await messages_collection.count_documents({"user_id": None}) assert anonymous_messages >= 1 # Query all sessions (should include anonymous) all_sessions = await sessions_collection.count_documents({}) assert all_sessions >= anonymous_sessions print("✅ Anonymous data queries work correctly") else: print("⚠️ Database not available - skipping query test") class TestMixedDataCompatibility: """Test that systems work with both anonymous and authenticated data""" async def test_mixed_data_analytics(self): """Test analytics functions with mixed anonymous and authenticated data""" # Create anonymous data anon_session = await create_session() await track_message( session_id=anon_session.session_id, prompt_length=50, response_length=100, response_time_ms=1000 ) # Create authenticated data auth_session = await create_session(user_id="test_user") await track_message( session_id=auth_session.session_id, prompt_length=60, response_length=120, response_time_ms=1200, user_id="test_user" ) # Test that analytics work with mixed data stats = await get_basic_stats() assert isinstance(stats, dict) assert stats["total_sessions"] >= 2 assert stats["total_messages"] >= 2 print("✅ Analytics work with mixed anonymous and authenticated data") async def test_mixed_data_queries(self): """Test database queries with mixed data""" # Create mixed data anon_session = await create_session() auth_session = await create_session(user_id="mixed_test_user") await track_message( session_id=anon_session.session_id, prompt_length=50, response_length=100, response_time_ms=1000 ) await track_message( session_id=auth_session.session_id, prompt_length=60, response_length=120, response_time_ms=1200, user_id="mixed_test_user" ) # Wait for data to be written await asyncio.sleep(1) # Test queries sessions_collection = await get_sessions_collection() messages_collection = await get_messages_collection() if sessions_collection and messages_collection: # Count anonymous vs authenticated anonymous_sessions = await sessions_collection.count_documents({"user_id": None}) authenticated_sessions = await sessions_collection.count_documents({"user_id": {"$ne": None}}) total_sessions = await sessions_collection.count_documents({}) assert anonymous_sessions >= 1 assert authenticated_sessions >= 1 assert total_sessions == anonymous_sessions + authenticated_sessions print("✅ Mixed data queries work correctly") else: print("⚠️ Database not available - skipping mixed query test") async def run_compatibility_tests(): """Run all backward compatibility tests""" print("🔄 Running Backward Compatibility Tests") print("=" * 50) # Test anonymous user compatibility anon_test = TestAnonymousUserCompatibility() await anon_test.test_create_session_without_user_id() await anon_test.test_create_session_with_none_user_id() await anon_test.test_track_message_without_user_id() await anon_test.test_track_message_with_none_user_id() await anon_test.test_track_search_without_user_id() await anon_test.test_track_search_with_none_user_id() print("✅ Anonymous user compatibility tests passed") # Test anonymous chat requests chat_test = TestAnonymousChatRequests() await chat_test.test_chat_request_without_user_id_field() await chat_test.test_chat_request_with_null_user_id() await chat_test.test_multiple_anonymous_requests() await chat_test.test_anonymous_session_continuation() print("✅ Anonymous chat request tests passed") # Test analytics function compatibility analytics_test = TestAnalyticsFunctionCompatibility() await analytics_test.test_basic_stats_with_anonymous_data() await analytics_test.test_hourly_stats_with_anonymous_data() await analytics_test.test_session_stats_with_anonymous_data() print("✅ Analytics function compatibility tests passed") # Test database compatibility db_test = TestDatabaseCompatibility() await db_test.test_anonymous_data_storage() await db_test.test_anonymous_data_queries() print("✅ Database compatibility tests passed") # Test mixed data compatibility mixed_test = TestMixedDataCompatibility() await mixed_test.test_mixed_data_analytics() await mixed_test.test_mixed_data_queries() print("✅ Mixed data compatibility tests passed") print("\n🎉 ALL BACKWARD COMPATIBILITY TESTS PASSED!") print("Existing anonymous user workflows continue to work as before.") if __name__ == "__main__": asyncio.run(run_compatibility_tests())