forked from jkcl/reposync
67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
import asyncio
|
||
import sys
|
||
import os
|
||
|
||
# 添加项目根目录到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_bidirectional_sync():
|
||
"""测试双向同步功能"""
|
||
print("=== 测试双向同步功能 ===")
|
||
|
||
# 创建同步请求
|
||
request = PRSyncRequest(
|
||
source_project="https://www.gitlink.org.cn/gonggong123zzz/repo2",
|
||
target_project="https://github.com/demon-two/repo1",
|
||
source_type="gitlink",
|
||
target_type="github",
|
||
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}")
|
||
|
||
try:
|
||
# 创建同步服务
|
||
sync_service = PRSyncService()
|
||
|
||
# 执行双向同步
|
||
result = await sync_service.sync_pull_requests_bidirectional(request)
|
||
|
||
print(f"\n同步结果: {result.message}")
|
||
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详细结果:")
|
||
for detail in result.details:
|
||
print(f" {detail}")
|
||
|
||
# 分析结果
|
||
if result.created_count > 0:
|
||
print(f"\n✅ 成功创建了 {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} 个操作失败")
|
||
if result.created_count == 0 and result.reopened_count == 0 and result.failed_count == 0:
|
||
print("\n✅ 双向同步完成，所有PR都已同步或已存在")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 测试失败: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(test_bidirectional_sync()) |