feat: Use mirror git cloning

Readme.md

@@ -36,6 +36,12 @@ Example config is available in [config.example.toml](config.example.toml).
 Use `main.api_token` for Gitea API access. Use `fetch.ssh_key` for SSH
 cloning, or `fetch.user` and `fetch.token` for HTTP(S) cloning.
 
+Repositories are stored as bare mirror clones. On the first run the tool uses
+`git clone --mirror <url> <path>`, and on following runs it updates each mirror
+with `git remote update --prune`. This mirrors all refs and handles forced
+updates from the server side, but the resulting directories are not working
+tree checkouts.
+
 
 ### Native
 Not recommended, but more efficient in space

src/cli.py

@@ -2,10 +2,9 @@ import argparse
 import logging
 import os.path
 import sys
-from os import makedirs
 
 from .config import Config, read_toml_config
-from .git import git_clone, git_pull, http_git_env, ssh_git_env
+from .git import git_clone, git_update_mirror, http_git_env, ssh_git_env
 from .gitea_api import GiteaApi, GiteaApiError
 from .log_fields import (
     OPERATION,
@@ -43,6 +42,12 @@ def get_fetch_params(config: Config, repo: GiteaRepository):
     )
 
 
+def is_mirror_repository(path: str) -> bool:
+    return os.path.isfile(os.path.join(path, "config")) and os.path.isdir(
+        os.path.join(path, "objects")
+    )
+
+
 def process_repo(config: Config, repo: GiteaRepository) -> bool:
     out_path = get_repository_path(
         out_dir=config.out_dir,
@@ -56,15 +61,14 @@ def process_repo(config: Config, repo: GiteaRepository) -> bool:
         PATH: out_path,
         REPOSITORY_URL: repository_url,
     }
-    makedirs(out_path, exist_ok=True)
-    if os.path.exists(os.path.join(out_path, ".git")):
+    if is_mirror_repository(out_path):
         logger.info(
-            "repository pull started",
-            extra=log_context | {OPERATION: "pull"},
+            "repository mirror update started",
+            extra=log_context | {OPERATION: "update_mirror"},
         )
