""" Gitea API 功能测试脚本 测试内容： 1. 创建 PR 2. Merge PR (在日志中台直接合并) 3. Close PR (在日志中台直接拒绝) 4. 获取 PR 评论（close 原因） 5. 重新打开 PR (可选) 使用方法： python test_gitea_api.py --gitea-url https://gitea.xxx.com --token YOUR_TOKEN --owner owner --repo repo """ import argparse import httpx import json from typing import Optional class GiteaAPITester: """Gitea API 测试器""" def __init__(self, gitea_url: str, token: str, owner: str, repo: str): self.gitea_url = gitea_url.rstrip("/") self.token = token self.owner = owner self.repo = repo self.base_api_url = f"{self.gitea_url}/api/v1" self.client = httpx.Client(timeout=30) def _headers(self) -> dict: """生成请求头""" return { "Authorization": f"token {self.token}", "Content-Type": "application/json", } def test_create_pr(self, head_branch: str, base_branch: str = "main") -> Optional[int]: """ 测试 1: 创建 PR Returns: PR 编号，失败返回 None """ print("\n" + "=" * 60) print("测试 1: 创建 Pull Request") print("=" * 60) url = f"{self.base_api_url}/repos/{self.owner}/{self.repo}/pulls" payload = { "title": "[测试] Gitea API 测试 PR", "head": head_branch, "base": base_branch, "body": "这是一个测试 PR，用于验证 Gitea API 功能。\n\n## 测试内容\n- 创建 PR\n- Merge PR\n- Close PR", } try: response = self.client.post(url, json=payload, headers=self._headers()) response.raise_for_status() data = response.json() pr_number = data.get("number") pr_url = data.get("html_url") print(f"✅ PR 创建成功") print(f" PR 编号: #{pr_number}") print(f" PR URL: {pr_url}") print(f" 状态: {data.get('state')}") return pr_number except httpx.HTTPStatusError as e: print(f"❌ PR 创建失败: HTTP {e.response.status_code}") print(f" 错误信息: {e.response.text}") return None except Exception as e: print(f"❌ PR 创建失败: {e}") return None def test_get_pr_info(self, pr_number: int): """获取 PR 信息""" print("\n" + "=" * 60) print(f"获取 PR #{pr_number} 信息") print("=" * 60) url = f"{self.base_api_url}/repos/{self.owner}/{self.repo}/pulls/{pr_number}" try: response = self.client.get(url, headers=self._headers()) response.raise_for_status() data = response.json() print(f"✅ PR 信息获取成功") print(f" 标题: {data.get('title')}") print(f" 状态: {data.get('state')}") print(f" 是否已合并: {data.get('merged')}") print(f" 创建时间: {data.get('created_at')}") return data except Exception as e: print(f"❌ 获取 PR 信息失败: {e}") return None def test_merge_pr(self, pr_number: int) -> bool: """ 测试 2: 合并 PR（在日志中台直接操作） API: POST /api/v1/repos/{owner}/{repo}/pulls/{index}/merge Returns: 是否成功 """ print("\n" + "=" * 60) print(f"测试 2: 合并 PR #{pr_number}") print("=" * 60) url = f"{self.base_api_url}/repos/{self.owner}/{self.repo}/pulls/{pr_number}/merge" payload = { "Do": "merge", # merge / squash / rebase "MergeMessageField": f"Merge PR #{pr_number} via API", "MergeTitleField": f"Merge pull request #{pr_number}", } try: response = self.client.post(url, json=payload, headers=self._headers()) response.raise_for_status() print(f"✅ PR 合并成功") print(f" 方式: merge commit") print(f" 结论: ✅ Gitea 支持通过 API 直接合并 PR") return True except httpx.HTTPStatusError as e: print(f"❌ PR 合并失败: HTTP {e.response.status_code}") print(f" 错误信息: {e.response.text}") # 分析失败原因 if e.response.status_code == 405: print(f" 可能原因: PR 已经合并或已关闭") elif e.response.status_code == 409: print(f" 可能原因: 有合并冲突") elif e.response.status_code == 403: print(f" 可能原因: Token 权限不足") return False except Exception as e: print(f"❌ PR 合并失败: {e}") return False def test_close_pr(self, pr_number: int, reason: str = "") -> bool: """ 测试 3: 关闭 PR（在日志中台直接拒绝） API: PATCH /api/v1/repos/{owner}/{repo}/pulls/{index} Returns: 是否成功 """ print("\n" + "=" * 60) print(f"测试 3: 关闭 PR #{pr_number}") print("=" * 60) # Step 1: 添加关闭原因评论 if reason: comment_success = self.add_pr_comment(pr_number, reason) if comment_success: print(f"✅ 已添加关闭原因评论") # Step 2: 关闭 PR url = f"{self.base_api_url}/repos/{self.owner}/{self.repo}/pulls/{pr_number}" payload = { "state": "closed", # open / closed } try: response = self.client.patch(url, json=payload, headers=self._headers()) response.raise_for_status() print(f"✅ PR 关闭成功") print(f" 结论: ✅ Gitea 支持通过 API 直接关闭 PR") return True except httpx.HTTPStatusError as e: print(f"❌ PR 关闭失败: HTTP {e.response.status_code}") print(f" 错误信息: {e.response.text}") return False except Exception as e: print(f"❌ PR 关闭失败: {e}") return False def add_pr_comment(self, pr_number: int, comment: str) -> bool: """ 添加 PR 评论（用于记录 close 原因） API: POST /api/v1/repos/{owner}/{repo}/issues/{index}/comments 注意: Gitea 中 PR 和 Issue 共用评论 API """ url = f"{self.base_api_url}/repos/{self.owner}/{self.repo}/issues/{pr_number}/comments" payload = { "body": comment, } try: response = self.client.post(url, json=payload, headers=self._headers()) response.raise_for_status() return True except Exception as e: print(f"⚠️ 添加评论失败: {e}") return False def test_get_pr_comments(self, pr_number: int) -> list[dict]: """ 测试 4: 获取 PR 评论（获取 close 原因） API: GET /api/v1/repos/{owner}/{repo}/issues/{index}/comments Returns: 评论列表 """ print("\n" + "=" * 60) print(f"测试 4: 获取 PR #{pr_number} 的评论") print("=" * 60) url = f"{self.base_api_url}/repos/{self.owner}/{self.repo}/issues/{pr_number}/comments" try: response = self.client.get(url, headers=self._headers()) response.raise_for_status() comments = response.json() print(f"✅ 获取评论成功，共 {len(comments)} 条") for i, comment in enumerate(comments, 1): print(f"\n 评论 #{i}:") print(f" - 作者: {comment.get('user', {}).get('login')}") print(f" - 时间: {comment.get('created_at')}") print(f" - 内容: {comment.get('body')[:100]}...") print(f"\n 结论: ✅ 可以通过评论获取 PR 关闭原因") return comments except Exception as e: print(f"❌ 获取评论失败: {e}") return [] def test_reopen_pr(self, pr_number: int) -> bool: """ 测试 5: 重新打开 PR（可选） API: PATCH /api/v1/repos/{owner}/{repo}/pulls/{index} """ print("\n" + "=" * 60) print(f"测试 5: 重新打开 PR #{pr_number}") print("=" * 60) url = f"{self.base_api_url}/repos/{self.owner}/{self.repo}/pulls/{pr_number}" payload = { "state": "open", } try: response = self.client.patch(url, json=payload, headers=self._headers()) response.raise_for_status() print(f"✅ PR 重新打开成功") print(f" 结论: ✅ Gitea 支持重新打开已关闭的 PR") return True except Exception as e: print(f"❌ 重新打开 PR 失败: {e}") return False def run_all_tests(self, test_pr_number: Optional[int] = None): """运行所有测试""" print("\n") print("╔" + "=" * 58 + "╗") print("║" + " " * 15 + "Gitea API 功能测试" + " " * 23 + "║") print("╚" + "=" * 58 + "╝") print(f"\n配置信息:") print(f" Gitea URL: {self.gitea_url}") print(f" 仓库: {self.owner}/{self.repo}") print(f" Token: {'*' * 20} (已配置)") if test_pr_number: # 使用已存在的 PR 进行测试 pr_number = test_pr_number print(f"\n使用已存在的 PR #{pr_number} 进行测试") # 获取 PR 信息 self.test_get_pr_info(pr_number) else: # 创建新 PR 进行测试（需要先在仓库中创建测试分支） print("\n⚠️ 提示: 需要先在仓库中创建一个测试分支（如 test-gitea-api）") test_branch = input("请输入测试分支名称（或直接回车跳过创建 PR）: ").strip() if test_branch: pr_number = self.test_create_pr(test_branch) if not pr_number: print("\n❌ 测试终止：无法创建 PR") return else: pr_number = int(input("请输入已存在的 PR 编号: ")) # 测试关闭 PR（带原因） close_reason = """## ❌ PR 被拒绝 **拒绝原因:** 1. 测试覆盖不足，缺少边界条件测试 2. 修改了核心业务逻辑，但未添加集成测试 3. 代码中存在潜在的 null 指针风险 **建议:** - 补充测试用例 - 重新审核业务逻辑 - 修复后重新提交 --- **操作人:** 审核人员 (通过日志中台) """ self.test_close_pr(pr_number, close_reason) # 获取评论（验证可以获取 close 原因） comments = self.test_get_pr_comments(pr_number) # 可选：重新打开 PR reopen = input("\n是否测试重新打开 PR？(y/n): ").strip().lower() if reopen == "y": self.test_reopen_pr(pr_number) # 可选：测试合并 PR if reopen == "y": merge = input("是否测试合并 PR？(y/n): ").strip().lower() if merge == "y": self.test_merge_pr(pr_number) print("\n" + "=" * 60) print("测试总结") print("=" * 60) print("✅ Gitea 支持的功能:") print(" 1. ✅ 通过 API 创建 PR") print(" 2. ✅ 通过 API 合并 PR（在日志中台直接操作）") print(" 3. ✅ 通过 API 关闭 PR（在日志中台直接操作）") print(" 4. ✅ 通过 API 添加评论（记录 close 原因）") print(" 5. ✅ 通过 API 获取评论（读取 close 原因）") print(" 6. ✅ 通过 API 重新打开 PR") print("\n结论: Gitea 完全支持在日志中台直接 merge/close PR！") def main(): parser = argparse.ArgumentParser(description="Gitea API 功能测试") parser.add_argument("--gitea-url", required=True, help="Gitea 服务器地址") parser.add_argument("--token", required=True, help="Gitea API Token") parser.add_argument("--owner", required=True, help="仓库所有者") parser.add_argument("--repo", required=True, help="仓库名称") parser.add_argument("--pr-number", type=int, help="测试用的 PR 编号（可选）") args = parser.parse_args() tester = GiteaAPITester( gitea_url=args.gitea_url, token=args.token, owner=args.owner, repo=args.repo, ) tester.run_all_tests(test_pr_number=args.pr_number) if __name__ == "__main__": # 示例用法 print("示例用法:") print("python test_gitea_api.py \\") print(" --gitea-url https://gitea.airlabs.art \\") print(" --token your_token_here \\") print(" --owner airlabs \\") print(" --repo rtc_backend \\") print(" --pr-number 45") print("\n") main()