chat/tests/system_test.py

#!/usr/bin/env python3
"""
Simple test script to verify the women's safety app is working correctly.
"""

import asyncio
import sys
from pathlib import Path

import asyncpg

# Add project root to path
sys.path.insert(0, str(Path(__file__).parent))

from sqlalchemy import text

from services.user_service.models import User
from services.user_service.schemas import UserCreate
from shared.auth import get_password_hash
from shared.config import settings
from shared.database import AsyncSessionLocal, engine


async def test_database_connection():
    """Test basic database connectivity."""
    print("🔍 Testing database connection...")
    try:
        # Test direct asyncpg connection
        conn = await asyncpg.connect(settings.DATABASE_URL.replace("+asyncpg", ""))
        await conn.execute("SELECT 1")
        await conn.close()
        print("✅ Direct asyncpg connection successful")

        # Test SQLAlchemy engine connection
        async with engine.begin() as conn:
            result = await conn.execute(text("SELECT version()"))
            version = result.scalar()
            print(f"✅ SQLAlchemy connection successful (PostgreSQL {version[:20]}...)")

        return True
    except Exception as e:
        print(f"❌ Database connection failed: {e}")
        return False


async def test_database_tables():
    """Test database table structure."""
    print("🔍 Testing database tables...")
    try:
        async with AsyncSessionLocal() as session:
            # Test that we can query the users table
            result = await session.execute(text("SELECT COUNT(*) FROM users"))
            count = result.scalar()
            print(f"✅ Users table exists with {count} users")

            # Test table structure
            result = await session.execute(
                text(
                    """
                SELECT column_name, data_type 
                FROM information_schema.columns 
                WHERE table_name = 'users' 
                ORDER BY ordinal_position
                LIMIT 5
            """
                )
            )
            columns = result.fetchall()
            print(f"✅ Users table has columns: {[col[0] for col in columns]}")

        return True
    except Exception as e:
        print(f"❌ Database table test failed: {e}")
        return False


async def test_user_creation():
    """Test creating a user in the database."""
    print("🔍 Testing user creation...")
    try:
        async with AsyncSessionLocal() as session:
            # Create test user
            test_email = "test_debug@example.com"

            # Delete if exists
            await session.execute(
                text("DELETE FROM users WHERE email = :email"), {"email": test_email}
            )
            await session.commit()

            # Create new user
            user = User(
                email=test_email,
                phone="+1234567890",
                password_hash=get_password_hash("testpass"),
                first_name="Test",
                last_name="User",
            )
            session.add(user)
            await session.commit()

            # Verify creation
            result = await session.execute(
                text("SELECT id, email FROM users WHERE email = :email"),
                {"email": test_email},
            )
            user_row = result.fetchone()

            if user_row:
                print(
                    f"✅ User created successfully: ID={user_row[0]}, Email={user_row[1]}"
                )
                return True
            else:
                print("❌ User creation failed - user not found after creation")
                return False

    except Exception as e:
        print(f"❌ User creation test failed: {e}")
        return False


async def test_auth_functions():
    """Test authentication functions."""
    print("🔍 Testing authentication functions...")
    try:
        from shared.auth import (
            create_access_token,
            get_password_hash,
            verify_password,
            verify_token,
        )

        # Test password hashing
        password = "testpassword123"
        hashed = get_password_hash(password)
        print(f"✅ Password hashing works")

        # Test password verification
        if verify_password(password, hashed):
            print("✅ Password verification works")
        else:
            print("❌ Password verification failed")
            return False

        # Test token creation and verification
        token_data = {"sub": "123", "email": "test@example.com"}
        token = create_access_token(token_data)
        verified_data = verify_token(token)

        if verified_data and verified_data["user_id"] == 123:
            print("✅ Token creation and verification works")
        else:
            print("❌ Token verification failed")
            return False

        return True

    except Exception as e:
        print(f"❌ Authentication test failed: {e}")
        return False


async def main():
    """Run all tests."""
    print("🚀 Starting Women's Safety App System Tests")
    print(f"Database URL: {settings.DATABASE_URL}")
    print("=" * 60)

    tests = [
        test_database_connection,
        test_database_tables,
        test_user_creation,
        test_auth_functions,
    ]

    results = []
    for test in tests:
        try:
            result = await test()
            results.append(result)
        except Exception as e:
            print(f"❌ Test {test.__name__} failed with exception: {e}")
            results.append(False)
        print()

    print("=" * 60)
    if all(results):
        print("🎉 All tests passed! The system is ready for use.")
        return 0
    else:
        failed = len([r for r in results if not r])
        print(f"❌ {failed}/{len(results)} tests failed. Please check the errors above.")
        return 1


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))