host
25d9b3d938
- github_app：纯标准库 RS256 签 App JWT（PKCS#1/PKCS#8）， 换 installation token（现换现用、不缓存），按仓库名查安装 记录，任一步失败回落 None - pr：app_token 路径 push 走 credential helper（token 只进 子进程环境），建 PR 走 REST API（读 default_branch 当 base）； 不传 token 时 gh 路径不变 - server：/pr 端点按项目 repo_url 解析仓库名、现换 token， 响应加 auth_via=app|pat，换失败回落 PAT 不报错 - 测试：JWT/换 token/解析 9 个、pr App 路径 6 个、端点级 App 与 PAT 回落 2 个
177 lines
6.5 KiB
Python
177 lines
6.5 KiB
Python
"""pr.py 的 App-token 路径单测：push 走 credential helper、建 PR
|
||
走 GitHub API、gh 完全不被调用；remote 解析与失败分支。
|
||
|
||
不联网、不跑真实 git/gh：git 用假 runner，GitHub API 用假
|
||
客户端；token 值绝不出现在记录的命令里（只出现变量名）。
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
try:
|
||
from cloud.pr import (
|
||
PrStepError,
|
||
check_pr_ready,
|
||
create_task_pr,
|
||
parse_git_remote_owner_repo,
|
||
)
|
||
except ImportError:
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.pr import (
|
||
PrStepError,
|
||
check_pr_ready,
|
||
create_task_pr,
|
||
parse_git_remote_owner_repo,
|
||
)
|
||
|
||
|
||
def _result(returncode=0, stdout="", stderr=""):
|
||
return SimpleNamespace(returncode=returncode, stdout=stdout,
|
||
stderr=stderr)
|
||
|
||
|
||
class _Git:
|
||
"""假 git runner：remote/branch/add/commit/push 全按脚本回。"""
|
||
|
||
def __init__(self, remote="https://github.com/octo/a.git"):
|
||
self.calls = []
|
||
self.remote = remote
|
||
|
||
def __call__(self, cmd, cwd):
|
||
self.calls.append(list(cmd))
|
||
if "remote" in cmd:
|
||
return _result(0, self.remote + "\n")
|
||
if "branch" in cmd:
|
||
return _result(0, "host/t1\n")
|
||
if "add" in cmd:
|
||
return _result(0, "")
|
||
if "commit" in cmd:
|
||
return _result(1, "",
|
||
"nothing to commit, working tree clean")
|
||
if "push" in cmd:
|
||
return _result(0, "")
|
||
return _result(0, "")
|
||
|
||
|
||
def _gh_forbidden(cmd, cwd):
|
||
raise AssertionError("App 路径不应调用 gh：%r" % (cmd,))
|
||
|
||
|
||
class _FakeGithubApi:
|
||
"""假 GitHub API：记录调用，GET 仓库回 default_branch，
|
||
POST pulls 回 PR 链接；失败模式可配。"""
|
||
|
||
def __init__(self, default_branch="main", pulls_status=201):
|
||
self.calls = []
|
||
self.default_branch = default_branch
|
||
self.pulls_status = pulls_status
|
||
|
||
def __call__(self, method, url, token, payload):
|
||
self.calls.append((method, url, token, payload))
|
||
if method == "GET" and url.endswith("/repos/octo/a"):
|
||
return 200, {"default_branch": self.default_branch}
|
||
if method == "POST" and url.endswith("/repos/octo/a/pulls"):
|
||
if self.pulls_status != 201:
|
||
return self.pulls_status, {"message": "Validation Failed"}
|
||
return 201, {"html_url": "https://github.com/octo/a/pull/42",
|
||
"number": 42}
|
||
raise AssertionError("unexpected API call: %s %s" % (method, url))
|
||
|
||
|
||
def _worktree(tmp):
|
||
d = os.path.join(tmp, "t1")
|
||
os.makedirs(d, exist_ok=True)
|
||
return d
|
||
|
||
|
||
class ParseRemoteTestCase(unittest.TestCase):
|
||
def test_variants(self):
|
||
self.assertEqual(
|
||
parse_git_remote_owner_repo(
|
||
"https://github.com/octo/a.git"), ("octo", "a"))
|
||
self.assertEqual(
|
||
parse_git_remote_owner_repo(
|
||
"https://github.com/octo/a"), ("octo", "a"))
|
||
self.assertEqual(
|
||
parse_git_remote_owner_repo(
|
||
"https://user@github.com/octo/a.git"), ("octo", "a"))
|
||
self.assertEqual(
|
||
parse_git_remote_owner_repo(
|
||
"git@github.com:octo/a.git"), ("octo", "a"))
|
||
self.assertEqual(
|
||
parse_git_remote_owner_repo(
|
||
"ssh://git@github.com/octo/a.git"), ("octo", "a"))
|
||
|
||
def test_non_github_is_none(self):
|
||
self.assertIsNone(parse_git_remote_owner_repo(
|
||
"https://example.com/octo/a.git"))
|
||
self.assertIsNone(parse_git_remote_owner_repo(""))
|
||
self.assertIsNone(parse_git_remote_owner_repo(None))
|
||
|
||
|
||
class AppPathTestCase(unittest.TestCase):
|
||
def test_push_uses_credential_helper_and_pr_via_api(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
_worktree(tmp)
|
||
git = _Git()
|
||
api = _FakeGithubApi()
|
||
result = create_task_pr(
|
||
"t1", 1, worktrees_root=tmp, git_runner=git,
|
||
gh_runner=_gh_forbidden, gh_which="/nonexistent/gh",
|
||
app_token="fake-app-token", github_api=api)
|
||
self.assertEqual(
|
||
result["pr_url"], "https://github.com/octo/a/pull/42")
|
||
self.assertEqual(result["pr_number"], 42)
|
||
self.assertEqual(result["branch"], "host/t1")
|
||
pushes = [c for c in git.calls if "push" in c]
|
||
self.assertEqual(len(pushes), 1)
|
||
helper = [a for a in pushes[0]
|
||
if a.startswith("credential.helper=")][0]
|
||
self.assertIn("$GIT_TOKEN", helper)
|
||
self.assertNotIn("fake-app-token", " ".join(pushes[0]))
|
||
posts = [c for c in api.calls if c[0] == "POST"]
|
||
self.assertEqual(len(posts), 1)
|
||
self.assertEqual(posts[0][2], "fake-app-token")
|
||
self.assertEqual(posts[0][3]["head"], "host/t1")
|
||
self.assertEqual(posts[0][3]["base"], "main")
|
||
self.assertEqual(posts[0][3]["title"], "cloud: t1")
|
||
|
||
def test_app_path_skips_gh_checks(self):
|
||
# gh_which 指到不存在的路径、gh_runner 会炸：App 路径下
|
||
# check_pr_ready 仍应通过（只查 worktree + remote）。
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
_worktree(tmp)
|
||
self.assertTrue(check_pr_ready(
|
||
os.path.join(tmp, "t1"), git_runner=_Git(),
|
||
gh_which="/nonexistent/gh", gh_runner=_gh_forbidden,
|
||
app_token="x"))
|
||
|
||
def test_non_github_remote_fails_at_push(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
_worktree(tmp)
|
||
with self.assertRaises(PrStepError) as ctx:
|
||
create_task_pr(
|
||
"t1", 1, worktrees_root=tmp,
|
||
git_runner=_Git("https://example.com/octo/a.git"),
|
||
gh_runner=_gh_forbidden, app_token="x",
|
||
github_api=_FakeGithubApi())
|
||
self.assertEqual(ctx.exception.step, "push")
|
||
|
||
def test_api_failure_is_pr_create_step(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
_worktree(tmp)
|
||
with self.assertRaises(PrStepError) as ctx:
|
||
create_task_pr(
|
||
"t1", 1, worktrees_root=tmp, git_runner=_Git(),
|
||
gh_runner=_gh_forbidden, app_token="x",
|
||
github_api=_FakeGithubApi(pulls_status=422))
|
||
self.assertEqual(ctx.exception.step, "pr_create")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|