222 lines
8.4 KiB
Python
222 lines
8.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Enhanced Gitea Webhook Ambassador feature test script
|
|
Demonstrates all new monitoring and management features
|
|
"""
|
|
|
|
import asyncio
|
|
import aiohttp
|
|
import json
|
|
from datetime import datetime
|
|
|
|
BASE_URL = "http://localhost:8000"
|
|
|
|
def print_divider():
|
|
print("-" * 50)
|
|
|
|
async def test_health_check():
|
|
"""Test enhanced health check"""
|
|
print("🧪 Testing enhanced health check")
|
|
print_divider()
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(f"{BASE_URL}/health") as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
print("✅ Health check passed")
|
|
print(f" Status: {data['status']}")
|
|
print(f" Service: {data['service']}")
|
|
print(f" Jenkins: {data['jenkins']['status']}")
|
|
print(f" Worker pool: {data['worker_pool']['active_workers']} active workers")
|
|
print(f" Queue size: {data['worker_pool']['queue_size']}")
|
|
print(f" Processed: {data['worker_pool']['total_processed']}")
|
|
print(f" Failed: {data['worker_pool']['total_failed']}")
|
|
else:
|
|
print(f"❌ Health check failed: {response.status}")
|
|
print()
|
|
|
|
async def test_webhook():
|
|
"""Test webhook feature"""
|
|
print("🧪 Testing webhook feature")
|
|
print_divider()
|
|
|
|
webhook_data = {
|
|
"ref": "refs/heads/dev",
|
|
"before": "abc123",
|
|
"after": "def456",
|
|
"repository": {
|
|
"full_name": "freeleaps/test-project",
|
|
"clone_url": "https://gitea.freeleaps.com/freeleaps/test-project.git"
|
|
},
|
|
"pusher": {
|
|
"login": "developer",
|
|
"email": "dev@freeleaps.com"
|
|
}
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
f"{BASE_URL}/webhook/gitea",
|
|
json=webhook_data
|
|
) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
print("✅ Webhook processed successfully")
|
|
print(f" Response: {data['message']}")
|
|
print(f" Data size: {data['data']['body_size']} bytes")
|
|
else:
|
|
print(f"❌ Webhook processing failed: {response.status}")
|
|
print()
|
|
|
|
async def test_api_key_management():
|
|
"""Test API key management"""
|
|
print("🧪 Testing API key management")
|
|
print_divider()
|
|
|
|
# Create API key
|
|
async with aiohttp.ClientSession() as session:
|
|
# Create key
|
|
create_data = {"name": "test-api-key"}
|
|
async with session.post(
|
|
f"{BASE_URL}/api/admin/api-keys",
|
|
json=create_data,
|
|
headers={"Authorization": "Bearer test-token"}
|
|
) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
api_key = data['key']
|
|
key_id = data['id']
|
|
print(f"✅ API key created successfully")
|
|
print(f" ID: {key_id}")
|
|
print(f" Name: {data['name']}")
|
|
print(f" Key: {api_key[:8]}...{api_key[-8:]}")
|
|
|
|
# Test logs endpoint with new key
|
|
print("\n Testing logs endpoint with new key...")
|
|
async with session.get(
|
|
f"{BASE_URL}/api/logs",
|
|
headers={"Authorization": f"Bearer {api_key}"}
|
|
) as log_response:
|
|
if log_response.status == 200:
|
|
logs = await log_response.json()
|
|
print(f" ✅ Logs access succeeded, retrieved {len(logs)} logs")
|
|
else:
|
|
print(f" ❌ Logs access failed: {log_response.status}")
|
|
|
|
# Delete key
|
|
async with session.delete(
|
|
f"{BASE_URL}/api/admin/api-keys/{key_id}",
|
|
headers={"Authorization": f"Bearer {api_key}"}
|
|
) as delete_response:
|
|
if delete_response.status == 200:
|
|
print(f" ✅ API key deleted successfully")
|
|
else:
|
|
print(f" ❌ API key deletion failed: {delete_response.status}")
|
|
else:
|
|
print(f"❌ API key creation failed: {response.status}")
|
|
print()
|
|
|
|
async def test_project_mapping():
|
|
"""Test project mapping management"""
|
|
print("🧪 Testing project mapping management")
|
|
print_divider()
|
|
|
|
mapping_data = {
|
|
"repository_name": "freeleaps/test-project",
|
|
"default_job": "test-project-build",
|
|
"branch_jobs": [
|
|
{"branch": "dev", "job": "test-project-dev"},
|
|
{"branch": "staging", "job": "test-project-staging"}
|
|
],
|
|
"branch_patterns": [
|
|
{"pattern": "feature/*", "job": "test-project-feature"},
|
|
{"pattern": "hotfix/*", "job": "test-project-hotfix"}
|
|
]
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
# Create project mapping
|
|
async with session.post(
|
|
f"{BASE_URL}/api/admin/projects",
|
|
json=mapping_data,
|
|
headers={"Authorization": "Bearer test-token"}
|
|
) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
print("✅ Project mapping created successfully")
|
|
print(f" ID: {data['id']}")
|
|
print(f" Repository: {data['repository_name']}")
|
|
print(f" Default job: {data['default_job']}")
|
|
print(f" Branch jobs: {len(data['branch_jobs'])}")
|
|
print(f" Branch patterns: {len(data['branch_patterns'])}")
|
|
else:
|
|
print(f"❌ Project mapping creation failed: {response.status}")
|
|
print()
|
|
|
|
async def test_logs_and_stats():
|
|
"""Test logs and statistics features"""
|
|
print("🧪 Testing logs and statistics features")
|
|
print_divider()
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
# Get log statistics
|
|
async with session.get(
|
|
f"{BASE_URL}/api/logs/stats",
|
|
headers={"Authorization": "Bearer test-token"}
|
|
) as response:
|
|
if response.status == 200:
|
|
stats = await response.json()
|
|
print("✅ Log statistics retrieved successfully")
|
|
print(f" Total logs: {stats['total_logs']}")
|
|
print(f" Successful logs: {stats['successful_logs']}")
|
|
print(f" Failed logs: {stats['failed_logs']}")
|
|
print(f" Recent logs (24h): {stats['recent_logs_24h']}")
|
|
print(f" Repository stats: {len(stats['repository_stats'])} repositories")
|
|
else:
|
|
print(f"❌ Log statistics retrieval failed: {response.status}")
|
|
print()
|
|
|
|
async def test_admin_stats():
|
|
"""Test admin statistics"""
|
|
print("🧪 Testing admin statistics")
|
|
print_divider()
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
f"{BASE_URL}/api/admin/stats",
|
|
headers={"Authorization": "Bearer test-token"}
|
|
) as response:
|
|
if response.status == 200:
|
|
stats = await response.json()
|
|
print("✅ Admin statistics retrieved successfully")
|
|
print(f" Total API keys: {stats['api_keys']['total']}")
|
|
print(f" Active keys: {stats['api_keys']['active']}")
|
|
print(f" Recently used: {stats['api_keys']['recently_used']}")
|
|
print(f" Total project mappings: {stats['project_mappings']['total']}")
|
|
else:
|
|
print(f"❌ Admin statistics retrieval failed: {response.status}")
|
|
print()
|
|
|
|
async def main():
|
|
"""Main test function"""
|
|
print("🚀 Starting enhanced Gitea Webhook Ambassador feature tests")
|
|
print("=" * 60)
|
|
print()
|
|
|
|
try:
|
|
await test_health_check()
|
|
await test_webhook()
|
|
await test_api_key_management()
|
|
await test_project_mapping()
|
|
await test_logs_and_stats()
|
|
await test_admin_stats()
|
|
|
|
print("=" * 60)
|
|
print("🎉 All tests completed!")
|
|
print("✅ Python version now has the same monitoring and management features as the Go version")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error occurred during testing: {str(e)}")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |