forked from jkcl/reposync
238 lines
10 KiB
Python
238 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Gitlink PR单操作测试文件
|
||
用于测试单个PR操作，方便调试特定问题
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import sys
|
||
import os
|
||
|
||
# 添加项目根目录到Python路径
|
||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from src.utils.pr_sync import GitlinkPRUtils
|
||
from src.base import config
|
||
|
||
class GitlinkSingleOperationTester:
|
||
def __init__(self):
|
||
# 从配置文件获取token或cookie
|
||
self.token = config.ACCOUNT.get('gitlink_token', '')
|
||
if not self.token:
|
||
self.token = config.ACCOUNT.get('gitlink_cookie', '')
|
||
|
||
if not self.token:
|
||
print("❌ 错误：未找到Gitlink token或cookie配置")
|
||
sys.exit(1)
|
||
|
||
self.pr_utils = GitlinkPRUtils(self.token)
|
||
self.owner = "gonggong123zzz" # 替换为你的用户名
|
||
self.repo = "repo2" # 替换为你的仓库名
|
||
|
||
print(f"🔧 测试配置:")
|
||
print(f" Token/Cookie: {self.token[:20]}..." if len(self.token) > 20 else f" Token/Cookie: {self.token}")
|
||
print(f" Token类型: {'Cookie' if self.token.startswith('autologin_') else 'Bearer'}")
|
||
print(f" Owner: {self.owner}")
|
||
print(f" Repo: {self.repo}")
|
||
print()
|
||
|
||
async def test_create_pr_only(self):
|
||
"""仅测试创建PR"""
|
||
print("=" * 60)
|
||
print("🆕 测试创建PR")
|
||
print("=" * 60)
|
||
|
||
test_title = f"测试PR - {asyncio.get_event_loop().time()}"
|
||
test_body = "这是一个测试PR，用于验证API功能"
|
||
head_branch = "test" # 确保这个分支存在
|
||
base_branch = "main" # GitHub使用main分支，会自动映射为Gitlink的master分支
|
||
|
||
print(f"创建PR参数:")
|
||
print(f" Title: {test_title}")
|
||
print(f" Body: {test_body}")
|
||
print(f" Head: {head_branch}")
|
||
print(f" Base: {base_branch}")
|
||
|
||
try:
|
||
# 直接调用API获取详细响应
|
||
url = f"{self.pr_utils.base_url_api}/{self.owner}/{self.repo}/pulls.json"
|
||
data = {
|
||
"title": test_title,
|
||
"body": test_body,
|
||
"head": head_branch,
|
||
"base": base_branch
|
||
}
|
||
|
||
import aiohttp
|
||
import ssl
|
||
|
||
# 创建SSL连接器
|
||
ssl_context = ssl.create_default_context()
|
||
ssl_context.check_hostname = False
|
||
ssl_context.verify_mode = ssl.CERT_NONE
|
||
connector = aiohttp.TCPConnector(ssl=ssl_context)
|
||
|
||
async with aiohttp.ClientSession(connector=connector) as session:
|
||
async with session.post(url, headers=self.pr_utils.headers, json=data) as response:
|
||
print(f"\n📡 API请求详情:")
|
||
print(f" URL: {url}")
|
||
print(f" Method: POST")
|
||
print(f" Headers: {json.dumps(dict(self.pr_utils.headers), indent=2, ensure_ascii=False)}")
|
||
print(f" Request Body: {json.dumps(data, indent=2, ensure_ascii=False)}")
|
||
|
||
print(f"\n📡 API响应详情:")
|
||
print(f" Status Code: {response.status}")
|
||
print(f" Response Headers: {dict(response.headers)}")
|
||
|
||
response_text = await response.text()
|
||
print(f" Response Body: {response_text}")
|
||
|
||
if response.status in [200, 201]:
|
||
try:
|
||
result = json.loads(response_text)
|
||
print(f"\n✅ 创建PR成功!")
|
||
print(f" 解析后的响应: {json.dumps(result, indent=2, ensure_ascii=False)}")
|
||
return result
|
||
except json.JSONDecodeError:
|
||
print(f"❌ 响应不是有效的JSON格式")
|
||
return None
|
||
else:
|
||
print(f"❌ 创建PR失败，状态码: {response.status}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
print(f"❌ 创建PR异常: {str(e)}")
|
||
return None
|
||
|
||
async def test_reopen_pr_only(self, pr_number):
|
||
"""仅测试重新开启PR"""
|
||
print("=" * 60)
|
||
print(f"🔓 测试重新开启PR #{pr_number}")
|
||
print("=" * 60)
|
||
|
||
try:
|
||
# 直接调用API获取详细响应
|
||
url = f"{self.pr_utils.base_url_v1}/{self.owner}/{self.repo}/pulls/{pr_number}/reopen"
|
||
|
||
import aiohttp
|
||
import ssl
|
||
|
||
ssl_context = ssl.create_default_context()
|
||
ssl_context.check_hostname = False
|
||
ssl_context.verify_mode = ssl.CERT_NONE
|
||
connector = aiohttp.TCPConnector(ssl=ssl_context)
|
||
|
||
async with aiohttp.ClientSession(connector=connector) as session:
|
||
async with session.post(url, headers=self.pr_utils.headers) as response:
|
||
print(f"\n📡 API请求详情:")
|
||
print(f" URL: {url}")
|
||
print(f" Method: POST")
|
||
print(f" Headers: {json.dumps(dict(self.pr_utils.headers), indent=2, ensure_ascii=False)}")
|
||
print(f" Request Body: 无")
|
||
|
||
print(f"\n📡 API响应详情:")
|
||
print(f" Status Code: {response.status}")
|
||
print(f" Response Headers: {dict(response.headers)}")
|
||
|
||
response_text = await response.text()
|
||
print(f" Response Body: {response_text}")
|
||
|
||
if response.status in [200, 201]:
|
||
print(f"\n✅ 重新开启PR成功!")
|
||
try:
|
||
result = json.loads(response_text)
|
||
print(f" 解析后的响应: {json.dumps(result, indent=2, ensure_ascii=False)}")
|
||
return True
|
||
except json.JSONDecodeError:
|
||
print(f" 响应不是JSON格式，但状态码表示成功")
|
||
return True
|
||
else:
|
||
print(f"❌ 重新开启PR失败，状态码: {response.status}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 重新开启PR异常: {str(e)}")
|
||
return False
|
||
|
||
async def test_close_pr_only(self, pr_number):
|
||
"""仅测试关闭PR"""
|
||
print("=" * 60)
|
||
print(f"🔒 测试关闭PR #{pr_number}")
|
||
print("=" * 60)
|
||
|
||
try:
|
||
# 直接调用API获取详细响应
|
||
url = f"{self.pr_utils.base_url_api}/{self.owner}/{self.repo}/pulls/{pr_number}.json"
|
||
data = {"state": "closed"}
|
||
|
||
import aiohttp
|
||
import ssl
|
||
|
||
ssl_context = ssl.create_default_context()
|
||
ssl_context.check_hostname = False
|
||
ssl_context.verify_mode = ssl.CERT_NONE
|
||
connector = aiohttp.TCPConnector(ssl=ssl_context)
|
||
|
||
async with aiohttp.ClientSession(connector=connector) as session:
|
||
async with session.put(url, headers=self.pr_utils.headers, json=data) as response:
|
||
print(f"\n📡 API请求详情:")
|
||
print(f" URL: {url}")
|
||
print(f" Method: PUT")
|
||
print(f" Headers: {json.dumps(dict(self.pr_utils.headers), indent=2, ensure_ascii=False)}")
|
||
print(f" Request Body: {json.dumps(data, indent=2, ensure_ascii=False)}")
|
||
|
||
print(f"\n📡 API响应详情:")
|
||
print(f" Status Code: {response.status}")
|
||
print(f" Response Headers: {dict(response.headers)}")
|
||
|
||
response_text = await response.text()
|
||
print(f" Response Body: {response_text}")
|
||
|
||
if response.status in [200, 201]:
|
||
print(f"\n✅ 关闭PR成功!")
|
||
try:
|
||
result = json.loads(response_text)
|
||
print(f" 解析后的响应: {json.dumps(result, indent=2, ensure_ascii=False)}")
|
||
return True
|
||
except json.JSONDecodeError:
|
||
print(f" 响应不是JSON格式，但状态码表示成功")
|
||
return True
|
||
else:
|
||
print(f"❌ 关闭PR失败，状态码: {response.status}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 关闭PR异常: {str(e)}")
|
||
return False
|
||
|
||
async def main():
|
||
"""主函数"""
|
||
tester = GitlinkSingleOperationTester()
|
||
|
||
# 根据命令行参数选择测试类型
|
||
if len(sys.argv) > 1:
|
||
operation = sys.argv[1].lower()
|
||
|
||
if operation == "create":
|
||
await tester.test_create_pr_only()
|
||
elif operation == "reopen" and len(sys.argv) > 2:
|
||
pr_number = int(sys.argv[2])
|
||
await tester.test_reopen_pr_only(pr_number)
|
||
elif operation == "close" and len(sys.argv) > 2:
|
||
pr_number = int(sys.argv[2])
|
||
await tester.test_close_pr_only(pr_number)
|
||
else:
|
||
print("用法:")
|
||
print(" python test_gitlink_single_operation.py create")
|
||
print(" python test_gitlink_single_operation.py reopen <pr_number>")
|
||
print(" python test_gitlink_single_operation.py close <pr_number>")
|
||
else:
|
||
print("请指定要测试的操作:")
|
||
print(" python test_gitlink_single_operation.py create")
|
||
print(" python test_gitlink_single_operation.py reopen <pr_number>")
|
||
print(" python test_gitlink_single_operation.py close <pr_number>")
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main()) |