feat(cli): wire --refresh/--pin through git cache

Closes #165.

Author: ulises-jeremias

packages/create-awesome-python-app/src/create_awesome_python_app/cli.py

@@ -20,6 +20,7 @@
     print_env_info,
     resolve_source,
 )
+from create_python_app_core.git_cache import RefreshMode
 from rich.console import Console
 
 from create_awesome_python_app import __version__
@@ -57,6 +58,16 @@ def _parse_set_options(set_opt: list[str] | None) -> dict[str, str]:
     return set_map
 
 
+def _normalize_refresh(refresh: str | None) -> RefreshMode | None:
+    if refresh == "always":
+        return "always"
+    if refresh == "stale":
+        return "stale"
+    if refresh == "manual":
+        return "manual"
+    return None
+
+
 def _stringify_option_value(value: Any) -> str:
     if value is None:
         return ""
@@ -88,12 +99,18 @@ def _prompt_custom_options(
     set_map: dict[str, str],
     cache_dir: Path | None,
     offline: bool,
+    refresh: str | None,
     registry_options: list[dict[str, Any]] | None = None,
 ) -> dict[str, str]:
     import questionary
 
     source = resolve_source(template, cache_dir=cache_dir)
-    root = download_repository(source, offline=offline, cache_root=cache_dir)
+    root = download_repository(
+        source,
+        offline=offline,
+        refresh=_normalize_refresh(refresh),
+        cache_root=cache_dir,
+    )
     try:
         config = load_cpa_config(_template_config_path(source.subdir, root))
     except ConfigParseError as err:
