service-finder/backend/scripts/test_soft_delete_and_asset_validation.py

"""
Test script for:
1. Asset VIN/License Plate OR-OR validation (Pydantic schemas)
2. Person-Preserving Soft Delete logic
"""
import asyncio
import sys
from datetime import datetime, timezone
from typing import Optional

# ─── Test 1: Pydantic Schema Validation ───────────────────────────────────────
print("=" * 70)
print("🧪 TEST 1: Asset Pydantic OR-OR Validation")
print("=" * 70)

from pydantic import ValidationError
sys.path.insert(0, "/app")
from app.schemas.asset import AssetCreate, AssetUpdate

def test_asset_create():
    """Test AssetCreate schema validation."""
    passed = 0
    failed = 0
    
    # Test 1.1: Both vin and plate provided - should pass
    try:
        a = AssetCreate(license_plate="ABC-123", vin="WDB1234567890ABCD")
        assert a.license_plate == "ABC-123"
        assert a.vin == "WDB1234567890ABCD"
        print("  ✅ 1.1 Both vin and plate provided: PASS")
        passed += 1
    except Exception as e:
        print(f"  ❌ 1.1 Both vin and plate provided: FAIL - {e}")
        failed += 1
    
    # Test 1.2: Only license_plate - should pass
    try:
        a = AssetCreate(license_plate="ABC-123")
        assert a.license_plate == "ABC-123"
        assert a.vin is None
        print("  ✅ 1.2 Only license_plate: PASS")
        passed += 1
    except Exception as e:
        print(f"  ❌ 1.2 Only license_plate: FAIL - {e}")
        failed += 1
    
    # Test 1.3: Only vin - should pass
    try:
        a = AssetCreate(vin="WDB1234567890ABCD")
        assert a.vin == "WDB1234567890ABCD"
        assert a.license_plate is None
        print("  ✅ 1.3 Only vin: PASS")
        passed += 1
    except Exception as e:
        print(f"  ❌ 1.3 Only vin: FAIL - {e}")
        failed += 1
    
    # Test 1.4: Neither vin nor plate - should FAIL
    try:
        a = AssetCreate()
        print(f"  ❌ 1.4 Neither vin nor plate: FAIL (should have raised ValueError)")
        failed += 1
    except ValidationError as e:
        print(f"  ✅ 1.4 Neither vin nor plate: PASS (correctly rejected)")
        passed += 1
    except Exception as e:
        print(f"  ❌ 1.4 Neither vin nor plate: FAIL - {e}")
        failed += 1
    
    # Test 1.5: Empty string license_plate - should be converted to None, then fail
    try:
        a = AssetCreate(license_plate="")
        print(f"  ❌ 1.5 Empty string license_plate: FAIL (should have raised ValueError)")
        failed += 1
    except ValidationError as e:
        print(f"  ✅ 1.5 Empty string license_plate: PASS (correctly rejected)")
        passed += 1
    except Exception as e:
        print(f"  ❌ 1.5 Empty string license_plate: FAIL - {e}")
        failed += 1
    
    # Test 1.6: Empty string vin - should be converted to None, then fail
    try:
        a = AssetCreate(vin="   ")
        print(f"  ❌ 1.6 Whitespace-only vin: FAIL (should have raised ValueError)")
        failed += 1
    except ValidationError as e:
        print(f"  ✅ 1.6 Whitespace-only vin: PASS (correctly rejected)")
        passed += 1
    except Exception as e:
        print(f"  ❌ 1.6 Whitespace-only vin: FAIL - {e}")
        failed += 1
    
    # Test 1.7: Both empty strings - should fail
    try:
        a = AssetCreate(license_plate="", vin="")
        print(f"  ❌ 1.7 Both empty strings: FAIL (should have raised ValueError)")
        failed += 1
    except ValidationError as e:
        print(f"  ✅ 1.7 Both empty strings: PASS (correctly rejected)")
        passed += 1
    except Exception as e:
        print(f"  ❌ 1.7 Both empty strings: FAIL - {e}")
        failed += 1
    
    return passed, failed

