freeleaps-ops/apps/gitea-webhook-ambassador-python/test_enhanced_features.py

#!/usr/bin/env python3
"""
Gitea Webhook Ambassador Enhanced Feature Test Script
"""

import requests
import json
import time

BASE_URL = "http://localhost:8000"
ADMIN_SECRET_KEY = "admin-secret-key-change-in-production"

def test_health_check():
    """Test health check"""
    print("🔍 Testing health check...")
    try:
        response = requests.get(f"{BASE_URL}/health")
        if response.status_code == 200:
            data = response.json()
            print(f"✅ Health check succeeded: {data['status']}")
            print(f"   Version: {data['version']}")
            print(f"   Uptime: {data['uptime']}")
            print(f"   Memory usage: {data['memory']}")
            return True
        else:
            print(f"❌ Health check failed: {response.status_code}")
            return False
    except Exception as e:
        print(f"❌ Health check exception: {e}")
        return False

def test_login():
    """Test login functionality"""
    print("\n🔐 Testing login functionality...")
    try:
        # Test login API
        login_data = {"secret_key": ADMIN_SECRET_KEY}
        response = requests.post(
            f"{BASE_URL}/api/auth/login",
            json=login_data,
            headers={"Content-Type": "application/json"}
        )
        
        if response.status_code == 200:
            data = response.json()
            token = data.get("token")
            print(f"✅ Login succeeded, obtained JWT token")
            return token
        else:
            print(f"❌ Login failed: {response.status_code} - {response.text}")
            return None
    except Exception as e:
        print(f"❌ Login exception: {e}")
        return None

def test_api_key_management(token):
    """Test API key management"""
    print("\n🔑 Testing API key management...")
    headers = {"Authorization": f"Bearer {token}"}
    
    try:
        # Create API key
        key_data = {"description": "Test API key"}
        response = requests.post(
            f"{BASE_URL}/api/keys",
            json=key_data,
            headers=headers
        )
        
        if response.status_code == 200:
            data = response.json()
            api_key = data["key"]
            key_id = data["id"]
            print(f"✅ API key created successfully: {api_key[:20]}...")
            
            # List API keys
            response = requests.get(f"{BASE_URL}/api/keys", headers=headers)
            if response.status_code == 200:
                keys_data = response.json()
                print(f"✅ API keys listed successfully, total {len(keys_data['keys'])}")
                
                # Delete API key
                response = requests.delete(f"{BASE_URL}/api/keys/{key_id}", headers=headers)
                if response.status_code == 200:
                    print(f"✅ API key deleted successfully")
                    return True
                else:
                    print(f"❌ API key deletion failed: {response.status_code}")
                    return False
            else:
                print(f"❌ API key listing failed: {response.status_code}")
                return False
        else:
            print(f"❌ API key creation failed: {response.status_code} - {response.text}")
            return False
    except Exception as e:
        print(f"❌ API key management exception: {e}")
        return False

def test_project_management(token):
    """Test project management"""
    print("\n📁 Testing project management...")
    headers = {"Authorization": f"Bearer {token}"}
    
    try:
        # Create project
        project_data = {
            "name": "Test Project",
            "jenkinsJob": "test-job",
            "giteaRepo": "test-owner/test-repo"
        }
        response = requests.post(
            f"{BASE_URL}/api/projects/",
            json=project_data,
            headers=headers
        )
        
        if response.status_code == 200:
            data = response.json()
            project_id = data["id"]
            print(f"✅ Project created successfully: {data['name']}")
            
            # List projects
            response = requests.get(f"{BASE_URL}/api/projects/", headers=headers)
            if response.status_code == 200:
                projects_data = response.json()
                print(f"✅ Projects listed successfully, total {len(projects_data['projects'])}")
                
                # Delete project
                response = requests.delete(f"{BASE_URL}/api/projects/{project_id}", headers=headers)
                if response.status_code == 200:
                    print(f"✅ Project deleted successfully")
                    return True
                else:
                    print(f"❌ Project deletion failed: {response.status_code}")
                    return False
            else:
                print(f"❌ Project listing failed: {response.status_code}")
                return False
        else:
            print(f"❌ Project creation failed: {response.status_code} - {response.text}")
            return False
    except Exception as e:
        print(f"❌ Project management exception: {e}")
        return False

def test_stats(token):
    """Test statistics information"""
    print("\n📊 Testing statistics information...")
    headers = {"Authorization": f"Bearer {token}"}
    
    try:
        response = requests.get(f"{BASE_URL}/api/stats", headers=headers)
        if response.status_code == 200:
            data = response.json()
            print(f"✅ Statistics retrieved successfully:")
            print(f"   Total projects: {data['total_projects']}")
            print(f"   API keys: {data['total_api_keys']}")
            print(f"   Triggers today: {data['today_triggers']}")
            print(f"   Successful triggers: {data['successful_triggers']}")
            return True
        else:
            print(f"❌ Statistics retrieval failed: {response.status_code}")
            return False
    except Exception as e:
        print(f"❌ Statistics exception: {e}")
        return False

def test_logs(token):
    """Test log functionality"""
    print("\n📝 Testing log functionality...")
    headers = {"Authorization": f"Bearer {token}"}
    
    try:
        response = requests.get(f"{BASE_URL}/api/logs", headers=headers)
        if response.status_code == 200:
            data = response.json()
            print(f"✅ Logs retrieved successfully, total {len(data['logs'])} records")
            return True
        else:
            print(f"❌ Log retrieval failed: {response.status_code}")
            return False
    except Exception as e:
        print(f"❌ Log functionality exception: {e}")
        return False

def main():
    """Main test function"""
    print("🚀 Gitea Webhook Ambassador Enhanced Feature Test")
    print("=" * 50)
    
    # Test health check
    if not test_health_check():
        print("❌ Health check failed, service may not be running")
        return
    
    # Test login
    token = test_login()
    if not token:
        print("❌ Login failed, cannot continue testing")
        return
    
    # Test features
    test_api_key_management(token)
    test_project_management(token)
    test_stats(token)
    test_logs(token)
    
    print("\n" + "=" * 50)
    print("🎉 Enhanced feature test completed!")
    print("\n📋 Implemented features:")
    print("  ✅ Web login interface")
    print("  ✅ Database storage for API keys")
    print("  ✅ Extended JWT validity (7 days)")
    print("  ✅ Frontend dashboard")
    print("  ✅ Project management")
    print("  ✅ API key management")
    print("  ✅ Log viewing")
    print("  ✅ Health status monitoring")
    print("\n🌐 Access URLs:")
    print(f"   Login page: {BASE_URL}/login")
    print(f"   Dashboard: {BASE_URL}/dashboard")

if __name__ == "__main__":
    main()