"""POST /tasks/{id}/pr 的 App-token 接线单测（端点级）： App 路径：项目 repo_url 解析出仓库名、库里有该仓库的 App 安装记录 -> 现换 installation token（fake 换 token 客户端 + 测试专用 pem，JWT 纯标准库签发、不联网）-> push 走 credential helper、建 PR 走 GitHub API（fake），响应里 auth_via=app。 PAT 回落：没有安装记录 -> 走原来的 gh 路径，auth_via=pat。 git/GitHub API 一律 fake，不跑真实命令；token 值绝不出现 在 git 命令记录里。 """ import http.client import json import os import sqlite3 import sys import tempfile import threading import unittest from pathlib import Path from types import SimpleNamespace try: from cloud.auth import AuthStore from cloud.db import connect as _db_connect from cloud.github_app import ( record_installation_repos, save_app_config, ) from cloud.server import create_server from cloud.tests.test_app_token import TEST_KEY_PKCS1 except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.auth import AuthStore from cloud.db import connect as _db_connect from cloud.github_app import ( record_installation_repos, save_app_config, ) from cloud.server import create_server from cloud.tests.test_app_token import TEST_KEY_PKCS1 def _result(returncode=0, stdout="", stderr=""): return SimpleNamespace(returncode=returncode, stdout=stdout, stderr=stderr) class _Git: """假 git：remote/branch/add/commit/push，记录命令。""" def __init__(self, remote, branch): self.calls = [] self.remote = remote self.branch = branch 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, self.branch + "\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, "") class _Gh: """假 gh：auth 通过、pr create 回链接。""" def __init__(self, pr_url): self.calls = [] self.pr_url = pr_url def __call__(self, cmd, cwd): self.calls.append(list(cmd)) if cmd[:2] == ["gh", "auth"]: return _result(0, "", "") if cmd[:2] == ["gh", "pr"]: return _result(0, self.pr_url + "\n") return _result(0, "") class _Api: """假 GitHub API：GET 仓库回 default_branch，POST pulls 回 PR。""" def __init__(self): self.calls = [] def __call__(self, method, url, token, payload): self.calls.append((method, url, token, payload)) if method == "GET": return 200, {"default_branch": "main"} if method == "POST": return 201, {"html_url": "https://github.com/octo/a/pull/77", "number": 77} raise AssertionError("unexpected API call: %s %s" % (method, url)) def _token_client(method, url, headers, body): """假换 token：JWT 已在前面签好（只验格式），直接回 token。""" assert method == "POST", method assert "/access_tokens" in url, url assert headers["Authorization"].startswith("Bearer "), headers return 201, {"token": "srv-app-token", "expires_at": "2030-01-01T00:00:00Z", "permissions": {"contents": "write"}, "repository_selection": "selected"} class _Case(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") self.auth = AuthStore(self.db_path) self.owner_id, self.owner_key = self.auth.create_user("owner") self.server = create_server(self.db_path, "127.0.0.1", 0) self.port = self.server.server_address[1] self.thread = threading.Thread( target=self.server.serve_forever, daemon=True) self.thread.start() self.worktrees = os.path.join(self.tmp.name, "worktrees") os.makedirs(self.worktrees, exist_ok=True) self.server.host_worktrees_dir = self.worktrees self.git = None self.api = _Api() self.server.pr_github_api = self.api self.server.pr_app_token_client = _token_client def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=5) try: self.auth.conn.close() except Exception: pass self.tmp.cleanup() def _req(self, method, path, body=None, key=None): conn = http.client.HTTPConnection( "127.0.0.1", self.port, timeout=10) hdrs = {} payload = None if body is not None: payload = json.dumps(body).encode("utf-8") hdrs["Content-Type"] = "application/json" if key is not None: hdrs["X-API-Key"] = key try: conn.request(method, path, body=payload, headers=hdrs) resp = conn.getresponse() data = resp.read() finally: conn.close() try: parsed = json.loads(data.decode("utf-8")) if data else None except ValueError: parsed = None return resp.status, parsed def _project(self, name, repo_url): status, project = self._req( "POST", "/projects", {"name": name, "repo_url": repo_url, "auto_pr": 1}, self.owner_key) self.assertEqual(status, 201, project) return project["id"] def _task(self, task_id, project_id): conn = sqlite3.connect(self.db_path) try: conn.execute( "INSERT INTO tasks(id, project_id, prompt, status," " depends_on, priority, created_by, created_at," " updated_at, last_heartbeat) VALUES(?, ?, ?, 'merged'," " '[]', 0, 'system', 0, 0, NULL)", (task_id, project_id, "do it")) conn.commit() finally: conn.close() os.makedirs(os.path.join(self.worktrees, task_id), exist_ok=True) def test_app_path_via_endpoint(self): project_id = self._project( "octo/a", "https://github.com/octo/a") conn = _db_connect(self.db_path) try: save_app_config(conn, { "app_id": 5476837, "slug": "fake-app", "client_id": "Iv1.x", "webhook_secret": "fake-secret", "pem": TEST_KEY_PKCS1, "html_url": "https://github.com/apps/fake-app"}) record_installation_repos(conn, 175380082, "octo", ["octo/a"]) finally: conn.close() self.git = _Git("https://github.com/octo/a.git", "host/t-app-1") self.server.pr_git_runner = self.git self._task("t-app-1", project_id) status, body = self._req( "POST", "/tasks/t-app-1/pr", key=self.owner_key) self.assertEqual(status, 200, body) self.assertEqual(body["pr_url"], "https://github.com/octo/a/pull/77") self.assertEqual(body["pr_number"], 77) self.assertEqual(body["auth_via"], "app") pushes = [c for c in self.git.calls if "push" in c] self.assertEqual(len(pushes), 1) self.assertTrue(any(a.startswith("credential.helper=") for a in pushes[0])) self.assertNotIn("srv-app-token", " ".join(" ".join(c) for c in self.git.calls)) posts = [c for c in self.api.calls if c[0] == "POST"] self.assertEqual(len(posts), 1) self.assertEqual(posts[0][2], "srv-app-token") def test_pat_fallback_via_endpoint(self): project_id = self._project( "octo/b", "https://github.com/octo/b") self.git = _Git("https://github.com/octo/b.git", "host/t-pat-1") self.server.pr_git_runner = self.git gh = _Gh("https://github.com/octo/b/pull/88") self.server.pr_gh_runner = gh self.server.pr_gh_which = "/usr/bin/gh" self._task("t-pat-1", project_id) status, body = self._req( "POST", "/tasks/t-pat-1/pr", key=self.owner_key) self.assertEqual(status, 200, body) self.assertEqual(body["pr_url"], "https://github.com/octo/b/pull/88") self.assertEqual(body["auth_via"], "pat") pushes = [c for c in self.git.calls if "push" in c] self.assertEqual(len(pushes), 1) self.assertFalse(any(a.startswith("credential.helper=") for a in pushes[0])) self.assertEqual(self.api.calls, []) if __name__ == "__main__": unittest.main()