feat: 仓库接入通用化——批量预检 + 多仓 ssh 别名配给

This commit is contained in:
host committed 2026-10-30 04:18:09 +08:00
1 parent 0787496427
commit 50ba8d9297
3 files changed
+77 -3

No files matched your search

+25 -1
View File
@@ -38,6 +38,7 @@ __all__ = [
"checkout_dir",
"resolve_repo",
"check_repo_access",
"check_repo_access_many",
]
@@ -203,10 +204,33 @@ def check_repo_access(repo_url, runner=None, timeout=30):
return (True, head[0] if head else "")
def check_repo_access_many(repo_urls, runner=None, timeout=30):
"""批量预检:逐个调 check_repo_access,返回 {url: (ok, detail)}。
顺序与输入一致;单个失败不中断其余(批量预检的意义就是一次
看清哪些仓通、哪些不通)。空列表返回空 dict。
"""
out = {}
for url in repo_urls or []:
out[url] = check_repo_access(url, runner=runner, timeout=timeout)
return out
def _main(argv):
if len(argv) >= 3 and argv[1] == "check-many":
results = check_repo_access_many(argv[2:])
failed = 0
for url, (ok, detail) in results.items():
if ok:
print("ok %s HEAD=%s" % (url, detail))
else:
failed += 1
print("fail %s: %s" % (url, detail))
return 1 if failed else 0
if len(argv) != 3 or argv[1] != "check":
sys.stderr.write(
"用法: python3 -m cloud.repos check <repo_url>\n")
"用法: python3 -m cloud.repos check <repo_url>\n"
" python3 -m cloud.repos check-many <repo_url>...\n")
return 2
ok, detail = check_repo_access(argv[2])
if ok:
+20
View File
@@ -146,6 +146,26 @@ class CheckRepoAccessTestCase(unittest.TestCase):
self.assertFalse(ok)
self.assertIn("timed out", detail)
def test_many_continues_after_failure(self):
seen = []
def runner(cmd, timeout=None):
seen.append(cmd[2])
if "good" in cmd[2]:
return SimpleNamespace(returncode=0,
stdout="abc123\tHEAD\n", stderr="")
return SimpleNamespace(returncode=128, stdout="",
stderr="fatal: denied")
from cloud.repos import check_repo_access_many
out = check_repo_access_many(
["git@github.com:o/good.git", "git@github.com:o/bad.git"],
runner=runner)
self.assertEqual(seen, ["git@github.com:o/good.git",
"git@github.com:o/bad.git"])
self.assertTrue(out["git@github.com:o/good.git"][0])
self.assertFalse(out["git@github.com:o/bad.git"][0])
self.assertEqual(check_repo_access_many([]), {})
class CheckoutDirTestCase(unittest.TestCase):
def test_sanitizes_and_stays_inside(self):