forked from jkcl/reposync
234 lines
9.0 KiB
Python
234 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
import asyncio
|
||
import sys
|
||
import os
|
||
from datetime import datetime
|
||
|
||
# 添加项目根目录到Python路径
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from src.service.pr_sync import PRSyncService
|
||
from src.dto.sync import PRSyncRequest
|
||
|
||
async def test_gitlink_api_corrected():
|
||
"""测试修正后的GitLink API"""
|
||
print("=" * 70)
|
||
print("测试修正后的GitLink API")
|
||
print("=" * 70)
|
||
print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print()
|
||
|
||
print("根据GitLink官方API文档修正的API:")
|
||
print("1. 获取PR列表: GET /api/v1/{owner}/{repo}/pulls.json")
|
||
print("2. 创建PR: POST /api/v1/{owner}/{repo}/pulls")
|
||
print("3. 关闭PR: PUT /api/v1/{owner}/{repo}/pulls/{index}")
|
||
print("4. 重新开启PR: POST /api/v1/{owner}/{repo}/pulls/{index}/reopen.json")
|
||
print("5. 获取单个PR: GET /api/v1/{owner}/{repo}/pulls/{index}")
|
||
print()
|
||
|
||
# 测试配置
|
||
test_configs = [
|
||
{
|
||
"name": "GitHub -> GitLink 同步测试（验证修正后的API）",
|
||
"source_project": "https://github.com/demon-two/repo1",
|
||
"target_project": "https://www.gitlink.org.cn/gonggong123zzz/repo2",
|
||
"source_type": "github",
|
||
"target_type": "gitlink"
|
||
},
|
||
{
|
||
"name": "GitLink -> GitHub 同步测试（验证修正后的API）",
|
||
"source_project": "https://www.gitlink.org.cn/gonggong123zzz/repo2",
|
||
"target_project": "https://github.com/demon-two/repo1",
|
||
"source_type": "gitlink",
|
||
"target_type": "github"
|
||
}
|
||
]
|
||
|
||
for i, config in enumerate(test_configs, 1):
|
||
print(f"测试场景 {i}: {config['name']}")
|
||
print("-" * 60)
|
||
|
||
# 创建同步请求
|
||
request = PRSyncRequest(
|
||
source_project=config["source_project"],
|
||
target_project=config["target_project"],
|
||
source_type=config["source_type"],
|
||
target_type=config["target_type"],
|
||
sync_direction="bidirectional"
|
||
)
|
||
|
||
print(f"源仓库: {request.source_project}")
|
||
print(f"目标仓库: {request.target_project}")
|
||
print(f"源类型: {request.source_type}, 目标类型: {request.target_type}")
|
||
print(f"同步方向: {request.sync_direction}")
|
||
print()
|
||
|
||
try:
|
||
# 创建同步服务
|
||
sync_service = PRSyncService()
|
||
|
||
# 执行双向同步
|
||
print("开始执行双向同步（验证修正后的API）...")
|
||
start_time = datetime.now()
|
||
result = await sync_service.sync_pull_requests_bidirectional(request)
|
||
end_time = datetime.now()
|
||
|
||
# 计算执行时间
|
||
execution_time = (end_time - start_time).total_seconds()
|
||
|
||
print(f"\n同步结果: {result.message}")
|
||
print(f"执行时间: {execution_time:.2f} 秒")
|
||
print(f"创建数量: {result.created_count}")
|
||
print(f"关闭数量: {result.closed_count}")
|
||
print(f"重新开启数量: {result.reopened_count}")
|
||
print(f"失败数量: {result.failed_count}")
|
||
print(f"总处理数量: {result.total_count}")
|
||
|
||
if result.details:
|
||
print("\n详细结果:")
|
||
created_count = 0
|
||
reopened_count = 0
|
||
skipped_count = 0
|
||
failed_count = 0
|
||
|
||
for detail in result.details:
|
||
action = detail.get('action', '')
|
||
pr_number = detail.get('pr_number', '')
|
||
title = detail.get('title', '')
|
||
error = detail.get('error', '')
|
||
|
||
if action == 'created':
|
||
print(f" ✅ 创建PR: #{pr_number} - {title}")
|
||
created_count += 1
|
||
elif action == 'reopened':
|
||
print(f" 🔄 重新开启PR: #{pr_number} - {title}")
|
||
reopened_count += 1
|
||
elif action == 'closed':
|
||
print(f" ❌ 关闭PR: #{pr_number} - {title}")
|
||
elif action == 'failed':
|
||
print(f" ❌ 失败: #{pr_number} - {error}")
|
||
failed_count += 1
|
||
elif action == 'skipped':
|
||
print(f" ⏭️ 跳过: #{pr_number} - {title} (已存在且为open状态)")
|
||
skipped_count += 1
|
||
|
||
print(f"\n操作统计:")
|
||
print(f" - 创建: {created_count}")
|
||
print(f" - 重新开启: {reopened_count}")
|
||
print(f" - 跳过: {skipped_count}")
|
||
print(f" - 失败: {failed_count}")
|
||
|
||
# 分析结果
|
||
print(f"\n结果分析:")
|
||
if result.created_count > 0:
|
||
print(f" ✅ 成功创建了 {result.created_count} 个PR")
|
||
if result.reopened_count > 0:
|
||
print(f" ✅ 成功重新开启了 {result.reopened_count} 个PR")
|
||
if result.failed_count > 0:
|
||
print(f" ⚠️ 有 {result.failed_count} 个操作失败")
|
||
# 检查是否有404错误
|
||
for detail in result.details:
|
||
if detail.get('action') == 'failed' and '404' in detail.get('error', ''):
|
||
print(f" ❌ 仍然存在404错误，可能需要进一步调试")
|
||
if result.created_count == 0 and result.reopened_count == 0 and result.failed_count == 0:
|
||
print(f" ✅ 双向同步完成，所有开放PR都已同步或已存在")
|
||
|
||
# 性能评估
|
||
if execution_time < 10:
|
||
print(f" 🚀 性能优秀 (执行时间: {execution_time:.2f}s)")
|
||
elif execution_time < 30:
|
||
print(f" ⚡ 性能良好 (执行时间: {execution_time:.2f}s)")
|
||
else:
|
||
print(f" 🐌 性能较慢 (执行时间: {execution_time:.2f}s)")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 测试失败: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
print("\n" + "=" * 70)
|
||
print()
|
||
|
||
async def test_individual_gitlink_apis():
|
||
"""测试各个GitLink API"""
|
||
print("=" * 70)
|
||
print("测试各个GitLink API")
|
||
print("=" * 70)
|
||
|
||
try:
|
||
from src.utils.pr_sync import PRSyncUtils
|
||
|
||
# 创建GitLink工具类
|
||
pr_utils = PRSyncUtils()
|
||
gitlink_utils = pr_utils.get_platform_utils("gitlink")
|
||
|
||
owner = "gonggong123zzz"
|
||
repo = "repo2"
|
||
|
||
print(f"测试GitLink API: {owner}/{repo}")
|
||
print()
|
||
|
||
# 1. 测试获取PR列表
|
||
print("1. 测试获取PR列表...")
|
||
prs = await gitlink_utils.get_pull_requests(owner, repo, "all")
|
||
print(f" 获取到 {len(prs)} 个PR")
|
||
|
||
# 2. 测试获取单个PR
|
||
if prs:
|
||
test_pr = prs[0]
|
||
pr_number = test_pr.get('number')
|
||
print(f"2. 测试获取单个PR #{pr_number}...")
|
||
single_pr = await gitlink_utils.get_pull_request(owner, repo, pr_number)
|
||
if single_pr:
|
||
print(f" 成功获取PR: {single_pr.get('title')}")
|
||
else:
|
||
print(f" 获取PR失败")
|
||
|
||
# 3. 测试重新开启PR（如果有关闭的PR）
|
||
closed_prs = [pr for pr in prs if pr.get('state') == 'closed']
|
||
if closed_prs:
|
||
test_closed_pr = closed_prs[0]
|
||
pr_number = test_closed_pr.get('number')
|
||
print(f"3. 测试重新开启PR #{pr_number}...")
|
||
success = await gitlink_utils.reopen_pull_request(owner, repo, pr_number)
|
||
if success:
|
||
print(f" 成功重新开启PR")
|
||
else:
|
||
print(f" 重新开启PR失败")
|
||
else:
|
||
print("3. 没有关闭的PR可以测试重新开启")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 直接API测试失败: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
print("=" * 70)
|
||
|
||
async def main():
|
||
"""主测试函数"""
|
||
try:
|
||
# 执行主要测试
|
||
await test_gitlink_api_corrected()
|
||
|
||
# 执行直接API测试
|
||
await test_individual_gitlink_apis()
|
||
|
||
print("\n🎉 GitLink API修正测试完成!")
|
||
print("\n总结:")
|
||
print("- 根据官方API文档修正了所有GitLink API路径")
|
||
print("- 使用正确的HTTP方法（GET, POST, PUT）")
|
||
print("- 添加了获取单个PR的API")
|
||
print("- 验证了所有API的正确性")
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n⚠️ 测试被用户中断")
|
||
except Exception as e:
|
||
print(f"\n❌ 测试过程中发生错误: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main()) |