CI/CD pipeline

tests/system_test.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""
+Simple test script to verify the women's safety app is working correctly.
+"""
+
+import asyncio
+import asyncpg
+import sys
+from pathlib import Path
+
+# Add project root to path
+sys.path.insert(0, str(Path(__file__).parent))
+
+from shared.config import settings
+from shared.database import engine, AsyncSessionLocal
+from services.user_service.models import User
+from services.user_service.schemas import UserCreate
+from shared.auth import get_password_hash
+from sqlalchemy import text
+
+
+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 get_password_hash, verify_password, create_access_token, 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()))