Merge pull request #158 from openqj/feat-repo-access-check

feat(repos): 仓库接入预检 + codexsvc deploy key 配给脚本

cloud/tests/test_project_repo.py

@@ -21,6 +21,7 @@ try:
     from cloud.auth import AuthStore
     from cloud.repos import (
         RepoError,
+        check_repo_access,
         checkout_dir,
         resolve_repo,
         validate_repo_url,
@@ -32,6 +33,7 @@ except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.p
     from cloud.auth import AuthStore
     from cloud.repos import (
         RepoError,
+        check_repo_access,
         checkout_dir,
         resolve_repo,
         validate_repo_url,
@@ -101,6 +103,50 @@ class ValidateRepoUrlTestCase(unittest.TestCase):
             validate_repo_url(123)
 
 
+class CheckRepoAccessTestCase(unittest.TestCase):
+    def test_ok_returns_head(self):
+        def runner(cmd, timeout=None):
+            self.assertEqual(cmd[:3], ["git", "ls-remote",
+                                       "git@github.com:o/r.git"])
+            return SimpleNamespace(returncode=0,
+                                   stdout="abc123\tHEAD\n", stderr="")
+        ok, detail = check_repo_access("git@github.com:o/r.git",
+                                       runner=runner)
+        self.assertTrue(ok)
+        self.assertEqual(detail, "abc123")
+
+    def test_failure_returns_reason_not_exception(self):
+        def runner(cmd, timeout=None):
+            return SimpleNamespace(returncode=128, stdout="",
+                                   stderr="fatal: Authentication failed")
+        ok, detail = check_repo_access("git@github.com:o/private.git",
+                                       runner=runner)
+        self.assertFalse(ok)
+        self.assertIn("Authentication failed", detail)
+
+    def test_https_failure_hints_ssh_form(self):
+        def runner(cmd, timeout=None):
+            return SimpleNamespace(returncode=128, stdout="",
+                                   stderr="fatal: unable to access")
+        ok, detail = check_repo_access("https://github.com/o/r.git",
+                                       runner=runner)
+        self.assertFalse(ok)
+        self.assertIn("git@", detail)
+
+    def test_invalid_url_returns_false(self):
+        ok, detail = check_repo_access("file:///tmp/r")
+        self.assertFalse(ok)
+        self.assertTrue(detail)
+
+    def test_timeout_returns_false(self):
+        def runner(cmd, timeout=None):
+            raise TimeoutError("timed out")
+        ok, detail = check_repo_access("git@github.com:o/r.git",
+                                       runner=runner)
+        self.assertFalse(ok)
+        self.assertIn("timed out", detail)
+
+
 class CheckoutDirTestCase(unittest.TestCase):
     def test_sanitizes_and_stays_inside(self):
         tmp = tempfile.TemporaryDirectory()