tests/test_user_operations.py

"""
Test script to verify deactivate_user and suspend_user functionality
"""
import asyncio
import sys
sys.path.insert(0, '/Users/mukeshkapoor/projects/cuatrolabs/backend/cuatrolabs-scm-ms')

from motor.motor_asyncio import AsyncIOMotorClient
from app.core.config import settings
from app.constants.collections import SCM_SYSTEM_USERS_COLLECTION

async def test_user_operations():
    """Test deactivate and suspend user operations"""
    client = AsyncIOMotorClient(settings.MONGODB_URI)
    db = client[settings.MONGODB_DB_NAME]
    users_collection = db[SCM_SYSTEM_USERS_COLLECTION]
    
    print("\n" + "="*70)
    print("🧪 Testing User Operations (Deactivate & Suspend)")
    print("="*70 + "\n")
    
    # Get a test user
    test_user = await users_collection.find_one(
        {"user_id": {"$exists": True}},
        {"user_id": 1, "email": 1, "is_active": 1}
    )
    
    if not test_user:
        print("❌ No users found in database!")
        client.close()
        return
    
    user_id = test_user["user_id"]
    print(f"📋 Using test user: {test_user['email']} (ID: {user_id})")
    print(f"   Current is_active: {test_user.get('is_active', True)}\n")
    
    # Test 1: Deactivate user
    print("1️⃣ Testing DEACTIVATE_USER")
    print("-" * 70)
    
    result = await users_collection.update_one(
        {"user_id": user_id},
        {"$set": {
            "is_active": False,
            "updated_by": "test_script"
        }}
    )
    
    print(f"   Matched: {result.matched_count}, Modified: {result.modified_count}")
    
    user_after = await users_collection.find_one({"user_id": user_id})
    print(f"   After deactivate - is_active: {user_after.get('is_active')}")
    
    if user_after.get('is_active') == False:
        print("   ✅ DEACTIVATE works correctly\n")
    else:
        print("   ❌ DEACTIVATE failed\n")
    
    # Test 2: Suspend user
    print("2️⃣ Testing SUSPEND_USER")
    print("-" * 70)
    
    result = await users_collection.update_one(
        {"user_id": user_id},
        {"$set": {
            "is_active": False,
            "suspension_reason": "Test suspension",
            "suspended_by": "test_script"
        }}
    )
    
    print(f"   Matched: {result.matched_count}, Modified: {result.modified_count}")
    
    user_after = await users_collection.find_one({"user_id": user_id})
    print(f"   After suspend - is_active: {user_after.get('is_active')}")
    print(f"   After suspend - suspension_reason: {user_after.get('suspension_reason')}")
    print(f"   After suspend - suspended_by: {user_after.get('suspended_by')}")
    
    if (user_after.get('is_active') == False and 
        user_after.get('suspension_reason') == "Test suspension"):
        print("   ✅ SUSPEND works correctly\n")
    else:
        print("   ❌ SUSPEND failed\n")
    
    # Test 3: Reactivate user
    print("3️⃣ Testing REACTIVATE_USER")
    print("-" * 70)
    
    result = await users_collection.update_one(
        {"user_id": user_id},
        {"$set": {
            "is_active": True,
            "updated_by": "test_script"
        },
        "$unset": {
            "suspension_reason": "",
            "suspended_by": ""
        }}
    )
    
    print(f"   Matched: {result.matched_count}, Modified: {result.modified_count}")
    
    user_after = await users_collection.find_one({"user_id": user_id})
    print(f"   After reactivate - is_active: {user_after.get('is_active')}")
    print(f"   After reactivate - suspension_reason: {user_after.get('suspension_reason')}")
    
    if user_after.get('is_active') == True:
        print("   ✅ REACTIVATE works correctly\n")
    else:
        print("   ❌ REACTIVATE failed\n")
    
    # Test 4: Unlock user
    print("4️⃣ Testing UNLOCK_USER")
    print("-" * 70)
    
    # First lock the user
    await users_collection.update_one(
        {"user_id": user_id},
        {"$set": {
            "account_locked_until": asyncio.get_event_loop().time() + 3600,
            "failed_login_attempts": 5
        }}
    )
    
    user_locked = await users_collection.find_one({"user_id": user_id})
    print(f"   Before unlock - account_locked_until: {user_locked.get('account_locked_until')}")
    print(f"   Before unlock - failed_login_attempts: {user_locked.get('failed_login_attempts')}")
    
    # Now unlock
    result = await users_collection.update_one(
        {"user_id": user_id},
        {
            "$set": {
                "updated_by": "test_script"
            },
            "$unset": {
                "account_locked_until": "",
                "failed_login_attempts": ""
            }
        }
    )
    
    print(f"   Matched: {result.matched_count}, Modified: {result.modified_count}")
    
    user_unlocked = await users_collection.find_one({"user_id": user_id})
    print(f"   After unlock - account_locked_until: {user_unlocked.get('account_locked_until')}")
    print(f"   After unlock - failed_login_attempts: {user_unlocked.get('failed_login_attempts')}")
    
    if (user_unlocked.get('account_locked_until') is None and
        user_unlocked.get('failed_login_attempts') is None):
        print("   ✅ UNLOCK works correctly\n")
    else:
        print("   ❌ UNLOCK failed\n")
    
    print("="*70)
    print("✅ All tests completed!\n")
    
    client.close()

if __name__ == "__main__":
    asyncio.run(test_user_operations())