GitHub 仓库选择菜单：安装仓库列表 + PATCH 补仓库 + 控制台下拉
This commit is contained in:
host committed
1 parent
5ba4d35839
commit
6ba671e051
4 files changed
+343
No files matched your search
@@ -0,0 +1,170 @@
|
||||
"""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()
|
||||
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", "openqj/demo"])
|
||||
self.assertEqual(
|
||||
d["repos"][0]["clone_url"],
|
||||
"https://github.com/openqj/codex-app-server.git")
|
||||
|
||||
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()
|
||||
Reference in new issue
Block a user