Spaces:
Sleeping
Sleeping
File size: 5,143 Bytes
04aa1ba | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | #!/usr/bin/env python3
"""
Test MongoDB connection for user authentication tests
This script verifies that the MongoDB connection is working properly
for the user authentication test suite.
"""
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()
print("✅ Environment variables loaded")
except ImportError:
print("⚠️ dotenv not available - continuing without .env file loading")
import asyncio
from analytics.database import connect_to_database, get_database
async def test_mongodb_connection():
"""Test MongoDB connection"""
print("\n🔍 Testing MongoDB Connection")
print("=" * 40)
# Check environment variables
mongodb_url = os.getenv("MONGODB_URL")
mongodb_db = os.getenv("MONGODB_DATABASE")
if mongodb_url:
print(f"✅ MONGODB_URL found: {mongodb_url[:30]}...")
else:
print("❌ MONGODB_URL not found")
return False
if mongodb_db:
print(f"✅ MONGODB_DATABASE found: {mongodb_db}")
else:
print("❌ MONGODB_DATABASE not found")
return False
# Test database connection
try:
print("\n🔗 Attempting to connect to MongoDB...")
database = await connect_to_database()
if database is not None:
print("✅ MongoDB connection successful!")
# Test a simple operation
try:
# Try to list collections
collections = await database.list_collection_names()
print(f"✅ Database accessible - found {len(collections)} collections")
if collections:
print(" Collections:", ", ".join(collections[:5]))
if len(collections) > 5:
print(f" ... and {len(collections) - 5} more")
return True
except Exception as e:
print(f"⚠️ Database accessible but operation failed: {e}")
return True # Connection works, operation might need permissions
else:
print("❌ MongoDB connection failed - falling back to JSON storage")
return False
except Exception as e:
print(f"❌ MongoDB connection error: {e}")
return False
async def test_analytics_collections():
"""Test analytics collections access"""
print("\n📊 Testing Analytics Collections")
print("=" * 40)
try:
from analytics.database import (
get_sessions_collection,
get_messages_collection,
get_search_analytics_collection
)
# Test sessions collection
sessions_collection = await get_sessions_collection()
if sessions_collection is not None:
count = await sessions_collection.count_documents({})
print(f"✅ Sessions collection accessible - {count} documents")
else:
print("⚠️ Sessions collection not available")
# Test messages collection
messages_collection = await get_messages_collection()
if messages_collection is not None:
count = await messages_collection.count_documents({})
print(f"✅ Messages collection accessible - {count} documents")
else:
print("⚠️ Messages collection not available")
# Test search analytics collection
search_collection = await get_search_analytics_collection()
if search_collection is not None:
count = await search_collection.count_documents({})
print(f"✅ Search analytics collection accessible - {count} documents")
else:
print("⚠️ Search analytics collection not available")
return True
except Exception as e:
print(f"❌ Analytics collections test failed: {e}")
return False
async def main():
"""Main test function"""
print("🧪 MongoDB Connection Test for User Authentication")
print("=" * 60)
# Test basic connection
connection_ok = await test_mongodb_connection()
# Test analytics collections
collections_ok = await test_analytics_collections()
print("\n" + "=" * 60)
if connection_ok and collections_ok:
print("🎉 MongoDB is ready for user authentication tests!")
print("✅ All database operations should work correctly")
elif connection_ok:
print("⚠️ MongoDB connection works but some collections may need setup")
print("✅ Basic tests should work, some advanced tests may be limited")
else:
print("❌ MongoDB connection failed")
print("⚠️ Tests will use JSON file fallback storage")
print("✅ Tests will still run but without persistent database storage")
return connection_ok
if __name__ == "__main__":
success = asyncio.run(main())
sys.exit(0 if success else 1) |