fix(core): advance git cache HEAD on refresh without ref

Closes #214.

Author: ulises-jeremias

packages/create-python-app-core/src/create_python_app_core/git_cache.py

@@ -103,6 +103,50 @@ def _run_git(args: list[str], *, cwd: Path | None = None) -> str:
         raise CpaError("git executable not found", code="CPA_GIT") from exc
 
 
+def _remote_default_ref(entry: Path) -> str:
+    """Resolve origin/HEAD (e.g. origin/main) for an existing cache clone."""
+    try:
+        sym = _run_git(["symbolic-ref", "refs/remotes/origin/HEAD"], cwd=entry)
+        if sym.startswith("refs/remotes/"):
+            return sym.removeprefix("refs/remotes/")
+        if sym.startswith("origin/"):
+            return sym
+    except CpaError:
+        pass
+    for candidate in ("origin/main", "origin/master"):
+        try:
+            _run_git(["rev-parse", "--verify", candidate], cwd=entry)
+            return candidate
+        except CpaError:
+            continue
+    raise CpaError(
+        "unable to resolve remote default branch for cache refresh",
+        code="CPA_GIT",
+    )
+
+
+def _refresh_cached_repo(entry: Path, ref: str | None) -> str:
+    """Fetch and hard-reset the cache clone to the remote tip (CNA pull parity)."""
+    _run_git(["fetch", "--all", "--tags"], cwd=entry)
+    if ref:
+        remote = ref if ref.startswith(("refs/", "origin/")) else f"origin/{ref}"
+        local_branch = ref.rsplit("/", 1)[-1]
+        _run_git(["checkout", "--force", "-B", local_branch, remote], cwd=entry)
+        _run_git(["reset", "--hard", remote], cwd=entry)
+    else:
+        remote = _remote_default_ref(entry)
+        local_branch = remote.rsplit("/", 1)[-1]
+        _run_git(["checkout", "--force", "-B", local_branch, remote], cwd=entry)
+        _run_git(["reset", "--hard", remote], cwd=entry)
+    return _run_git(["rev-parse", "HEAD"], cwd=entry)
+
+
+def _subdir_missing(entry: Path, source: ResolvedSource) -> bool:
+    if not source.subdir:
+        return False
+    return not (entry / source.subdir).is_dir()
+
+
 def download_repository(
     source: ResolvedSource,
     *,
@@ -128,14 +172,14 @@ def download_repository(
             )
         return entry
 
-    if entry.exists() and not _should_refresh(meta, mode):
+    needs_refresh = _should_refresh(meta, mode) or (
+        entry.exists() and _subdir_missing(entry, source)
+    )
+    if entry.exists() and not needs_refresh:
         return entry
 
     if entry.exists() and (entry / ".git").is_dir() and mode != "manual":
-        _run_git(["fetch", "--all", "--tags"], cwd=entry)
-        if source.ref:
-            _run_git(["checkout", source.ref], cwd=entry)
-        commit = _run_git(["rev-parse", "HEAD"], cwd=entry)
+        commit = _refresh_cached_repo(entry, source.ref)
     else:
         if entry.exists():
             shutil.rmtree(entry)

packages/create-python-app-core/tests/test_git_cache.py

@@ -1,4 +1,5 @@
 import json
+import subprocess
 import time
 from pathlib import Path
 
@@ -7,12 +8,29 @@
 from create_python_app_core.git_cache import (
     CacheMeta,
     download_repository,
-    read_cache_meta,
     write_cache_meta,
 )
 from create_python_app_core.paths import ResolvedSource
 
 
+def _git(args: list[str], *, cwd: Path) -> str:
+    return subprocess.check_output(
+        ["git", *args], cwd=cwd, text=True, stderr=subprocess.STDOUT
+    ).strip()
+
+
+def _init_remote(path: Path) -> None:
+    path.mkdir(parents=True)
+    _git(["init", "-b", "main"], cwd=path)
+    _git(["config", "user.email", "test@example.com"], cwd=path)
+    _git(["config", "user.name", "Test"], cwd=path)
+    (path / "README.md").write_text("v1\n", encoding="utf-8")
+    (path / "extensions" / "legacy").mkdir(parents=True)
+    (path / "extensions" / "legacy" / "ok.txt").write_text("legacy\n", encoding="utf-8")
+    _git(["add", "."], cwd=path)
+    _git(["commit", "-m", "init"], cwd=path)
+
+
 def test_file_source_returns_path(tmp_path: Path) -> None:
     src = ResolvedSource(kind="file", url=f"file://{tmp_path}", local_path=tmp_path)
     assert download_repository(src) == tmp_path
@@ -29,6 +47,8 @@ def test_meta_roundtrip(tmp_path: Path) -> None:
     entry = tmp_path / "e"
     meta = CacheMeta(url="u", ref="main", fetched_at=time.time(), commit="abc")
     write_cache_meta(entry, meta)
+    from create_python_app_core.git_cache import read_cache_meta
+
     loaded = read_cache_meta(entry)
     assert loaded is not None
     assert loaded.url == "u"
@@ -41,3 +61,77 @@ def test_skip_git_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
     with pytest.raises(CpaError) as ei:
         download_repository(src, cache_root=tmp_path, refresh="always")
     assert ei.value.code == "CPA_SKIP_GIT"
+
+
+def test_refresh_without_ref_advances_to_remote_tip(tmp_path: Path) -> None:
+    work = tmp_path / "work"
+    bare = tmp_path / "bare.git"
+    cache = tmp_path / "cache"
+
+    _init_remote(work)
+    _git(["clone", "--bare", str(work), str(bare)], cwd=tmp_path)
+
+    src = ResolvedSource(kind="git", url=str(bare))
+    first = download_repository(src, cache_root=cache, refresh="always")
+    first_sha = _git(["rev-parse", "HEAD"], cwd=first)
+    assert (first / "extensions" / "legacy" / "ok.txt").is_file()
+    assert not (first / "extensions" / "all-github-setup").exists()
+
+    _git(["clone", str(bare), str(tmp_path / "push")], cwd=tmp_path)
+    push = tmp_path / "push"
+    _git(["config", "user.email", "test@example.com"], cwd=push)
+    _git(["config", "user.name", "Test"], cwd=push)
+    (push / "extensions" / "all-github-setup").mkdir(parents=True)
+    (push / "extensions" / "all-github-setup" / "ok.txt").write_text(
+        "new\n", encoding="utf-8"
+    )
+    _git(["add", "."], cwd=push)
+    _git(["commit", "-m", "rename extension"], cwd=push)
+    _git(["push", "origin", "main"], cwd=push)
+    new_sha = _git(["rev-parse", "HEAD"], cwd=push)
+    assert new_sha != first_sha
+
+    refreshed = download_repository(src, cache_root=cache, refresh="always")
+    assert refreshed == first
+    assert _git(["rev-parse", "HEAD"], cwd=refreshed) == new_sha
+    assert (refreshed / "extensions" / "all-github-setup" / "ok.txt").is_file()
+
+
+def test_missing_subdir_forces_refresh_when_meta_is_fresh(tmp_path: Path) -> None:
+    work = tmp_path / "work"
+    bare = tmp_path / "bare.git"
+    cache = tmp_path / "cache"
+    _init_remote(work)
+    _git(["clone", "--bare", str(work), str(bare)], cwd=tmp_path)
+
+    src = ResolvedSource(
+        kind="git",
+        url=str(bare),
+        subdir="extensions/all-github-setup",
+    )
+    entry = download_repository(src, cache_root=cache, refresh="always")
+    write_cache_meta(
+        entry,
+        CacheMeta(
+            url=str(bare),
+            ref=None,
+            fetched_at=time.time(),
+            commit=_git(["rev-parse", "HEAD"], cwd=entry),
+        ),
+    )
+    assert not (entry / "extensions" / "all-github-setup").exists()
+
+    _git(["clone", str(bare), str(tmp_path / "push")], cwd=tmp_path)
+    push = tmp_path / "push"
+    _git(["config", "user.email", "test@example.com"], cwd=push)
+    _git(["config", "user.name", "Test"], cwd=push)
+    (push / "extensions" / "all-github-setup").mkdir(parents=True)
+    (push / "extensions" / "all-github-setup" / "ok.txt").write_text(
+        "new\n", encoding="utf-8"
+    )
+    _git(["add", "."], cwd=push)
+    _git(["commit", "-m", "add all-github-setup"], cwd=push)
+    _git(["push", "origin", "main"], cwd=push)
+
+    updated = download_repository(src, cache_root=cache, refresh="stale")
+    assert (updated / "extensions" / "all-github-setup" / "ok.txt").is_file()