def test_asset_update():
    """Test AssetUpdate schema validation."""
    passed = 0
    failed = 0
    
    # Test 2.1: Update with only license_plate - should pass
    try:
        a = AssetUpdate(license_plate="NEW-999")
        assert a.license_plate == "NEW-999"
        print("  ✅ 2.1 Update with only license_plate: PASS")
        passed += 1
    except Exception as e:
        print(f"  ❌ 2.1 Update with only license_plate: FAIL - {e}")
        failed += 1
    
    # Test 2.2: Update with only vin - should pass
    try:
        a = AssetUpdate(vin="NEWVIN123456789")
        assert a.vin == "NEWVIN123456789"
        print("  ✅ 2.2 Update with only vin: PASS")
        passed += 1
    except Exception as e:
        print(f"  ❌ 2.2 Update with only vin: FAIL - {e}")
        failed += 1
    
    # Test 2.3: Update with both set to None explicitly - should FAIL
    try:
        a = AssetUpdate(license_plate=None, vin=None)
        print(f"  ❌ 2.3 Both set to None: FAIL (should have raised ValueError)")
        failed += 1
    except ValidationError as e:
        print(f"  ✅ 2.3 Both set to None: PASS (correctly rejected)")
        passed += 1
    except Exception as e:
        print(f"  ❌ 2.3 Both set to None: FAIL - {e}")
        failed += 1
    
    # Test 2.4: Empty update (no fields) - should pass (exclude_unset)
    try:
        a = AssetUpdate()
        print(f"  ✅ 2.4 Empty update (no fields): PASS")
        passed += 1
    except Exception as e:
        print(f"  ❌ 2.4 Empty update (no fields): FAIL - {e}")
        failed += 1
    
    return passed, failed

# Run Pydantic tests
p1, f1 = test_asset_create()
p2, f2 = test_asset_update()
print(f"\n📊 Asset Validation Results: {p1+p2} passed, {f1+f2} failed out of {p1+p2+f1+f2} tests")

# ─── Test 2: Soft Delete Logic ────────────────────────────────────────────────
print("\n" + "=" * 70)
print("🧪 TEST 2: Person-Preserving Soft Delete Logic")
print("=" * 70)

async def test_soft_delete():
    """Test the soft_delete_user method via direct DB calls."""
    from app.database import AsyncSessionLocal
    from app.models.identity.identity import User, Person
    from app.services.auth_service import AuthService
    from app.services.security_service import security_service
    from sqlalchemy import select, text
    
    async with AsyncSessionLocal() as db:
        try:
            # Find a test user (use first active user)
            result = await db.execute(
                select(User).where(User.is_deleted == False).limit(1)
            )
            user = result.scalar_one_or_none()
            
            if not user:
                print("  ⚠️  No active test user found - creating one...")
                # Create a minimal test scenario
                print("  ⚠️  Skipping DB soft delete test (no test user available)")
                return 0, 0
            
            user_id = user.id
            old_email = user.email
            print(f"  📋 Test user: id={user_id}, email={old_email}")
            
            # Check if Person exists
            person = None
            if user.person_id:
                result = await db.execute(
                    select(Person).where(Person.id == user.person_id)
                )
                person = result.scalar_one_or_none()
                if person:
                    print(f"  📋 Person record: id={person.id}, is_active={person.is_active}")
            
            # Perform soft delete
            success = await AuthService.soft_delete_user(
                db=db,
                user_id=user_id,
                reason="test_script_validation",
                actor_id=user_id
            )
            
            if not success:
                print(f"  ❌ Soft delete returned False (user already deleted)")
                return 0, 1
            
            # Verify user state
            await db.refresh(user)
            assert user.is_deleted == True, "is_deleted should be True"
            assert user.is_active == False, "is_active should be False"
            assert user.deleted_at is not None, "deleted_at should be set"
            assert "deleted_" in user.email, f"Email should be anonymized: {user.email}"
            assert str(user_id) in user.email, "Email should contain user_id"
            
            print(f"  ✅ User.is_deleted = {user.is_deleted}")
            print(f"  ✅ User.is_active = {user.is_active}")
            print(f"  ✅ User.deleted_at = {user.deleted_at}")
            print(f"  ✅ User.email anonymized: {user.email}")
            
            # Verify Person is UNTOUCHED by our soft delete
            if person:
                await db.refresh(person)
                # Person.deleted_at must remain None (our code never sets it)
                assert person.deleted_at is None, "Person.deleted_at should remain None (not touched by soft delete)"
                # Person data must be unchanged (no email rewrite, no name change)
                print(f"  ✅ Person.deleted_at preserved = {person.deleted_at} (not touched)")
                print(f"  ✅ Person data preserved (no modifications by soft delete)")
            else:
                print(f"  ⚠️  No Person record linked to this user")
            
            # Restore user for other tests
            user.is_deleted = False
            user.is_active = True
            user.deleted_at = None
            user.email = old_email
            await db.commit()
            print(f"  ✅ User restored to original state")
            
            return 2, 0
            
        except Exception as e:
            await db.rollback()
            print(f"  ❌ Soft delete test failed: {e}")
            import traceback
            traceback.print_exc()
            return 0, 1

p3, f3 = asyncio.run(test_soft_delete())

# ─── Final Summary ────────────────────────────────────────────────────────────
print("\n" + "=" * 70)
total_p = p1 + p2 + p3
total_f = f1 + f2 + f3
print(f"📊 FINAL RESULTS: {total_p} passed, {total_f} failed out of {total_p+total_f} tests")
print("=" * 70)

if total_f > 0:
    print("❌ SOME TESTS FAILED!")
    sys.exit(1)
else:
    print("✅ ALL TESTS PASSED!")
    sys.exit(0)