fix(git-gate): use smart http for smolmachines pushes

tests/unit/test_git_http_backend.py

@@ -0,0 +1,169 @@
+"""Unit: smart-HTTP git-gate wrapper."""
+
+import os
+import subprocess
+import tempfile
+import threading
+import unittest
+import urllib.request
+from pathlib import Path
+from unittest import mock
+
+from bot_bottle.git_http_backend import GitHttpHandler
+
+
+class TestGitHttpBackend(unittest.TestCase):
+    def test_real_git_push_reaches_bare_repo(self):
+        from http.server import ThreadingHTTPServer
+
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            bare = root / "repo.git"
+            subprocess.run(["git", "init", "--bare", str(bare)],
+                           check=True, capture_output=True, text=True)
+            subprocess.run(
+                ["git", "-C", str(bare), "config", "http.receivepack", "true"],
+                check=True,
+            )
+
+            old_root = os.environ.get("GIT_PROJECT_ROOT")
+            os.environ["GIT_PROJECT_ROOT"] = str(root)
+            self.addCleanup(self._restore_env, old_root)
+            old_hook = os.environ.get("GIT_GATE_ACCESS_HOOK")
+            hook = root / "access-hook"
+            hook.write_text("#!/bin/sh\nexit 0\n")
+            hook.chmod(0o700)
+            os.environ["GIT_GATE_ACCESS_HOOK"] = str(hook)
+            self.addCleanup(self._restore_hook, old_hook)
+
+            server = ThreadingHTTPServer(("127.0.0.1", 0), GitHttpHandler)
+            thread = threading.Thread(target=server.serve_forever, daemon=True)
+            thread.start()
+            self.addCleanup(server.shutdown)
+            self.addCleanup(server.server_close)
+
+            work = root / "work"
+            work.mkdir()
+            subprocess.run(["git", "init"], cwd=work, check=True,
+                           capture_output=True, text=True)
+            subprocess.run(["git", "config", "user.name", "test"],
+                           cwd=work, check=True)
+            subprocess.run(["git", "config", "user.email", "test@example.invalid"],
+                           cwd=work, check=True)
+            (work / "README.md").write_text("test\n")
+            subprocess.run(["git", "add", "README.md"], cwd=work, check=True)
+            subprocess.run(["git", "commit", "-m", "init"], cwd=work,
+                           check=True, capture_output=True, text=True)
+
+            url = f"http://127.0.0.1:{server.server_port}/repo.git"
+            subprocess.run(
+                ["git", "push", url, "HEAD:refs/heads/main"],
+                cwd=work,
+                check=True,
+                capture_output=True,
+                text=True,
+                timeout=5,
+            )
+
+            pushed = subprocess.check_output(
+                ["git", "-C", str(bare), "rev-parse", "refs/heads/main"],
+                text=True,
+            ).strip()
+            head = subprocess.check_output(
+                ["git", "-C", str(work), "rev-parse", "HEAD"],
+                text=True,
+            ).strip()
+            self.assertEqual(head, pushed)
+            subprocess.run(
+                ["git", "-C", str(bare), "symbolic-ref", "HEAD", "refs/heads/main"],
+                check=True,
+            )
+
+            clone = root / "clone"
+            subprocess.run(
+                ["git", "clone", url, str(clone)],
+                check=True,
+                capture_output=True,
+                text=True,
+                timeout=5,
+            )
+            cloned = subprocess.check_output(
+                ["git", "-C", str(clone), "rev-parse", "HEAD"],
+                text=True,
+            ).strip()
+            self.assertEqual(head, cloned)
+
+    def test_post_forwards_git_cgi_headers(self):
+        from http.server import ThreadingHTTPServer
+
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / "repo.git").mkdir()
+
+            old_root = os.environ.get("GIT_PROJECT_ROOT")
+            os.environ["GIT_PROJECT_ROOT"] = str(root)
+            self.addCleanup(self._restore_env, old_root)
+
+            server = ThreadingHTTPServer(("127.0.0.1", 0), GitHttpHandler)
+            thread = threading.Thread(target=server.serve_forever, daemon=True)
+            thread.start()
+            self.addCleanup(server.shutdown)
+            self.addCleanup(server.server_close)
+
+            backend_response = (
+                b"Status: 200 OK\r\n"
+                b"Content-Type: application/x-git-upload-pack-result\r\n"
+                b"\r\n"
+                b"0000"
+            )
+            calls = [
+                subprocess.CompletedProcess(["hook"], 0, b"", b""),
+                subprocess.CompletedProcess(["git"], 0, backend_response, b""),
+            ]
+            with mock.patch(
+                "bot_bottle.git_http_backend.subprocess.run",
+                side_effect=calls,
+            ) as run:
+                request = urllib.request.Request(
+                    f"http://127.0.0.1:{server.server_port}"
+                    "/repo.git/git-upload-pack",
+                    data=b"compressed",
+                    headers={
+                        "Accept": "application/x-git-upload-pack-result",
+                        "Content-Encoding": "gzip",
+                        "Content-Type": "application/x-git-upload-pack-request",
+                        "Git-Protocol": "version=2",
+                        "User-Agent": "git/test",
+                    },
+                    method="POST",
+                )
+                with urllib.request.urlopen(request, timeout=5) as response:
+                    self.assertEqual(200, response.status)
+                    self.assertEqual(b"0000", response.read())
+
+            env = run.call_args_list[1].kwargs["env"]
+            self.assertEqual("gzip", env["HTTP_CONTENT_ENCODING"])
+            self.assertEqual("version=2", env["HTTP_GIT_PROTOCOL"])
+            self.assertEqual(
+                "application/x-git-upload-pack-result",
+                env["HTTP_ACCEPT"],
+            )
+            self.assertEqual("git/test", env["HTTP_USER_AGENT"])
+
+    @staticmethod
+    def _restore_env(value: str | None) -> None:
+        if value is None:
+            os.environ.pop("GIT_PROJECT_ROOT", None)
+        else:
+            os.environ["GIT_PROJECT_ROOT"] = value
+
+    @staticmethod
+    def _restore_hook(value: str | None) -> None:
+        if value is None:
+            os.environ.pop("GIT_GATE_ACCESS_HOOK", None)
+        else:
+            os.environ["GIT_GATE_ACCESS_HOOK"] = value
+
+
+if __name__ == "__main__":
+    unittest.main()