App-token：推分支建 PR 改走 GitHub App installation token
- 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 个
This commit is contained in:
host committed
1 parent
ee4b01c8f6
commit
25d9b3d938
7 files changed
+1060
-20
No files matched your search
@@ -0,0 +1,176 @@
|
||||
"""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()
|
||||
Reference in new issue
Block a user