@@ -189,14 +206,22 @@ def scaffold(
             la(template)
         raise typer.Exit(0)
 
+    effective_refresh = _normalize_refresh(refresh)
+    if refresh and effective_refresh is None:
+        console.print(
+            "[red]Invalid --refresh mode: "
+            f"'{refresh}'. Use one of: always, stale, manual.[/red]"
+        )
+        raise typer.Exit(2)
+
     # env wiring (#36)
     if no_cache:
         os.environ["CPA_NO_CATALOG_CACHE"] = "1"
-        os.environ["CPA_REFRESH"] = "always"
+        effective_refresh = effective_refresh or "always"
     if cache_dir:
         os.environ["CPA_CACHE_DIR"] = str(cache_dir)
-    if refresh:
-        os.environ["CPA_REFRESH"] = refresh
+    if effective_refresh:
+        os.environ["CPA_REFRESH"] = effective_refresh
     if offline:
         pass  # passed to core
 
@@ -339,6 +364,7 @@ def scaffold(
                 set_map=set_map,
                 cache_dir=cache_dir,
                 offline=offline,
+                refresh=effective_refresh,
                 registry_options=registry_options,
             )
         except ImportError:
@@ -371,6 +397,7 @@ def scaffold(
                 "force": force,
                 "verbose": verbose,
                 "offline": offline,
+                "refresh": effective_refresh,
                 "keep_on_failure": keep_on_failure,
                 "cache_dir": str(cache_dir) if cache_dir else None,
                 "set": set_map,

packages/create-awesome-python-app/tests/test_cli.py

@@ -1,7 +1,20 @@
+from pathlib import Path
+
+import pytest
 from create_awesome_python_app.cli import app
 from typer.testing import CliRunner
 
 runner = CliRunner()
+_CPA_ENV_VARS = ("CPA_REFRESH", "CPA_NO_CATALOG_CACHE", "CPA_CACHE_DIR")
+
+
+@pytest.fixture(autouse=True)
+def _clean_cpa_env(monkeypatch):
+    for name in _CPA_ENV_VARS:
+        monkeypatch.delenv(name, raising=False)
+    yield
+    for name in _CPA_ENV_VARS:
+        monkeypatch.delenv(name, raising=False)
 
 
 def test_version() -> None:
@@ -15,3 +28,143 @@ def test_help() -> None:
     result = runner.invoke(app, ["--help"])
     assert result.exit_code == 0
     assert "Scaffold" in result.stdout or "create" in result.stdout.lower()
+
+
+def test_refresh_flag_is_forwarded(tmp_path: Path, monkeypatch) -> None:
+    tpl = tmp_path / "tpl"
+    tpl.mkdir()
+    captured: dict[str, object] = {}
+
+    async def fake_check_for_latest_version(_package_name):
+        return None
+
+    async def fake_create_python_app(project_directory, options, *_args, **_kwargs):
+        captured["project_directory"] = project_directory
+        captured["options"] = options
+
+    monkeypatch.setattr(
+        "create_awesome_python_app.cli.check_for_latest_version",
+        fake_check_for_latest_version,
+    )
+    monkeypatch.setattr(
+        "create_awesome_python_app.cli.create_python_app",
+        fake_create_python_app,
+    )
+
+    result = runner.invoke(
+        app,
+        [
+            "--template",
+            f"file://{tpl}",
+            "--refresh",
+            "always",
+            "--no-install",
+            "--no-interactive",
+            "api",
+        ],
+    )
+
+    assert result.exit_code == 0, result.stdout + result.stderr
+    options = captured["options"]
+    assert isinstance(options, dict)
+    assert options["refresh"] == "always"
+
+
+def test_invalid_refresh_mode_fails(tmp_path: Path) -> None:
+    tpl = tmp_path / "tpl"
+    tpl.mkdir()
+
+    result = runner.invoke(
+        app,
+        [
+            "--template",
+            f"file://{tpl}",
+            "--refresh",
+            "bogus",
+            "--no-install",
+            "--no-interactive",
+            "api",
+        ],
+    )
+
+    assert result.exit_code == 2
+    assert "Invalid --refresh mode" in result.stdout + result.stderr
+
+
+def test_no_cache_sets_explicit_refresh(tmp_path: Path, monkeypatch) -> None:
+    tpl = tmp_path / "tpl"
+    tpl.mkdir()
+    captured: dict[str, object] = {}
+
+    async def fake_check_for_latest_version(_package_name):
+        return None
+
+    async def fake_create_python_app(project_directory, options, *_args, **_kwargs):
+        captured["project_directory"] = project_directory
+        captured["options"] = options
+
+    monkeypatch.setattr(
+        "create_awesome_python_app.cli.check_for_latest_version",
+        fake_check_for_latest_version,
+    )
+    monkeypatch.setattr(
+        "create_awesome_python_app.cli.create_python_app",
+        fake_create_python_app,
+    )
+
+    result = runner.invoke(
+        app,
+        [
+            "--template",
+            f"file://{tpl}",
+            "--no-cache",
+            "--no-install",
+            "--no-interactive",
+            "api",
+        ],
+    )
+
+    assert result.exit_code == 0, result.stdout + result.stderr
+    options = captured["options"]
+    assert isinstance(options, dict)
+    assert options["refresh"] == "always"
+
+
+def test_pin_appends_ref_to_template_url(tmp_path: Path, monkeypatch) -> None:
+    tpl = tmp_path / "tpl"
+    tpl.mkdir()
+    captured: dict[str, object] = {}
+
+    async def fake_check_for_latest_version(_package_name):
+        return None
+
+    async def fake_create_python_app(project_directory, options, *_args, **_kwargs):
+        captured["project_directory"] = project_directory
+        captured["options"] = options
+
+    monkeypatch.setattr(
+        "create_awesome_python_app.cli.check_for_latest_version",
+        fake_check_for_latest_version,
+    )
+    monkeypatch.setattr(
+        "create_awesome_python_app.cli.create_python_app",
+        fake_create_python_app,
+    )
+
+    result = runner.invoke(
+        app,
+        [
+            "--template",
+            f"file://{tpl}",
+            "--pin",
+            "abc123",
+            "--no-install",
+            "--no-interactive",
+            "api",
+        ],
+    )
+
+    assert result.exit_code == 0, result.stdout + result.stderr
+    options = captured["options"]
+    assert isinstance(options, dict)
+    assert options["template"] == f"file://{tpl}?ref=abc123"

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

@@ -10,6 +10,7 @@
 from typing import Any
 
 from create_python_app_core._version import __version__
+from create_python_app_core.git_cache import RefreshMode
 
 CPA_USER_AGENT = (
     f"create-python-app-core/{__version__} "
@@ -84,6 +85,10 @@ async def create_python_app(
         options = await transform_options(options)
 
     cache = options.get("cache_dir")
+    refresh = options.get("refresh")
+    refresh_mode: RefreshMode | None = (
+        refresh if refresh in ("always", "stale", "manual") else None
+    )
     scaffold_project(
         project_directory,
         template=str(options.get("template") or ""),
@@ -92,6 +97,7 @@ async def create_python_app(
         force=bool(options.get("force", False)),
         install=bool(options.get("install", True)),
         offline=bool(options.get("offline", False)),
+        refresh=refresh_mode,
         keep_on_failure=bool(options.get("keep_on_failure", False)),
         cache_dir=Path(cache) if cache else None,
         options=options,

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

@@ -14,7 +14,7 @@
     load_cpa_config,
 )
 from create_python_app_core.errors import CpaError, ScaffoldAbortedError
-from create_python_app_core.git_cache import download_repository
+from create_python_app_core.git_cache import RefreshMode, download_repository
 from create_python_app_core.loaders import merge_layers
 from create_python_app_core.paths import ResolvedSource, resolve_source
 
@@ -67,6 +67,7 @@ def scaffold_project(
     force: bool = False,
     install: bool = True,
     offline: bool = False,
+    refresh: RefreshMode | None = None,
     keep_on_failure: bool = False,
     cache_dir: Path | None = None,
     options: dict[str, Any] | None = None,
@@ -82,7 +83,12 @@ def scaffold_project(
     try:
         for spec in specs:
             source = resolve_source(spec, cache_dir=cache_dir)
-            root = download_repository(source, offline=offline, cache_root=cache_dir)
+            root = download_repository(
+                source,
+                offline=offline,
+                refresh=refresh,
+                cache_root=cache_dir,
+            )
             layers.append((source, root))
             configs.append(load_cpa_config(_config_path(source, root)))
 

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

@@ -1,5 +1,7 @@
+import asyncio
+
 import pytest
-from create_python_app_core import check_python_version
+from create_python_app_core import check_python_version, create_python_app
 
 
 def test_check_python_version_accepts_current() -> None:
@@ -9,3 +11,30 @@ def test_check_python_version_accepts_current() -> None:
 def test_check_python_version_rejects_impossible() -> None:
     with pytest.raises(SystemExit):
         check_python_version(">=99.0", "create-python-app-core")
+
+
+def test_create_python_app_forwards_refresh(monkeypatch) -> None:
+    captured: dict[str, object] = {}
+
+    def fake_scaffold_project(project_directory: str, **kwargs) -> None:
+        captured["project_directory"] = project_directory
+        captured.update(kwargs)
+
+    monkeypatch.setattr(
+        "create_python_app_core.installer.scaffold_project",
+        fake_scaffold_project,
+    )
+
+    asyncio.run(
+        create_python_app(
+            "api",
+            {
+                "template": "file:///template",
+                "install": False,
+                "refresh": "always",
+            },
+        )
+    )
+
+    assert captured["project_directory"] == "api"
+    assert captured["refresh"] == "always"

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

@@ -2,6 +2,7 @@
 
 import pytest
 from create_python_app_core.installer import scaffold_project
+from create_python_app_core.paths import ResolvedSource
 
 
 def _tpl(tmp: Path, name: str) -> str:
@@ -20,3 +21,34 @@ def test_scaffold_file_template(
     url = _tpl(tmp_path, "tpl")
     scaffold_project(str(dest), template=url, install=False)
     assert (dest / "hello.txt").read_text() == "hi"
+
+
+def test_scaffold_forwards_refresh_to_download_repository(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setenv("CPA_SKIP_GIT", "1")
+    dest = tmp_path / "app"
+    url = _tpl(tmp_path, "tpl")
+    refresh_values: list[str | None] = []
+
+    def fake_download_repository(
+        source: ResolvedSource,
+        *,
+        offline: bool = False,
+        refresh: str | None = None,
+        cache_root: Path | None = None,
+    ) -> Path:
+        refresh_values.append(refresh)
+        assert offline is False
+        assert cache_root is None
+        assert source.local_path is not None
+        return source.local_path
+
+    monkeypatch.setattr(
+        "create_python_app_core.installer.download_repository",
+        fake_download_repository,
+    )
+
+    scaffold_project(str(dest), template=url, install=False, refresh="always")
+
+    assert refresh_values == ["always"]