forked from jkcl/reposync
88 lines
3.4 KiB
Python
88 lines
3.4 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_simple():
|
||
"""简单测试双向同步功能"""
|
||
print("=== 测试双向同步功能 ===")
|
||
print("修改后的逻辑:")
|
||
print("1. 只同步源仓库中的开放状态PR")
|
||
print("2. 如果有重复的同名PR并且处于open状态就跳过")
|
||
print("3. 如果被同步仓库有关闭状态的同名的PR，直接开放")
|
||
print("4. 如果目标仓库没有相同名称的PR，就创建新的")
|
||
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}")
|
||
print()
|
||
|
||
try:
|
||
# 创建同步服务
|
||
sync_service = PRSyncService()
|
||
|
||
# 执行双向同步
|
||
print("开始执行双向同步...")
|
||
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:
|
||
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}")
|
||
elif action == 'reopened':
|
||
print(f" 🔄 重新开启PR: #{pr_number} - {title}")
|
||
elif action == 'closed':
|
||
print(f" ❌ 关闭PR: #{pr_number} - {title}")
|
||
elif action == 'failed':
|
||
print(f" ❌ 失败: #{pr_number} - {error}")
|
||
|
||
# 分析结果
|
||
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} 个操作失败")
|
||
if result.created_count == 0 and result.reopened_count == 0 and result.failed_count == 0:
|
||
print("✅ 双向同步完成，所有PR都已同步或已存在")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 测试失败: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(test_bidirectional_sync_simple()) |