-        return git_pull(out_path, env=git_env)
+        return git_update_mirror(out_path, env=git_env)
     logger.info(
-        "repository clone started",
+        "repository mirror clone started",
         extra=log_context | {OPERATION: "clone"},
     )
     return git_clone(

src/git.py

@@ -3,7 +3,6 @@ import logging
 import os
 import shlex
 import subprocess
-from os import makedirs
 
 from .log_fields import EXIT_CODE, OPERATION, PATH, REPOSITORY_URL
 
@@ -34,12 +33,14 @@ def git_clone(
     repository: str,
     env: dict[str, str] | None = None,
 ) -> bool:
-    makedirs(repository, exist_ok=True)
+    repository_parent = os.path.dirname(repository)
+    if repository_parent:
+        os.makedirs(repository_parent, exist_ok=True)
     git_env = os.environ | (env or {})
     try:
         subprocess.check_call(
-            ["git", "clone", repository_url, "."],
-            cwd=repository,
+            ["git", "clone", "--mirror", repository_url, repository],
+            cwd=None,
             env=git_env,
         )
     except subprocess.CalledProcessError as err:
@@ -56,15 +57,22 @@ def git_clone(
     return True
 
 
-def git_pull(repository: str, env: dict[str, str] | None = None) -> bool:
+def git_update_mirror(
+    repository: str,
+    env: dict[str, str] | None = None,
+) -> bool:
     git_env = os.environ | (env or {})
     try:
-        subprocess.check_call(["git", "pull"], cwd=repository, env=git_env)
+        subprocess.check_call(
+            ["git", "remote", "update", "--prune"],
+            cwd=repository,
+            env=git_env,
+        )
     except subprocess.CalledProcessError as err:
         logger.error(
-            "git pull failed",
+            "git mirror update failed",
             extra={
-                OPERATION: "pull",
+                OPERATION: "update_mirror",
                 PATH: repository,
                 EXIT_CODE: err.returncode,
             },

tests/test_git.py

@@ -2,7 +2,7 @@ import base64
 import os
 import subprocess
 
-from src.git import git_clone, git_pull, http_git_env, ssh_git_env
+from src.git import git_clone, git_update_mirror, http_git_env, ssh_git_env
 
 
 def test_ssh_git_env_uses_requested_identity_file():
@@ -32,7 +32,7 @@ def test_http_git_env_uses_basic_auth_header_without_url_credentials():
     }
 
 
-def test_git_clone_passes_url_and_env_to_git(monkeypatch, tmp_path):
+def test_git_clone_uses_mirror_clone(monkeypatch, tmp_path):
     calls = []
     env = {"GIT_SSH_COMMAND": "ssh -i /tmp/test_key -o IdentitiesOnly=yes"}
 
@@ -48,13 +48,19 @@ def test_git_clone_passes_url_and_env_to_git(monkeypatch, tmp_path):
     )
 
     args, cwd, actual_env = calls[0]
-    assert args == ["git", "clone", "ssh://git@example.com/owner/repo.git", "."]
-    assert cwd == str(tmp_path)
+    assert args == [
+        "git",
+        "clone",
+        "--mirror",
+        "ssh://git@example.com/owner/repo.git",
+        str(tmp_path),
+    ]
+    assert cwd is None
     assert os.environ.items() <= actual_env.items()
     assert env.items() <= actual_env.items()
 
 
-def test_git_pull_passes_env_to_git(monkeypatch, tmp_path):
+def test_git_update_mirror_prunes_remote_refs(monkeypatch, tmp_path):
     calls = []
     env = {"GIT_CONFIG_COUNT": "1"}
 
@@ -63,10 +69,10 @@ def test_git_pull_passes_env_to_git(monkeypatch, tmp_path):
 
     monkeypatch.setattr(subprocess, "check_call", fake_check_call)
 
-    assert git_pull(repository=str(tmp_path), env=env)
+    assert git_update_mirror(repository=str(tmp_path), env=env)
 
     args, cwd, actual_env = calls[0]
-    assert args == ["git", "pull"]
+    assert args == ["git", "remote", "update", "--prune"]
     assert cwd == str(tmp_path)
     assert os.environ.items() <= actual_env.items()
     assert env.items() <= actual_env.items()

tests/test_mirror_process.py

@@ -79,16 +79,20 @@ def test_process_repo_clones_with_https_url_and_auth_env(monkeypatch, tmp_path):
     assert env["GIT_CONFIG_VALUE_0"].startswith("Authorization: Basic ")
 
 
-def test_process_repo_pulls_existing_repo_with_fetch_env(monkeypatch, tmp_path):
+def test_process_repo_updates_existing_mirror_with_fetch_env(
+    monkeypatch,
+    tmp_path,
+):
     calls = []
-    repo_path = tmp_path / "alex" / "repo" / ".git"
-    repo_path.mkdir(parents=True)
+    repo_path = tmp_path / "alex" / "repo"
+    (repo_path / "objects").mkdir(parents=True)
+    (repo_path / "config").write_text("[core]\nrepositoryformatversion = 0\n")
 
-    def fake_git_pull(repository, env):
+    def fake_git_update_mirror(repository, env):
         calls.append((repository, env))
         return True
 
-    monkeypatch.setattr(cli, "git_pull", fake_git_pull)
+    monkeypatch.setattr(cli, "git_update_mirror", fake_git_update_mirror)
 
     assert cli.process_repo(
         config=make_config(tmp_path, FetchConfig(ssh_key="/tmp/test_key")),
@@ -116,10 +120,15 @@ def test_process_repo_returns_false_when_clone_fails(monkeypatch, tmp_path):
     )
 
 
-def test_process_repo_returns_false_when_pull_fails(monkeypatch, tmp_path):
-    repo_path = tmp_path / "alex" / "repo" / ".git"
-    repo_path.mkdir(parents=True)
-    monkeypatch.setattr(cli, "git_pull", lambda repository, env: False)
+def test_process_repo_returns_false_when_update_fails(monkeypatch, tmp_path):
+    repo_path = tmp_path / "alex" / "repo"
+    (repo_path / "objects").mkdir(parents=True)
+    (repo_path / "config").write_text("[core]\nrepositoryformatversion = 0\n")
+    monkeypatch.setattr(
+        cli,
+        "git_update_mirror",
+        lambda repository, env: False,
+    )
 
     assert not cli.process_repo(
         config=make_config(tmp_path, FetchConfig(ssh_key="/tmp/test_key")),