forked from jkcl/reposync
206 lines
7.9 KiB
Python
206 lines
7.9 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_reopen_fix():
|
||
"""测试修复后的GitLink重新开启API"""
|
||
print("=" * 70)
|
||
print("测试修复后的GitLink重新开启API")
|
||
print("=" * 70)
|
||
print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print()
|
||
|
||
print("修复内容:")
|
||
print("1. 修正GitLink重新开启PR的API端点")
|
||
print("2. 从 PATCH /api/v1/{owner}/{repo}/pulls/{pr_number} 改为 POST /api/v1/{owner}/{repo}/pulls/{pr_number}/reopen.json")
|
||
print("3. 移除不必要的请求体数据")
|
||
print()
|
||
|
||
# 测试配置 - 使用GitHub到GitLink的同步来测试重新开启功能
|
||
test_configs = [
|
||
{
|
||
"name": "GitHub -> GitLink 同步测试（测试重新开启功能）",
|
||
"source_project": "https://github.com/demon-two/repo1",
|
||
"target_project": "https://www.gitlink.org.cn/gonggong123zzz/repo2",
|
||
"source_type": "github",
|
||
"target_type": "gitlink"
|
||
}
|
||
]
|
||
|
||
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("开始执行双向同步（测试重新开启功能）...")
|
||
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")
|
||
print(f" 🎉 GitLink重新开启API修复成功！")
|
||
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_direct_gitlink_api():
|
||
"""直接测试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")
|
||
|
||
# 测试重新开启一个已知的PR
|
||
owner = "gonggong123zzz"
|
||
repo = "repo2"
|
||
pr_number = 11075 # 从日志中看到的PR编号
|
||
|
||
print(f"测试重新开启GitLink PR: {owner}/{repo}/pulls/{pr_number}")
|
||
print(f"API端点: https://www.gitlink.org.cn/api/v1/{owner}/{repo}/pulls/{pr_number}/reopen.json")
|
||
print()
|
||
|
||
# 尝试重新开启PR
|
||
success = await gitlink_utils.reopen_pull_request(owner, repo, pr_number)
|
||
|
||
if success:
|
||
print("✅ GitLink重新开启API测试成功！")
|
||
else:
|
||
print("❌ GitLink重新开启API测试失败")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 直接API测试失败: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
print("=" * 70)
|
||
|
||
async def main():
|
||
"""主测试函数"""
|
||
try:
|
||
# 执行主要测试
|
||
await test_gitlink_reopen_fix()
|
||
|
||
# 执行直接API测试
|
||
await test_direct_gitlink_api()
|
||
|
||
print("\n🎉 GitLink重新开启API修复测试完成!")
|
||
print("\n总结:")
|
||
print("- 修正了GitLink重新开启PR的API端点")
|
||
print("- 使用正确的POST方法而不是PATCH")
|
||
print("- 移除了不必要的请求体数据")
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n⚠️ 测试被用户中断")
|
||
except Exception as e:
|
||
print(f"\n❌ 测试过程中发生错误: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main()) |