"""GitHub 仓库选择菜单单测：安装仓库列表 + 项目 PATCH 补仓库。 不跑真实模型回合、不联网：server 侧走本地临时端口 + 临时 SQLite；安装仓库行直接由 record_installation_repos 写入测试库 （模拟 installation_repositories webhook 同步），验证列表端点、 PATCH 权限与校验、以及截图里的死结——无仓库项目建环境 400， PATCH 补仓库后建环境 201。 """ import http.client import json import os import sqlite3 import sys import tempfile import threading import unittest from pathlib import Path try: from cloud.auth import AuthStore from cloud.github_app import record_installation_repos from cloud.server import create_server except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.auth import AuthStore from cloud.github_app import record_installation_repos from cloud.server import create_server def _req(port, method, path, body=None, key=None): conn = http.client.HTTPConnection("127.0.0.1", 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 class ServerGithubReposTestCase(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.other_id, self.other_key = self.auth.create_user("other") 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() s, proj = _req(self.port, "POST", "/projects", {"name": "repoproj"}, self.owner_key) self.assertEqual(s, 201, proj) self.project_id = proj["id"] def tearDown(self): self.server.shutdown() self.server.server_close() self.tmp.cleanup() def _seed_installations(self): conn = sqlite3.connect(self.db_path) try: record_installation_repos( conn, 175380082, "openqj", ["openqj/codex-app-server", "openqj/demo"]) finally: conn.close() def _project_repo(self, key): s, d = _req(self.port, "GET", "/projects", key=key) self.assertEqual(s, 200, d) for p in d["projects"]: if p["id"] == self.project_id: return p["repo_url"] self.fail("project not in list") def test_list_requires_auth(self): s, d = _req(self.port, "GET", "/github/repos") self.assertEqual(s, 401, d) def test_list_returns_seeded_installations(self): self._seed_installations() # P1：repos 只返回调用者成员项目对应的仓库；先把项目仓库 # 绑到其中一个已安装仓库，再断言只返回该仓库，非成员为空。 s, d = _req(self.port, "PATCH", "/projects/" + self.project_id, {"repo_url": "https://github.com/openqj/codex-app-server.git"}, self.owner_key) self.assertEqual(s, 200, d) s, d = _req(self.port, "GET", "/github/repos", key=self.owner_key) self.assertEqual(s, 200, d) names = [r["repo_full_name"] for r in d["repos"]] self.assertEqual(names, ["openqj/codex-app-server"]) self.assertEqual( d["repos"][0]["clone_url"], "https://github.com/openqj/codex-app-server.git") s, d = _req(self.port, "GET", "/github/repos", key=self.other_key) self.assertEqual(s, 200, d) self.assertEqual(d["repos"], []) def test_patch_repo_url_owner(self): url = "https://github.com/openqj/demo.git" s, d = _req(self.port, "PATCH", "/projects/" + self.project_id, {"repo_url": url}, self.owner_key) self.assertEqual(s, 200, d) self.assertEqual(d["project"]["repo_url"], url) self.assertEqual(self._project_repo(self.owner_key), url) def test_patch_forbidden_and_validation(self): url = "https://github.com/openqj/demo.git" s, d = _req(self.port, "PATCH", "/projects/" + self.project_id, {"repo_url": url}, self.other_key) self.assertEqual(s, 403, d) s, d = _req(self.port, "PATCH", "/projects/" + self.project_id, {"repo_url": "not a url"}, self.owner_key) self.assertEqual(s, 400, d) s, d = _req(self.port, "PATCH", "/projects/" + self.project_id, {"repo_url": "https://user:pass@github.com/a/b.git"}, self.owner_key) self.assertEqual(s, 400, d) s, d = _req(self.port, "PATCH", "/projects/" + self.project_id, {}, self.owner_key) self.assertEqual(s, 400, d) s, d = _req(self.port, "PATCH", "/projects/no-such-project", {"repo_url": url}, self.owner_key) self.assertEqual(s, 404, d) def test_patch_clears_repo_url(self): url = "https://github.com/openqj/demo.git" s, _ = _req(self.port, "PATCH", "/projects/" + self.project_id, {"repo_url": url}, self.owner_key) self.assertEqual(s, 200) s, d = _req(self.port, "PATCH", "/projects/" + self.project_id, {"repo_url": None}, self.owner_key) self.assertEqual(s, 200, d) self.assertIsNone(d["project"]["repo_url"]) def test_env_blocked_then_unblocked_by_patch(self): # 截图场景：无仓库项目建环境 400。 s, d = _req(self.port, "POST", "/environments", {"project_id": self.project_id, "name": "env1"}, self.owner_key) self.assertEqual(s, 400, d) self.assertIn("no repo_url", d["error"]) # PATCH 补仓库后同一请求 201。 s, _ = _req(self.port, "PATCH", "/projects/" + self.project_id, {"repo_url": "https://github.com/openqj/demo.git"}, self.owner_key) self.assertEqual(s, 200) s, d = _req(self.port, "POST", "/environments", {"project_id": self.project_id, "name": "env1"}, self.owner_key) self.assertEqual(s, 201, d) if __name__ == "__main__": unittest.main()