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

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()