test_db_connection.py

"""Test MongoDB connection and verify database access."""

from __future__ import annotations

import sys
from pymongo import MongoClient
from pymongo.errors import ConnectionFailure, OperationFailure




def test_connection():
    """Test MongoDB connection and database access."""
    print("Testing MongoDB connection...")
    print(f"URI: {MONGO_URI.replace('Kem_6o%3F%3F', 'Kem_6o??')}")
    print("-" * 60)
    
    try:
        # Connect to MongoDB
        client = MongoClient(MONGO_URI, serverSelectionTimeoutMS=5000)
        
        # Test connection
        print("1. Testing connection...")
        client.admin.command('ping')
        print("   [OK] Connection successful!")
        
        # Get database
        db = client['expense']
        print(f"\n2. Accessing database: 'expense'")
        print("   [OK] Database accessed successfully!")
        
        # List collections
        print("\n3. Checking collections...")
        collections = db.list_collection_names()
        print(f"   Found {len(collections)} collections:")
        for col_name in sorted(collections):
            count = db[col_name].count_documents({})
            print(f"   - {col_name}: {count} documents")
        
        # Check specific collections used by the app
        print("\n4. Checking required collections...")
        required_collections = {
            'transactions': 'expense transactions',
            'merchant_aliases': 'merchant aliases',
            'merge_suggestions': 'merge suggestions'
        }
        
        for col_name, description in required_collections.items():
            if col_name in collections:
                count = db[col_name].count_documents({})
                print(f"   [OK] {col_name} ({description}): {count} documents")
            else:
                print(f"   [WARN] {col_name} ({description}): NOT FOUND (will be created when needed)")
        
        # Sample data from transactions
        print("\n5. Sampling transaction data...")
        transactions_col = db['transactions']
        sample_count = transactions_col.count_documents({})
        if sample_count > 0:
            sample = transactions_col.find_one()
            print(f"   Total transactions: {sample_count}")
            print(f"   Sample transaction fields: {list(sample.keys())[:10]}")
        else:
            print("   [WARN] No transactions found in database")
        
        # Sample merge suggestions
        print("\n6. Checking merge suggestions...")
        suggestions_col = db['merge_suggestions']
        suggestions_count = suggestions_col.count_documents({})
        print(f"   Total merge suggestions: {suggestions_count}")
        if suggestions_count > 0:
            sample_suggestion = suggestions_col.find_one()
            print(f"   Sample suggestion ID: {sample_suggestion.get('_id')}")
            print(f"   Candidate IDs: {sample_suggestion.get('candidate_ids', [])[:3]}")
        
        # Test write operation
        print("\n7. Testing write operation...")
        test_col = db['_connection_test']
        test_col.insert_one({'test': True, 'timestamp': 'test'})
        test_col.delete_one({'test': True})
        print("   [OK] Write/Delete operations working!")
        
        print("\n" + "=" * 60)
        print("[SUCCESS] All tests passed! Database connection is working.")
        print("=" * 60)
        
        return True
        
    except ConnectionFailure as e:
        print(f"\n[ERROR] Connection failed: {e}")
        print("   Check if MongoDB server is reachable and credentials are correct.")
        return False
        
    except OperationFailure as e:
        print(f"\n[ERROR] Authentication failed: {e}")
        print("   Check if username and password are correct.")
        return False
        
    except Exception as e:
        print(f"\n[ERROR] Error: {type(e).__name__}: {e}")
        return False
        
    finally:
        try:
            client.close()
        except:
            pass


if __name__ == "__main__":
    success = test_connection()
    sys.exit(0 if success else 1)